Metadata-Version: 2.5
Name: fastship-py
Version: 0.8.0
Summary: モジュラモノリス + DDD のカーネル。集約・ドメインイベント・UnitOfWork・トランザクショナル outbox/inbox を業務語彙ゼロで提供する
Project-URL: Homepage, https://github.com/theindiehacker/fastship-py
Project-URL: Repository, https://github.com/theindiehacker/fastship-py
Author-email: taiyo tamura <gtaiyou24@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: ddd,hexagonal-architecture,modular-monolith,outbox,unit-of-work
Requires-Python: >=3.14
Requires-Dist: injector<1.0.0,>=0.22.0
Requires-Dist: psycopg[binary]<4.0.0,>=3.3.6
Requires-Dist: pydantic-settings<3.0.0,>=2.14.2
Requires-Dist: pydantic<3.0.0,>=2.13.0
Requires-Dist: pytz>=2026.2
Requires-Dist: sqlalchemy<3.0.0,>=2.1.0
Provides-Extra: gcp
Requires-Dist: cachecontrol<1.0.0,>=0.14.4; extra == 'gcp'
Requires-Dist: google-auth<3.0.0,>=2.62.0; extra == 'gcp'
Requires-Dist: google-cloud-pubsub<3.0.0,>=2.43.0; extra == 'gcp'
Requires-Dist: requests<3.0.0,>=2.34.2; extra == 'gcp'
Provides-Extra: linter
Requires-Dist: import-linter<3.0.0,>=2.14; extra == 'linter'
Description-Content-Type: text/markdown

# 🚀 fastship-py

```bash
# app.fastship.jp テンプレートから作ったリポジトリの backend/ (pyproject.toml のあるところ) で実行する
uv sync
```

DB は PostgreSQL を前提にしており、SQLAlchemy と psycopg も一緒に入る (extras の指定は要らない)。

モジュールの足し方は、テンプレートの `backend/README.md` にある。

## ⚙️ 環境変数

インフラ (Cloud Run など) が注入し、アプリ側では設定しない。使う部品の行だけを設定すればよい。
Secret Manager の値は、実行基盤でシークレットを同名の環境変数として公開する。

| 環境変数 | 使う部品 | 必須か | 内容 |
|:---|:---|:---|:---|
| `DATABASE_WRITER_URL` | SQLAlchemy | 必須 | プライマリの接続先 |
| `DATABASE_READER_URL` | SQLAlchemy | 任意 | リードレプリカ (Read Pool のエンドポイント)。未設定ならすべてプライマリへ送る |
| `NOTIFICATION_PUBLISH_TO` | outbox の中継 | 任意 | 通知を発行するトピック。既定は `notifications` |
| `GOOGLE_CLOUD_PROJECT` | Pub/Sub (`[gcp]`) | エミュレータでは必須 | トピックのあるプロジェクト。未設定なら ADC から引く (Cloud Run ではメタデータサーバー) |
| `PUBSUB_PUSH_AUDIENCE` | Pub/Sub の受信 (`[gcp]`) | push を受けるなら必須 | push サブスクリプションの audience (`--push-auth-token-audience`。省略したときは push エンドポイントの URL) |
| `PUBSUB_PUSH_SERVICE_ACCOUNT` | Pub/Sub の受信 (`[gcp]`) | push を受けるなら必須 | push サブスクリプションのサービスアカウントのメールアドレス (`--push-auth-service-account`) |
| `PUBSUB_EMULATOR_HOST` | Pub/Sub (`[gcp]`) | ローカルだけ | エミュレータの `host:port`。設定すると ADC (実際の GCP の資格情報) を引かず、push の OIDC トークンも検証しない。本番では設定しない |
| `DI_FOR_PY` | DI コンテナ | 任意 | 有効にするプロファイル (カンマ区切り)。テストで `InMem` にするなど |

## 📚 How To
### 🧩 既定のモジュール (Common)

`Common` を `CompositeModule` の先頭に並べると、カーネルの部品が DI コンテナに登録され、outbox / inbox のテーブルも Alembic に載る

```python
from fastship.common.core import Common
from fastship.common.module import CompositeModule


class Backend(CompositeModule):
    def __init__(self) -> None:
        super().__init__([Common(), 〇〇()])
```

| interface | 本番・ローカル | テスト (`DI_FOR_PY=InMem`) |
|:---|:---|:---|
| `UnitOfWork` | `SQLAlchemyUnitOfWork` (`DATABASE_*` の接続先) | `InMemUnitOfWork` |
| `EventContextProvider` | `NullEventContextProvider` | 同左 |
| `EventStore` (outbox) | `PostgreSQLEventStore` | `InMemEventStore` |
| `ConsumedNotificationStore` (inbox) | `PostgreSQLConsumedNotificationStore` | `InMemConsumedNotificationStore` |
| `NotificationPublisher` (中継) | `NotificationPublisherImpl` | 同左 |
| `MessagePublisher` | `CloudPubSubPublisher` | `MessagePublisherStub` (発行した通知を記録する) |
| `MessageSubscriber` | `CloudPubSubSubscriber` | `MessageSubscriberStub` (Notification の JSON を listener に振り分ける) |

- ローカルは本番と同じ実装を使い、環境変数 (`DATABASE_WRITER_URL` / `PUBSUB_EMULATOR_HOST` など) をローカルの DB とエミュレータに向ける
- 実装は最初に使うときに作られる。InMem では DB や Pub/Sub の設定が要らない
- Pub/Sub の実装を使うには extras の `[gcp]` を入れる (「📤 Pub/Sub への発行」)。入れていなくても、`Common` の import と InMem での起動はできる

束縛を差し替えるときは、差し替えるモジュールを `Common` の後ろに並べる (後から登録した束縛が勝つ)

```python
class 〇〇(AppModule):
    @override
    async def startup(self) -> None:
        DIContainer.instance().register(
            DI.of(EventContextProvider, {}, TenantEventContextProvider),  # マルチテナントの場合
        )
```

### 🌐 API

<details><summary><b>🌐 モジュールの API を載せる</b></summary>

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

```python
from fastapi import APIRouter


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

`api.py`:
```python
from fastapi import FastAPI

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

backend = Backend()
app = FastAPI(lifespan=lifespan)  # lifespan で backend.startup() / backend.shutdown() を呼ぶ

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

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

</details>

### 💾 Persistence

<details><summary><b>💾 テーブルクラス ( SQLAlchemy ) の実装</b></summary>

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

`〇〇.port.adapter.persistence.repository.sqlalchemy.user.driver.users_table_row.py`
```python
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`
```python
from .users_table_row import UsersTableRow
# 他のテーブルクラス
```

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

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

</details>

<details><summary><b>🚚 マイグレーション (Alembic)</b></summary>

事前にマイグレーション用のパッケージを生成
```bash
uv add alembic
uv run alembic init migration
```

`migration/env.py`:
```python
# 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())
```

</details>

<details><summary><b>🔀 リードレプリカへの読み書き振り分け (SQLAlchemy)</b></summary>

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

```python
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()`) は、以下の規則でリーダーとライターに振り分ける。

| クエリ | 送り先 |
|:---|:---|
| 読み取り (`select()` / `session.get()` / `text()`) | リーダー |
| 書き込み (flush / `insert()` / `update()` / `delete()`) | ライター |
| `SELECT ... FOR UPDATE` (`with_for_update()`) | ライター |
| 一度でもライターへ送ったセッションの以降のクエリ | ライター (書き込み直後に古いデータを読まない) |
| `session.use_writer()` を呼んだ後のクエリ | ライター (レプリカ遅延を許容できない読み取り) |

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

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

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

</details>


<details><summary><b>📬 outbox / inbox (PostgreSQL)</b></summary>

ドメインイベントは、集約の更新と同じトランザクションで `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 日) より長く取る。短いと、消した後に届いた再配送を二重に処理する。

</details>

### 📨 Messaging

<details><summary><b>📤 Pub/Sub への発行</b></summary>

GCP の SDK は extras に分けている。`gcp` を付けて入れる
```bash
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` を属性にも載せる。サブスクリプションのフィルタで、購読するイベントを絞れる

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

</details>

<details><summary><b>📥 push サブスクリプションの受信</b></summary>

```mermaid
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` があるときだけ。

fastship は FastAPI に依存しないため、受け口のルートはアプリに書く

```python
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`) になり、テンプレートの例外ハンドラが 401 を返す
- 読めないメッセージも再配送され続けるため、push サブスクリプションにはデッドレタートピックを設定する

トークンの検証に加えて、Cloud Run の IAM でも入口を閉じておく (設定を誤ったときの二重の守り)。
- subscriber の Cloud Run は、未認証の呼び出しを拒否する (`--no-allow-unauthenticated`)
- `roles/run.invoker` は、push サブスクリプションのサービスアカウントにだけ付ける

</details>

<details><summary><b>📮 API の応答で通知を発行する</b></summary>

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

```python
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 の処理と発行が切り離され、再試行もキューに任せられる

</details>


### 🛠️ Other

<details><summary><b>📦 DI コンテナ</b></summary>

登録方法:
```python
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)),
)
```

利用方法:
```python
from fastship.di import DI, DIContainer

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

```python
from injector import inject


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

```python
from fastship.common.domain.model import DomainRegistry

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

</details>

<details><summary><b>⚠️ エラー応答 (RFC 9457)</b></summary>

`ErrorJson` は [RFC 9457](https://www.rfc-editor.org/rfc/rfc9457) の Problem Details を表す。`ErrorCode` から組み立てる。
- `type` は `/errors/{コード名}` になる。クライアントは `type` の完全一致で分岐する
- fastship は FastAPI に依存しないため、応答への変換は例外ハンドラに書く (app.fastship.jp テンプレートの `backend/src/api.py` にある)
- `SystemException.detail` はログ用で内部の情報を含みうるため、`detail` には渡さない

```python
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
    )
```

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

</details>

### ✏️ Style Check

```bash
uv run ruff format --check . && uv run ruff check . && uv run mypy && uv run deptry . && uv run lint-imports
```

<details><summary><b>🪄 lint の設定 (ruff / mypy / deptry)</b></summary>

設定は fastship が持たない。テンプレートの `pyproject.toml` にある設定を、プロジェクトで自由に変える。
- fastship-py 経由で入る依存 (`injector` / `pydantic` / `pydantic-settings` / `sqlalchemy`) も、import するならプロジェクトの `dependencies` に直接書く
- `src` 直下のモジュールは `[tool.deptry]` の `known_first_party` に並べる (並べないと自分のモジュールを未宣言の依存とみなす)

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

</details>

<details><summary><b>🏛 アーキテクチャ規約 (import-linter)</b></summary>

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

プロジェクト独自の設定や contract も `[tool.importlinter]` に足す
```toml
[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"]
```

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

<img src="docs/images/hexagonal.svg" alt="core → middleware → port → application → domain → exception の向きにだけ import できる。domain を import できるのは core / application / port.adapter.persistence / port.adapter.service だけ">

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

<img src="docs/images/modular-monolith.svg" alt="モジュール間の import は ACL アダプター (port.adapter.service.〇〇.adapter) から相手の公開 IF (port.adapter.resource) へだけ">

</details>
