データモデル層(ORM)
2.9.0 からフレームワークに宣言的データモデル層が内蔵されています。Model を継承し、Field でフィールドを宣言することで、自動的なテーブル作成と CRUD 操作(挿入、削除、更新、検索)が可能になります。モデルは、内蔵ストレージ層の上に直接構築されています。SQLite / MySQL / PostgreSQL は ErisPulse.storage.backend で設定され、バックエンドを切り替えてもモデルコードを変更する必要はありません。
モデルの宣言
from ErisPulse.Core.Bases import Model, Field
class User(Model):
id: int = Field(primary_key=True, autoincrement=True)
name: str = Field(max_length=64)
age: int = Field(default=0, ge=0, le=150)
role: str = Field(default="user", choices=["user", "admin"])
tags: list = Field(default_factory=list) # JSON 列、読み書き時に自動的にシリアライズ
bio: str = Field(default="", description={"i18n": "user.bio", "default": "自己紹介"})
- テーブル名はデフォルトでクラス名を snake_case に変換(
UserProfile → user_profile)し、__tablename__ = "xxx"で上書き可能 Fieldの制約語は宣言型設定クラスと同じ(choices/ge/le/max_length、descriptionは i18n 辞書もサポート);バリデータエンジンは設定検証と同じソース
フィールドパラメータ
| パラメータ | 説明 |
|---|---|
default |
既定値(提供されていない場合、かつ自動増分でない → NOT NULL が必須) |
default_factory |
変更可能な既定値の工場(例: list) |
primary_key |
主キー |
autoincrement |
自動増分主キー(暗黙の主キー、create 後に自動で戻り値を埋め込む) |
max_length |
文字列の最大長さ(生成される VARCHAR(n)、書き込み時の検証) |
nullable |
NULL を許可するかどうか(デフォルトは False) |
index |
普通のインデックスを生成 |
unique |
一意制約 |
choices / ge / le |
列挙 / 数値範囲(書き込み時の検証) |
description |
説明(i18n ディクショナリ、構成クラスと同じ形式) |
column_type |
SQL 列の型定義を上書き |
建表と CRUD
await User.create_table() # 幂等(IF NOT EXISTS)
user = await User.create(name="Alice", age=20) # 挿入(自動増分主キーは自動的に戻り値として返される)
got = await User.get(id=user.id) # 等値検索で最初の1件を取得
users = await User.where(User.age > 18).order_by("-age").limit(10).all()
first = await User.where(User.name == "Alice").first()
total = await User.count(User.age > 18)
user.age = 21
await user.save() # 主キーに基づいて更新(事前制約検証);主キーが存在しない場合は挿入に降格(自動増分主キーもインスタンスに戻り値として返される)
await user.delete() # 主キーに基づいて削除
await User.update_all(User.age > 18, role="adult") # バッチ更新
await User.delete_all(User.age > 100) # バッチ削除
クエリ式は > >= < <= == !=、in_([...])、&(AND)/ |(OR)による組み合わせをサポートします:
await User.where((User.age > 18) & User.name.in_(["Alice", "Bob"])).all()
トランザクション
ORM の読み書きとフレームワークのストレージ層は、同じトランザクションルーティングを使用します。storage.atransaction() 環境のトランザクション内では、create / save / delete とクエリは自動的にトランザクション接続を再利用し、トランザクションに合わせて一括でコミットまたはロールバックされます。個別にコミットされることはありません。
async with storage.atransaction():
await User.create(name="Alice")
... # ブロック内で例外が発生した場合、上記の INSERT もロールバックされます
宣言式構成クラスとの関係
モデルの宣言と ConfigClass(@dataclass + field(metadata=...))は同じ基底を共有しています。制約辞書、検証エンジン(validate_field_constraints)、型カテゴリ登録表(python_type_category)などです。ただし、クラスの基盤は意図的に分離されています。構成フィールドは普通の値(TOMLの往復、一度のロードによるホット更新)であり、モデルフィールドは列記述子(クラスアクセス = クエリ式、行ごとのインスタンス)です。同じ宣言構文を用い、それぞれ異なる役割を持つ2つの基盤です。
どちらの宣言を使うべきか
| 維度 | 構成クラス field(metadata=...) |
モデルフィールド Field() |
|---|---|---|
| 適用場面 | モジュールの動作パラメータ(少数、人間が読みやすい、ホット更新が必要) | 業務データのレコード(複数行、プログラムによる読み書き、クエリが必要) |
| ストレージ形態 | config.toml(コメントを保持、TOMLの往復) |
データベーステーブル(自動作成、SQL方言) |
| 値の形態 | 普通の値(dataclassの属性として直接アクセス) |
記述子(クラスアクセス = 列式、インスタンスアクセス = 行値) |
| 制約の宣言 | metadata={"choices": ..., "min": ..., "max": ...} |
Field(choices=..., ge=..., le=..., max_length=...) |
| 共有基底 | 検証エンジン + 制約辞書 + 型カテゴリ登録表(同一のもの) | 同左 |
経験則として:「モジュールがどのように動作するか」は構成クラスを使い、「ユーザーがどのようなデータを生成したか」はモデルを使う。
辺界と注意事項
- バックエンドはグローバルなストレージ設定によって決定されます。モデルは
__storage__クラス属性を用いて、カスタムのBaseStorageインスタンスに上書きすることができます(テスト用の注入に使用)。 list/dictフィールド(パラメータ化されたジェネリック、例えばlist[int]、dict[str, int])は、コンテナの種類に応じて JSON テキストとして列に格納され、読み書き時に自動的にシリアライズされます。create/saveの前に自動的に制約検証が実行され、失敗した場合はValueError(ローカライズされたメッセージ)が送出されます。- 自動マイグレーションは追加の列の場面に対して提供されています(下記を参照);列の型の変更や列の削除は手動で処理する必要があります。
- ストレージ接続の失敗が発生した際の動作は、ストレージ層と一致します:フレームワークはクラッシュせず、冷却後に自動的に再接続されます。
自動マイグレーション(フェーズ2)
create_table() は、テーブルが既に存在する場合、既存の列とモデルフィールドを自動的に比較します。追加されたフィールドは、ALTER TABLE ADD COLUMN(マイグレーションで NOT NULL 制約を削除し、既存の行に NULL を挿入)を自動的に実行し、index=True を宣言した新しいフィールドは同期的にインデックスが作成されます。手動でマイグレーションスクリプトを書く必要はありません。
# v1リリース後にモデルを進化させ、email / bioフィールドを追加
class User(Model):
__tablename__ = "orm_users"
id: int = Field(primary_key=True, autoincrement=True)
name: str = Field(max_length=64)
age: int = Field(default=0)
email: str = Field(default="") # 新規追加:次回の create_table() で自動的に ADD COLUMN が実行される
bio: str = Field(default="")
await User.create_table() # 再帰的:追加された列のみマイグレーションが実行される
制限事項:追加列のみをサポートしています。主キーの変更、列の型の変更、列の削除は手動で処理する必要があります(破壊的な ALTER による誤操作を避けるため)。
外部キー(リレーションマッピングの基礎)
foreign_key="テーブル.列" は、列レベルの外部キー制約を宣言し、DDL 生成時に REFERENCES 子句を生成します。
class Post(Model):
__tablename__ = "posts"
id: int = Field(primary_key=True, autoincrement=True)
author: int = Field(foreign_key="orm_users.id")
content: str = Field(max_length=255)
関係マッピング(relationship)
モデルクラス内で relationship() をクラス属性として宣言することで、外キー列がどのテーブルに宣言されているかによって自動的に方向が判定されます:
from ErisPulse.Core.Bases import Model, Field, relationship
class User(Model):
__tablename__ = "orm_users"
id: int = Field(primary_key=True, autoincrement=True)
name: str = Field(max_length=64)
posts = relationship("Post", foreign_key="author") # 外キーが相手のテーブルにある → has-many
class Post(Model):
__tablename__ = "posts"
id: int = Field(primary_key=True, autoincrement=True)
author: int = Field(foreign_key="orm_users.id")
content: str = Field(max_length=255)
writer = relationship("User", foreign_key="author") # 外キーがこのテーブルにある → belongs-to
has-many:インスタンス属性はクエリセットを返し、QuerySet の全チェイン機能が利用可能で、create は自動的に本テーブルの主キーを相手の外キー列に埋め込みます:
alice = await User.get(name="Alice")
posts = await alice.posts.all() # このユーザーの全記事
latest = await alice.posts.order_by("-id").first()
total = await alice.posts.count()
hot = await alice.posts.where(Post.content != "").all() # 相手のテーブルのフィールド条件を追加
await alice.posts.delete() # このユーザーのものだけ削除
new_post = await alice.posts.create(content="hi") # author は自動的に alice.id に設定
belongs-to:await を直接実行すると相手のインスタンスが得られます(一致するものがない場合や外キーが NULL の場合は None を返します):
post = await Post.get(id=1)
writer = await post.writer # User インスタンスまたは None
await writer.posts.count() # 双方向にアクセス可能
ポイント:
relatedにはクラス名の文字列(モデルクラス名の登録表を惰性で解析)または直接モデルクラスを渡すことができます。- 関係のクエリと相手のモデルはそれぞれ独自のストレージバックエンドを使用します(バックエンド間の JOIN はサポートされていません。関係のクエリは独立した 2 つの SQL です)。
- 関係のクエリはキャッシュされず、アクセスするたびに即時クエリが実行されます。
User.where(...)を使用して任意のカスタムクエリを行うことも可能です。