TortoiseORMで論理削除を小テーブルに伝播させる

すどたく（エンジニア）

2024.08.27

IT技術

Python

TortoiseORMでレコードの削除を行った際、Model作成時のデフォルトでfields.CASCADEが指定されているためレコード削除時に自動で子テーブルも削除してくれます。

しかし論理削除においては削除はUPDATE処理であるため、子テーブルに情報が伝播しません。

なんとかして情報を伝播させるための方法を考えてみます。

作戦１：describeメソッドから情報が取得できないか試してみる

TortoiseORMのModelクラスには情報取得用のdescribe()が用意されており、各種フィールドや外部キーの参照情報などを辞書型で取得することができます。

1class Parent(Model):
2    id = fields.BigIntField(pk=True)
3    name = fields.CharField(max_length=255, unique=True)
4    created_at = fields.DatetimeField(null=True, auto_now_add=True)
5    updated_at = fields.DatetimeField(null=True, auto_now=True)
6    deleted_at = fields.DatetimeField(null=True, default=None)
7
8class Child(Model):
9    id = fields.BigIntField(pk=True)
10    parent_id = fields.ForeignKeyField("models.Parent", related_name="children")
11    name = fields.CharField(max_length=255, unique=True)
12    created_at = fields.DatetimeField(null=True, auto_now_add=True)
13    updated_at = fields.DatetimeField(null=True, auto_now=True)
14    deleted_at = fields.DatetimeField(null=True, default=None)

このようなモデルの場合、

1Parent.describe(False)["backward_fk_fields"]

このようにして自身を参照しているテーブルの情報を取得できます。"backward_fk_fields"が自身を参照しているテーブル、
しかし、ここで得られる情報は限定的で、今回必要な情報は得られません。

作戦２：フィールドの情報を直接探してみる

1Model._meta.fields_map

Modelインスタンスは_metaという変数を持っており、ここからより詳細な情報を取得できます。_metaという変数名が示す通り外からの参照は想定されていない値のため注意が必要です。

_metaのfields_mapにdescribe()で整形される前のFieldクラスが格納されているので、ここから情報を取得できるか試してみます。今回必要なものは、①参照されている子テーブル②子テーブルのキー名の二点です。

まずは①を探します。fields_mapには関連するフィールド情報が沢山格納されていますが、子テーブルの参照情報を持っているのはBackwardFKRelationクラスであるため、型チェックで抽出します。

1fields = [ _ for _ in Parent._meta.fields_map.values() if isinstance(_, BackwardFKRelation)]

BackwardFKRelationクラスには、field_type: "Type[MODEL]"とrelation_field: strという定義がされています。これらがそれぞれ子テーブルのModel、フィールド名の参照先であるようです。

これを元にTortoiseORMの処理を呼ぶことができそうです。

1fields = [ _ for _ in Parent._meta.fields_map.values() if isinstance(_, BackwardFKRelation)]
2  for field in fields:
3  model = field.field_type field_name = field.relation_name await model.filter(**{ field_name: ParentのID }).update(deleted_at=datetime.now())

これで小要素の更新に使うための値を全て取得し呼ぶことができました。

再起呼び出しで孫テーブルに対応させる

このままだと孫テーブルが存在する場合対応できないので再起処理によって全ての子孫を発見できるようにします。

1T = TypeVar("T", bound=Model)
2
3async def delete_cascade(T, filter_dict):
4  records = T.filter(**filter_dict)
5  await records.update(deleted_at=datetime.now())
6  for record in await records:
7    backward_fk_fields = [ _ for _ in T._meta.fields_map.values() if isinstance(_, BackwardFKRelation)]
8    for field in backward_fk_fields:
9      await delete_cascade(field.related_model, {field.relation_field: record.id})