テーブルクラスは sqlalchemy を直接 import するため、uv add sqlalchemy で直接の依存として宣言する (fastship-py 経由で入っていても、宣言しないと deptry が推移的な依存 (DEP003) として検出する)。

〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver.users_table_row.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

# NOTE: テーブルクラスは `DataBase` を継承して作ること
from fastship.common.port.adapter.persistence.sqlalchemy import DataBase


class UsersTableRow(DataBase):
    __tablename__ = "users"

    id: Mapped[str] = mapped_column(String(36), primary_key=True)
    name: Mapped[str] = mapped_column(String(50), index=True)  # index 名は ix_users_name

〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver.__init__.py

from .users_table_row import UsersTableRow
# 他のテーブルクラス

テーブルクラスを足したら、そのモジュールの tables にドライバーモジュールのパスを足す

class 〇〇(AppModule):
    @property
    @override
    def tables(self) -> list[str]:
        return ["〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver"]

事前にマイグレーション用のパッケージを生成

uv add alembic
uv run alembic init migration

migration/env.py:

# alembic init の生成物から、target_metadata = None をこの 3 行に差し替える

from fastship.common.port.adapter.persistence.sqlalchemy import DataBase
from core import Backend  # 全モジュールを束ねた CompositeModule

target_metadata = DataBase.metadata_of(Backend())

接続先はインフラが環境変数で注入する (Secret Manager の値は実行基盤で環境変数として公開する)。アプリ側では設定しない。

from fastship.common.port.adapter.persistence.sqlalchemy import SQLAlchemyUnitOfWork

unit_of_work = SQLAlchemyUnitOfWork.from_settings(pool_pre_ping=True)

トランザクション (unit_of_work.session()) は、読み取りも含めてすべてプライマリで実行する。 レプリカの遅延した値を元に更新すると、プライマリの新しい値を上書きしてしまう (lost update) ため。 更新系の生 SQL (text()) もここで実行する。

トランザクション外の読み取り (unit_of_work.query()) は、以下の規則でリーダーとライターに振り分ける。

with unit_of_work.query() as session:
    session.use_writer()
    session.get(...)

認可の判定 (無効化された API キー、停止されたユーザー、取り消された権限など) は、レプリカ遅延の間に 古い状態で許可しないよう、session.use_writer() を呼んでプライマリから読むこと。

Read Pool のノード数を増減しても、エンドポイントは変わらないためアプリ側の変更は要らない。 SessionPreparer (RLS の SET LOCAL など) は、振り分け先のコネクションそれぞれに適用される。

登録方法:

from sqlalchemy import create_engine

from fastship.common.application import UnitOfWork
from fastship.common.port.adapter.persistence.inmem import InMemUnitOfWork
from fastship.common.port.adapter.persistence.sqlalchemy import SQLAlchemyUnitOfWork
from fastship.di import DI, DIContainer

engine = create_engine("postgresql+psycopg://...")

DIContainer.instance().register(
    DI.of(UnitOfWork, {"InMem": InMemUnitOfWork()}, SQLAlchemyUnitOfWork(engine)),
)

利用方法:

from fastship.di import DI, DIContainer

# `DI_FOR_PY=InMem` なら InMemUnitOfWork のインタンスを取得
# `DI_FOR_PY=` なら SQLAlchemyUnitOfWork のインタンスを取得
unit_of_work = DIContainer.instance().resolve(UnitOfWork)

from injector import inject


class 〇〇ApplicationService:
    @inject
    def __init__(self, 〇〇_repository: 〇〇Repository) -> None: ...

from fastship.common.domain.model import DomainRegistry

〇〇_repository = DomainRegistry.resolve(〇〇Repository)

ErrorJson は RFC 9457 の Problem Details を表す。ErrorCode から組み立てる。

• type は /errors/{コード名} になる。クライアントは type の完全一致で分岐する
• fastship は FastAPI に依存しないため、応答への変換は例外ハンドラに書く (app.fastship.jp テンプレートの backend/src/api.py にある)
• SystemException.detail はログ用で内部の情報を含みうるため、detail には渡さない

from fastapi import Request, Response
from fastapi.responses import JSONResponse
from fastship.common.exception import SystemException
from fastship.common.port.adapter.resource import ErrorJson


@app.exception_handler(SystemException)
async def system_exception_handler(request: Request, exception: SystemException) -> Response:
    exception.logging()
    problem = ErrorJson.of(exception.error_code, str(request.url))
    return JSONResponse(
        problem.model_dump(mode="json", exclude_none=True), problem.status, media_type=ErrorJson.MEDIA_TYPE
    )

{"type": "/errors/USER_NOT_FOUND", "title": "ユーザーが見つかりません", "status": 404, "instance": "https://api.example.com/users/1"}

設定は fastship が持たない。テンプレートの pyproject.toml にある設定を、プロジェクトで自由に変える。

• fastship-py 経由で入る依存 (injector / pydantic / pydantic-settings / sqlalchemy) も、import するならプロジェクトの dependencies に直接書く
• src 直下のモジュールは [tool.deptry] の known_first_party に並べる (並べないと自分のモジュールを未宣言の依存とみなす)

[tool.deptry.per_rule_ignores]
DEP002 = ["uvicorn"]  # import せずに使う依存

テンプレートの pyproject.toml の [tool.importlinter] が規約を有効にしている。

• 規約の中身 (層の順序や許可する import) は fastship のコード (fastship.linter.architecture) にあり、fastship-py を更新すると変わる
• root_packages には src 直下のモジュールをすべて並べる。足し忘れは lint-imports が検出する
• import-linter は root_packages が空だと動かない。テンプレートが見本の example モジュールを持つのはそのため

プロジェクト独自の設定や contract も [tool.importlinter] に足す

[tool.importlinter]
root_packages = ["billing"]
include_external_packages = true  # sqlalchemy などの外部パッケージを forbidden_modules に書くときに要る

[[tool.importlinter.contracts]]
id = "no-sqlalchemy-in-domain"
name = "ドメイン層は永続化の技術を知らない"
type = "forbidden"
source_modules = ["billing.domain"]
forbidden_modules = ["sqlalchemy"]

ヘキサゴナルアーキテクチャ (モジュール内部の依存)

モジュラモノリス (モジュール間の依存)