Alembicでマイグレーション（FastAPI+SQLModel）

前回、FastAPIとSQLmoelを使用して、webアプリケーションを作成しました。
https://zenn.dev/keita_f/articles/4493e3cfd76aec
このアプリケーションに、別のモデルを追加して、さらに、二つのデータベースに関連性を持たせます。

BaseUserModelの設定変更

前回作成したBaseUserModelほかを少し変更します。

backend/v1/model/user_model.py

 class BaseUserModel(BaseUser, table = True):
   id: Optional[int] = Field(default = None, primary_key = True)
+  uuid: str = Field(default_factory = uuid_pkg.uuid4, nullable = False, unique = True)
-  uuid: str = Field(default_factory = uuid_pkg.uuid4, nullable = False)
   hashed_password: str = Field(nullable = False)
   member_id: str = Field(nullable = True)
   created_at: datetime = Field(default = datetime.now(), nullable = False)
   updated_at: datetime = Field(default_factory = datetime.now, nullable = False, sa_column_kwargs = {'onupdate': datetime.now})
   refresh_token: str = Field(nullable = True)

〜中略〜

+ class BaseUserRead(BaseUser):
+   id: int
+   uuid: str
+   created_at: datetime
+   updated_at: datetime
+   contents: List[BaseContentReadWithUser] = []

これから作成するモデルと関係性を持たせるため、foreign_keyに設定するキーには外部キー制約（foreign key constraints）の設定が必要です。

外部キー制約とは？
親テーブルと子テーブルの2つのテーブル間でデータの整合性を保つために設定される制約のこと

外部キー制約として、foreign_keyに使用される値が一意に決まることが必要です。
よって、primary_key=Trueまたはunique=Trueを付与することが必要になります。

ここでは、foreign_keyにuuidを使用することにします。
既にidにprimary_keyを設定しているので、unique=Trueを設定します。

BaseUserReadは、この後設定するBaseContentReadWithUserを設定して、ユーザー検索結果が返されるときに、ユーザーに紐づいているContentの情報を一緒に返すように設定します。

変更したBaseUserModelをマイグレートする

それでは変更したBaseUserModelをマイグレートします。

bash

(poetry run) alembic revision --autogenerate -m 'Change BaseUseModel'

poetryを使用している場合は、（）内のフレーズを記載して実行してください

作成されたマイグレーションファイルを実行し、マイグレートします。

bash

(poetry run) alembic upgrade head

これで、BaseUserModelの更新は終了です。

Contentモデルの定義

次に、BaseUserModelと関係性を持たせるContentモデルを作成していきます。
基本的にはBaseUserModelをベースに作成しています。

コードはこちら

backend/v1/model/content_model.py

from sqlmodel import SQLModel, Field, Relationship
from typing import TYPE_CHECKING, Optional
from datetime import datetime

import uuid as uuid_pkg

if TYPE_CHECKING:
  from .user_model import BaseUserModel, BaseUserOut, BaseUserOutWithContent

class BaseContent(SQLModel):
  title: str
  content: str

class BaseContentModel(BaseContent, table=True):
  id: Optional[int] = Field(default = None, primary_key = True)
  uuid: str = Field(default_factory = uuid_pkg.uuid4, nullable = False, unique = True)
  created_at: datetime = Field(default = datetime.now(), nullable = False)
  updated_at: datetime = Field(default_factory = datetime.now, nullable = False, sa_column_kwargs = {'onupdate': datetime.now})
  
  user_uuid: Optional[str] = Field(default = None, foreign_key="baseusermodel.uuid")
  user: Optional["BaseUserModel"] = Relationship(back_populates="contents")
  
class BaseContentOut(SQLModel):
  id: int
  uuid: str
  title: str
  content: str
  created_at: datetime
  updated_at: datetime

class BaseContentCreate(BaseContent):
  pass

class BaseContentRead(BaseContent):
  id: int
  uuid: str
  created_at: datetime
  updated_at: datetime
  user_uuid: Optional[str] = Field(default=None, foreign_key="baseusermodel.uuid")

class BaseContentReadWithUser(BaseContent):
  """User検索したときに、コンテンツを返す用のschema"""
  id: int
  uuid: str
  created_at: datetime
  updated_at: datetime

class BaseContentUpdate(SQLModel):
  title: Optional[str] = None
  content: Optional[str] = None

いくつかBaseUserModelとの違いもあるため、メモしておきます。

foreign_keyの設定

BaseContentModelにはforeign_keyを設定しています（以下の部分です）。

user_uuid: Optional[str] = Field(default = None, foreign_key="baseusermodel.uuid")

書き方は、
「キー名: 型 = Field(設定内容, foreign_key="外部テーブル名.外部テーブルのキー名")」
です。

外部テーブル名の部分は、大文字で書かないように注意してください。
データベースで設定しているテーブル名になります。

「設定内容」の部分は、必要に応じて記載してください。

Relationshipの設定

次は、Relationshipの設定をしていきます（以下の部分です）。

user: Optional["BaseUserModel"] = Relationship(back_populates="contents")

データ結合時のテーブル同士の関連性を明確にするためにRelationshipクラスを使用します。

Relationshipに渡す引数

• back_populatesは関連するテーブルにそれぞれ記載が必要
  back_populatesが無いとcommit()し忘れたときにエラーがでます
• 型ヒントの書き方に注意
  OK：Optional["BaseUserModel"]
  NG：Optional[BaseUserModel]
  　※Pythonのinterpreterがクラスを認識できないため文字列で記載する必要がある

https://sqlmodel.tiangolo.com/tutorial/relationship-attributes/define-relationships-attributes/

Relationshipの引数であるback_populatesに設定するキーは、接続するデータベースでも設定が必要になります。

backend/v1/model/user_model.py

 class BaseUserModel(BaseUser, table = True):
   id: Optional[int] = Field(default = None, primary_key = True)
   uuid: str = Field(default_factory = uuid_pkg.uuid4, nullable = False, 
unique = True)
   hashed_password: str = Field(nullable = False)
   member_id: str = Field(nullable = True)
   created_at: datetime = Field(default = datetime.now(), nullable = False)
   updated_at: datetime = Field(default_factory = datetime.now, nullable = False, sa_column_kwargs = {'onupdate': datetime.now})
   refresh_token: str = Field(nullable = True)
  
  # 1対多の関係を設定
+  contents: List["BaseContentModel"] = Relationship(back_populates = 'user')

これでモデルの設定は完了です。

以下のように、必要に応じて他のモデルにも外部テーブルからデータを追加する設定を書いていくことができます。

class BaseContentRead(BaseContent):
  id: int
  uuid: str
  created_at: datetime
  updated_at: datetime
  user: Optional["BaseUserOut"] = None

TYPE_CHECKINGについて

上記で修正したBaseUserModelや追加で設定したBaseContentModelには、「TYPE_CHECKING」という書き方を追記しています（例えば以下のように書いています）。

backend/v1/model/content_model.py

if TYPE_CHECKING:
  from .content_model import BaseContentModel

これは、今回のようにファイルをUserとContentで分けて書く場合に、それぞれのファイルで互いのモデルをインポートしなければならないときに、循環インポートが生じてしまうことを防ぐための設定です。
モデルをインポートして使用するのは、型チェックの時のみです。
そのため、「TYPE_CHECKING」とあるように、型チェック時のみ必要なコードとして認識させるため、このような書き方をしています。
同じファイル内にモデルを定義するのであれば、問題は生じませんが、通常、モデルが多くなると、ファイル分割は必要になりますので、このような手法が取られているようです。

これを書かないと、循環インポートが生じてしまい、エラーになってしまいます。
注意してください。

https://sqlmodel.tiangolo.com/tutorial/code-structure/#import-only-while-editing-with-type_checking

複数モデルのマイグレーション（SQLModel）

それでは設定したモデルをベースにマイグレーションしていきます。

env.pyファイルに設定を追記

まずはコードの全体像を示します。

コードはこちら

backend/migrations/env.py

  import os
  from logging.config import fileConfig

  from sqlalchemy import engine_from_config
+ from sqlalchemy import MetaData, pool

  from alembic import context
  from sqlmodel import SQLModel

  from v1.model.user_model import BaseUserModel
+ from v1.model.content_model import BaseContentModel
  from dotenv import load_dotenv
  load_dotenv()

  # this is the Alembic Config object, which provides
  # access to the values within the .ini file in use.
  config = context.config

  # Interpret the config file for Python logging.
  # This line sets up loggers basically.
  if config.config_file_name is not None:
      fileConfig(config.config_file_name)

  # add your model's MetaData object here
  # for 'autogenerate' support
  # from myapp import mymodel
  # target_metadata = mymodel.Base.metadata

  # add
+ def combine_medtadata(*args):
+     m = MetaData()
+     for metadata in args:
+         for t in metadata.tables.values():
+             t.tometadata(m)
+     return m
+ target_metadata = combine_medtadata(BaseUserModel.metadata, BaseContentModel.metadata)

  # other values from the config, defined by the needs of env.py,
  # can be acquired:
  # my_important_option = config.get_main_option("my_important_option")
  # ... etc.


  def run_migrations_offline() -> None:
      """Run migrations in 'offline' mode.

      This configures the context with just a URL
      and not an Engine, though an Engine is acceptable
      here as well.  By skipping the Engine creation
      we don't even need a DBAPI to be available.

      Calls to context.execute() here emit the given string to the
      script output.

      """
      url = config.get_main_option("sqlalchemy.url")
      context.configure(
          url=url,
          target_metadata=target_metadata,
          literal_binds=True,
          dialect_opts={"paramstyle": "named"},
      )

      with context.begin_transaction():
          context.run_migrations()


  def run_migrations_online() -> None:
      """Run migrations in 'online' mode.

      In this scenario we need to create an Engine
      and associate a connection with the context.

      """
      config.set_section_option('alembic', 'POSTGRES_USER', os.environ['POSTGRES_USER']) # add
      config.set_section_option('alembic', 'POSTGRES_PASSWORD', os.environ['POSTGRES_PASSWORD']) # add
      config.set_section_option('alembic', 'POSTGRES_DOCKER', os.environ['POSTGRES_DOCKER']) # add
      config.set_section_option('alembic', 'POSTGRES_PORT', os.environ['POSTGRES_PORT']) # add
      config.set_section_option('alembic', 'POSTGRES_DB', os.environ['POSTGRES_DB']) # add
    
      connectable = engine_from_config(
          config.get_section(config.config_ini_section, {}),
          prefix="sqlalchemy.",
          poolclass=pool.NullPool,
      )

      with connectable.connect() as connection:
          context.configure(
              connection=connection, target_metadata=target_metadata
          )

          with context.begin_transaction():
              context.run_migrations()


  if context.is_offline_mode():
      run_migrations_offline()
  else:
      run_migrations_online()

参考にした記事はこちらです。
https://qiita.com/haruki-lo-shelon/items/c469132f6b81b2c650a7

通常のAlembicの使用方法はこちらを参考にしました。
https://zenn.dev/shimakaze_soft/articles/4c0784d9a87751

マイグレートをしていく

それでは先ほどと同様にマイグレートします。

bash

(poetry run) alembic revision --autogenerate -m 'Create new table'

poetryを使用している場合は、（）内のフレーズを記載して実行してください

作成されたマイグレーションファイルを実行し、マイグレートします。

bash

(poetry run) alembic upgrade head

以上で、新しいモデル、テーブルの設定ができました。

つまずきポイント

私のつまづきポイントを共有しておきます（誰かの参考になれば）。

• 新しいモデルを先に作成してから一度にマイグレーションを行おうとすると、既存モデルがあるために、そのモデルをベースに外部キー制約を設定する必要があるとエラーが出る
• alembicでマイグレーションファイルを作成しても、マイグレートできなかった（マイグレーション用のファイルは作成された）