モジュールの routers に APIRouter を並べる。CompositeModule に並べた全モジュールの routers が 1 つにまとまる。 routers を書かないモジュール (購読だけのものなど) は何も載せない。

from fastapi import APIRouter


class 〇〇(AppModule):
    @property
    @override
    def routers(self) -> list[APIRouter]:
        return [UserResource().router]

api.py:

from fastapi import FastAPI
from fastship.common.port.adapter.resource import add_exception_handlers

from core import Backend  # 全モジュールを束ねた CompositeModule

backend = Backend()
app = FastAPI(lifespan=lifespan)  # lifespan で backend.startup() / backend.shutdown() を呼ぶ
add_exception_handlers(app)  # エラー応答を RFC 9457 にそろえる (「⚠️ エラー応答」)

for router in backend.routers:
    app.include_router(router)

• lifespan の中ではなく、app を作った直後に載せる。lifespan の外で呼ぶ app.openapi() にも載り、テストごとに lifespan を回してもルートが重複しない
• routers は起動 (DI 登録) の前に呼ばれる。Resource のコンストラクタでは DI を解決せず、リクエストを処理するときに解決する

テーブルクラスは 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())

接続先は DATABASE_WRITER_URL / DATABASE_READER_URL で注入する (冒頭の「⚙️ 環境変数」を参照)。

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 など) は、振り分け先のコネクションそれぞれに適用される。

ドメインイベントは、集約の更新と同じトランザクションで stored_events に保存する (outbox)。中継が未発行のイベントを MQ に発行する。 受信側は、listener の処理と同じトランザクションで consumed_notifications に記録し、同じ通知が 2 回届いても 1 回だけ処理する (inbox)。

テーブル (stored_events / consumed_notifications) と、ストア・中継の DI 登録は Common が行う (「🧩 既定のモジュール」)。 ストアは UnitOfWork に登録した SQLAlchemyUnitOfWork のセッションに書く

中継 (NotificationPublisherImpl)

• 未発行 (published_at が空) の行を FOR UPDATE SKIP LOCKED で 100 件まで取り出し、NOTIFICATION_PUBLISH_TO に発行して published_at を付ける
• 発行済みを行ごとに記録するため、id の小さいイベントが後から commit されても取りこぼさない
• 中継が同時に走っても、ほかの中継がロックしている行は飛ばして分担する
• 発行に途中で失敗したら、発行できた行だけを発行済みにする。残りは次の中継で発行する
• 発行に失敗した行は後回しにする (失敗した回数の少ない順に取り出す)。発行できないイベントがあっても、ほかのイベントは止まらない
• 同じイベントを 2 回発行するのは、発行した後 commit する前に中継が落ちたときだけ。受信側の inbox が重複を捨てる

処理済みの記録は、Cloud Scheduler などから ConsumedNotificationApplicationService.prune_consumed_notifications() を定期的に呼んで消す (既定は 14 日を過ぎたもの)。 保持期間は、MQ が再配送しうる期間 (Pub/Sub は最大 7 日) より長く取る。短いと、消した後に届いた再配送を二重に処理する。

GCP の SDK は extras に分けている。gcp を付けて入れる

uv add 'fastship-py[gcp]'

接続先は GOOGLE_CLOUD_PROJECT / PUBSUB_EMULATOR_HOST で注入する (冒頭の「⚙️ 環境変数」を参照)。 MessagePublisher / MessageSubscriber の DI 登録は Common が行う (「🧩 既定のモジュール」)。Pub/Sub 以外の MQ を使うときは、後ろのモジュールで差し替える。

• 発行は Pub/Sub が受け取るまで同期で待つ (最大 300 秒)。async の処理から呼ぶときは、asyncio.to_thread などでスレッドに逃がす
• 発行に失敗すると例外になる (黙って捨てない)
• 本文 (Notification の JSON) に加えて、event_type と partition_key を属性にも載せる。サブスクリプションのフィルタで、購読するイベントを絞れる

gcloud pubsub subscriptions create billing --topic=notifications \
  --message-filter='attributes.partition_key = "tenant-1"'

flowchart LR
    publisher["CloudPubSubPublisher"] --> topic[["Pub/Sub トピック"]]
    topic --> push["push サブスクリプション<br>(サービスアカウントの OIDC トークンを付ける)"]
    push -->|"Cloud Run の IAM が<br>run.invoker を確かめる"| subscriber["subscriber の Cloud Run<br>(CloudPubSubSubscriber が OIDC トークンを検証する)"]
    subscriber --> listener["ExchangeListener"]

CloudPubSubSubscriber は、push サブスクリプションが付けた OIDC トークン (JWT) の署名・有効期限・発行者・audience を検証し、 設定したサービスアカウントからのリクエストだけを受け付ける。 設定 (PUBSUB_PUSH_AUDIENCE / PUBSUB_PUSH_SERVICE_ACCOUNT。冒頭の「⚙️ 環境変数」を参照) が無いと、 CloudPubSubSubscriber.from_settings() の時点で例外になる。検証しないのは PUBSUB_EMULATOR_HOST があるときだけ。

受け口のルートはアプリに書く

from fastapi import FastAPI, Request
from fastship.common.port.adapter.messaging.pubsub import CloudPubSubSubscriber

from core import Backend  # 全モジュールを束ねた CompositeModule

app = FastAPI()
subscriber = CloudPubSubSubscriber.from_settings()
subscriber.add(Backend().listeners)


@app.post("/pubsub/push", status_code=204)
async def push(request: Request) -> None:
    # 2xx 以外を返すと Pub/Sub が再配送する。同じメッセージが 2 回届いても、inbox が重複を捨てる
    await subscriber.receive(await request.json(), request.headers.get("Authorization"))

• 送信元を認証できないと SystemException (COMMON_2002) になり、add_exception_handlers の例外ハンドラが 401 を返す
• 読めないメッセージも再配送され続けるため、push サブスクリプションにはデッドレタートピックを設定する

トークンの検証に加えて、Cloud Run の IAM でも入口を閉じておく (設定を誤ったときの二重の守り)。

• subscriber の Cloud Run は、未認証の呼び出しを拒否する (--no-allow-unauthenticated)
• roles/run.invoker は、push サブスクリプションのサービスアカウントにだけ付ける

PublishNotificationsMiddleware は、API が 2xx を返すリクエストで NotificationApplicationService.publish_notifications() を呼び、 outbox に溜まった未発行の通知を MQ に発行する。発行と、発行済みの記録は 1 つのトランザクションで行う (中継の動きは「📬 outbox / inbox」)。 NotificationPublisher の実装 (NotificationPublisherImpl) は Common が DI コンテナに登録する。API と、push を受ける worker の両方に入れる。

from fastship.common.middleware import PublishNotificationsMiddleware

app.add_middleware(PublishNotificationsMiddleware, excluded_paths={"/health/check"})

• 応答の最後の本文を送る直前に発行し、応答は発行を待つ。Cloud Run の cpu_idle (CPU をリクエストの処理中だけ割り当てる) のままで動く
• 応答を返した後に発行を残さない (FastAPI の BackgroundTasks を使わない)。リクエストの外では CPU が絞られ、発行が止まるため
• 発行はミドルウェアが持つタスク 1 本で行い、同期の発行はスレッドに逃がす。イベントループを塞がず、他のリクエストの処理は止まらない
• 同じプロセスでは発行を並行させない。発行中に来たリクエストは、今の発行の後の 1 回にまとめて待つ
• 発行に失敗しても応答は 2xx のまま返り、失敗はログに残る。未発行の通知は outbox に残り、次の発行で送られる
• 発行は、リクエストの文脈 (ContextVar に置いたテナントなど) を引き継がない
• 複数のインスタンスからは並行に発行しうる。未発行の通知は中継どうしで分担し、同じ通知を二重に発行しない

応答の遅れが気になるとき

• Pub/Sub への発行は受け取られるまで最大 300 秒待つため、Pub/Sub の障害中は応答もその分遅れる
• 発行が重くなったら、ミドルウェアを外し、Cloud Tasks のキュー (または Cloud Scheduler) から NotificationApplicationService.publish_notifications() を呼ぶエンドポイントを叩く。API の処理と発行が切り離され、再試行もキューに任せられる

登録方法:

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 の完全一致で分岐する
• SystemException.detail はログ用で内部の情報を含みうるため、detail には渡さない

add_exception_handlers(app) は、例外を ErrorJson (application/problem+json) の応答に変える例外ハンドラを登録する。

• ValueError / AssertionError のログには、例外の型と送出した場所だけを出す (メッセージには入力値が入りうる)
• 応答を変えたい例外は、add_exception_handlers(app) の後に app.add_exception_handler で登録する (同じ例外なら後から登録したものが勝つ)。応答は ErrorJsonResponse.of(エラーコード, request) で返す

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

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

• fastship-py 経由で入る依存 (fastapi / 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"]

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

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