Pythonアプリの責務と依存方向を図にするPrompt

このPythonアプリのアーキテクチャを、ファイル名の羅列ではなく「責務と依存方向」で設計してください。単純な1回限りのスクリプトに重い層を強制しないでください。

起動点（CLI/API/job）と実行方法：- 月次バッチ 1本。`python -m billing.run_monthly --month 2026-09` を cron で毎月1日 6:00 に実行。
- 手動の再実行あり（失敗時にオペレーターが同じコマンドを叩く）。
- 他に小さなCLIが1本：`python -m billing.resend --invoice-id INV-000123`（1件だけ再送する）。
主要ユースケースと変えたくない業務ルール：- 月次で全契約の請求書PDFを作り、顧客のメールアドレスへ送る。
- 業務ルール：①消費税は10%、ただし2026-10-01以降の役務提供分は軽減対象品目のみ8%を維持 ②請求額が0円の契約には請求書を発行しない ③前月に未払いがある契約は、当月請求書に「未払い残高」を併記する ④締め日は毎月末日だが、末日が土日祝の場合は前営業日に繰り上げる
- このうち②③④は過去に監査で指摘を受けた経緯があり、勝手に変えられない。
DB、外部API、ファイル等のI/O：- PostgreSQL（契約・請求・入金テーブル）
- SendGrid API（メール送信）
- S3（生成したPDFの保管）
- 祝日判定のため、社内の祝日マスタCSVを月初に手動更新して S3 に置いている
失敗時の整合性・再試行/トランザクション：- 現在、再実行すると同じ月の請求書がもう一度生成され、メールも再送される。オペレーターが手で止めている。
- DBトランザクションは請求書1件単位ではなく、月次バッチ全体を1トランザクションで囲んでいる。途中で1件失敗すると全件ロールバックされる。
設定と秘密情報の置き場所：- `billing/config.py` に直接書かれている。抜粋：
```
SENDGRID_API_KEY = "SG.xxxxxxxxxxxxxxxxxxxxxx"
DB_URL = "postgresql://billing:prod_password@10.0.1.22:5432/billing"
S3_BUCKET = "acme-invoices-prod"
TAX_RATE = 0.10
```
現在の構造とコード断片：```
billing/
  __init__.py
  config.py
  run_monthly.py        # 400行。ループ・PDF生成呼び出し・メール送信・ログを全部ここでやっている
  resend.py             # 120行
  models/
    __init__.py
    contract.py         # SQLAlchemy モデル
    invoice.py          # SQLAlchemy モデル。下記のように税計算がプロパティになっている
    payment.py
  services/
    pdf.py              # ReportLab で PDF 生成
    mailer.py           # SendGrid 呼び出し
  utils/
    dates.py            # 締め日計算。祝日CSVをここで直接 S3 から読んでいる
```

`models/invoice.py` の抜粋：
```
from billing.services.pdf import render_invoice_pdf

class Invoice(Base):
    __tablename__ = "invoices"
    amount_ex_tax = Column(Numeric)
    ...
    @property
    def tax(self):
        from billing.config import TAX_RATE
        return self.amount_ex_tax * TAX_RATE

    @property
    def total(self):
        return self.amount_ex_tax + self.tax

    def to_pdf(self):
        return render_invoice_pdf(self)
```

`services/mailer.py` の抜粋：
```
from billing.models.invoice import Invoice

def send_invoice(invoice: Invoice):
    pdf = invoice.to_pdf()
    ...
```

`utils/dates.py` の抜粋：
```
import boto3

def closing_date(year, month):
    s3 = boto3.client("s3")
    obj = s3.get_object(Bucket="acme-invoices-prod", Key="holidays.csv")
    ...
```
変わりやすい要件、チーム規模：- チームは1人（兼任）。この人が仕様も実装も運用もやっている。
- 変更頻度は月1回程度。過去1年の変更は、税率まわりが2回、メール文面が3回、PDFレイアウトが1回。
- 今後の予定：請求書の英語版対応（時期未定）、入金消込の自動化（検討中）。
可能な単体/統合テスト：- 現在テストは無い。
- ローカルに PostgreSQL は立てられる。SendGrid と S3 は本番アカウントしか無い。

1つの要求が入口→入力検証→業務判断→外部I/O→結果へ進む経路を図示する。変更理由の異なる責務を分け、業務ルールが特定フレームワーク/DBに縛られる箇所を指摘。抽象境界やProtocolは交換・テストの必要がある時だけ提案し、不要なら関数で十分と判断する。循環依存とimport時の外部I/O、秘密情報ハードコードを避ける。部分失敗をどう扱うか未定なら設計判断を保留する。

出力：最小構造案と拡張案の採否、責務/依存図、代表ユースケースのデータフロー、外部I/O境界、失敗/再試行境界、3つのテスト点。実コードが無い時は既存構造を推測で断定しない。

今回の規模なら、**「CLI → ユースケース → 純粋な業務ルール → 外部I/O」**の一方向だけ守れば十分です。Repositoryを全テーブルに作る、DIコンテナを入れる、Entityを何層にも分ける、といった構成は現時点では過剰です。

1. 最小構造案と拡張案の採否
責務	今回	理由
CLI起動点	採用	引数検証と終了コードだけ担当
月次請求ユースケース	採用	400行の処理をここへ集約して流れを見える化
再送ユースケース	採用	「通常送信」と「人間が意図した再送」を分離
業務ルール関数	採用	税、0円除外、未払い、締め日をDB/SendGrid/S3から独立させる
PostgreSQLアクセス	採用	SQLAlchemy依存を境界に閉じ込める
SendGrid境界	Protocol採用	本番アカウントしかなく、Fake差し替えが必要
S3境界	Protocol採用	同じくローカルテストでFakeが必要
PDF Renderer	単純な関数で十分	ReportLabを別製品へ交換する予定がなく、ネットワークI/Oでもない
全RepositoryのProtocol化	不採用	1人チームでは抽象化コストが勝つ
DIコンテナ	不採用	関数引数・constructor injectionで十分
Event Bus	不採用	現在のユースケース数では不要
Outbox/Worker	保留	メールの厳密な再送保証が必要になった場合に検討
CQRS / DDD Aggregate	不採用	問題規模に対して重すぎる

概念上は次の程度で十分です。ファイル名は例であり、重要なのはこの責務分割です。

billing
│
├─ entrypoints
│   ├─ monthly CLI
│   └─ resend CLI
│
├─ application
│   ├─ 月次請求ユースケース
│   └─ 再送ユースケース
│
├─ domain
│   ├─ 税計算
│   ├─ 請求可否
│   ├─ 未払い残高
│   └─ 締め日計算
│
└─ infrastructure
    ├─ PostgreSQL
    ├─ SendGrid
    ├─ S3 Holiday Source
    ├─ S3 PDF Store
    └─ ReportLab Renderer

models / services / utils という分類より、なぜ変更されるかで分ける方が今回には合います。

2. 責務と依存方向

依存方向はこれだけにします。

                    ┌─────────────────┐
                    │ CLI entrypoints │
                    │ monthly/resend  │
                    └────────┬────────┘
                             │
                             ▼
                    ┌─────────────────┐
                    │   Application   │
                    │ use case        │
                    │ orchestration   │
                    └──────┬────┬─────┘
                           │    │
              業務判断     │    │ 外部I/O
                           ▼    ▼
                 ┌────────────┐   ┌─────────────────┐
                 │   Domain   │   │ Infrastructure  │
                 │ pure rules │   │ DB/S3/SendGrid  │
                 └────────────┘   └─────────────────┘

重要なのは、逆方向を禁止することです。

Domain ─X→ SQLAlchemy
Domain ─X→ boto3
Domain ─X→ SendGrid
Domain ─X→ ReportLab
ORM Model ─X→ PDF service
ORM Model ─X→ Mail service

現在の

Python
実行する
class Invoice(Base):
    def to_pdf(self):
        return render_invoice_pdf(self)

は外します。

SQLAlchemyモデルはDB表現なのに、PDF生成という別理由で変更される責務を持っているためです。

代わりに、

Python
実行する
pdf_bytes = render_invoice_pdf(invoice_data)

とApplication側から呼びます。

同様に、

Python
実行する
@property
def tax(self):
    from billing.config import TAX_RATE

もDomainへ移します。

例えば概念的には、

Python
実行する
calculate_tax(
    amount_ex_tax,
    service_date,
    is_reduced_rate_item,
)

です。

税率は環境設定ではありません。監査対象の業務ルールです。

したがって、

Python
実行する
TAX_RATE = 0.10

を環境変数へ移すのではなく、適用日を含むDomain Policyとしてコード＋テストで固定する方が安全です。

3. 業務ルールの置き場所

特に監査指摘済みの②③④は、ユースケースの途中に散在させません。

たとえばDomainを次のような純粋な判断として扱います。

税計算
(amount, service_date, item_type)
        ↓
tax amount

請求書発行判定
(current_charge)
        ↓
issue / skip

未払い表示
(previous_unpaid_amount)
        ↓
invoice display data

締め日
(year, month, holiday_dates)
        ↓
closing_date

この形なら、

Python
実行する
closing_date(2026, 9, holidays)

を実行してもS3にはアクセスしません。

現在の

Python
実行する
def closing_date(year, month):
    s3 = boto3.client("s3")
    ...

は、

「締め日を計算する」
+
「S3から祝日CSVを取得する」

という変更理由の違う仕事を混ぜています。

月次処理開始時に一度だけ

S3
 ↓
HolidaySource
 ↓
holiday_dates
 ↓
closing_date(...)

とします。

契約ごとにS3を読む必要もなくなります。

4. 代表ユースケースのデータフロー

python -m billing.run_monthly --month 2026-09

なら、入口から結果までの経路は次です。

CLI
 │
 │ "--month 2026-09"
 ▼
入力検証
 │
 │ BillingMonth(2026, 9)
 ▼
設定読込
 │
 │ DB URL / SendGrid Key / Bucket
 ▼
HolidaySource(S3)
 │
 │ holidays
 ▼
月次請求UseCase
 │
 ├── PostgreSQLから対象契約を取得
 │
 ├── PostgreSQLから前月未払いを取得
 │
 ├── Domainで締め日決定
 │
 ├── Domainで当月請求額・税額決定
 │
 ├── 0円なら SKIP
 │
 ├── 同月請求書の既存有無確認
 │
 ├── 請求データをDBへ保存
 │
 ├── ReportLabでPDF生成
 │
 ├── S3へPDF保存
 │
 ├── SendGridへ送信
 │
 └── 送信結果をDBへ記録
 │
 ▼
BatchResult
 │
 ├── success
 ├── skipped
 ├── failed
 └── uncertain
 │
 ▼
CLI exit code + log

このときApplicationは「順番」を知りますが、

10%なのか8%なのか
0円なら発行するか
締め日はいつか

という判断ロジックそのものは持ちません。

5. 再実行を安全にする不変条件

現在の最大の問題はここです。

月次請求にはDB上で最低でも、

(contract_id, billing_month)

の一意性を持たせます。

つまり、

契約A + 2026-09

の請求書は論理的に1件だけです。

再実行しても、

INSERT new invoice

ではなく、

既存状態を確認
↓
必要な地点から再開

になります。

推奨する状態イメージは、

CREATED
   ↓
PDF_STORED
   ↓
SENDING
   ↓
SENT

失敗時は、

CREATED → PDF生成失敗
PDF_STORED → メール送信失敗

のように、どこまで成功したかを残します。

これによって同じコマンドを再実行したとき、

SENT
→ 何もしない

PDF_STORED
→ PDFを作り直さず送信から再開

CREATED
→ PDF生成から再開

が可能になります。

PDFのS3キーも、

invoices/2026-09/INV-000123.pdf

のように決定的にしておく方が扱いやすいです。

6. トランザクション境界

現在の

全契約
↓
BEGIN
↓
1000件処理
↓
1件失敗
↓
ROLLBACK 全件

は月次バッチには不向きです。

基本境界は請求書1件単位にします。

Contract A
BEGIN
DB更新
COMMIT

Contract B
BEGIN
DB更新
COMMIT

Contract C
BEGIN
DB更新
COMMIT

ただし、

DB transaction
+
SendGrid
+
S3

を1つのACID transactionにはできません。

したがって、DBトランザクションを開いたままSendGridへ送る設計にはしません。

7. メール送信だけは「完全なexactly-once」を断定しない

ここは重要です。

例えば、

SendGrid送信成功
↓
プロセス強制終了
↓
DBへ SENT を保存できなかった

というケースがあります。

再実行すると、

DB上は未送信
↓
もう一度SendGrid

となり得ます。

つまり、SendGrid側に利用可能な冪等性保証を別途設けない限り、DBだけでメールのexactly-once送信を保証できるとは設計しません。

そこで最小構成では、送信開始前に

SENDING

を記録し、結果が確定しないままプロセスが落ちた場合は、

DELIVERY_UNKNOWN

相当として自動再送しない方が安全です。

FAILED
→ 再試行可能

DELIVERY_UNKNOWN
→ オペレーター確認

SENT
→ 再送しない

とします。

将来、

SendGrid Event Webhook
Outbox
delivery attempt ID
provider message ID

まで導入すれば自動復旧を強化できます。

ただし、「部分失敗した全契約を自動再送するのか」「不明状態をどう扱うのか」という運用要件は提示されていないので、そこから先の自動再試行ポリシーは設計判断を保留します。

8. resend は通常処理と意味を分ける
Bash
python -m billing.resend --invoice-id INV-000123

は月次バッチの失敗再開とは別物です。

これは、

人間が意図的に、すでに作成済みの請求書をもう一度送信する操作

として扱います。

したがって、

invoice-id取得
↓
DBから既存Invoice取得
↓
S3の既存PDF取得
↓
SendGrid送信
↓
delivery_attempt記録

とします。

原則として再送時に、

税再計算
PDF再生成
請求額再計算

はしません。

そうしないと、後から税ロジックやテンプレートが変わった場合、

最初に送ったINV-000123と再送したINV-000123の中身が違う

という監査上かなり嫌な状態になります。

9. 外部I/O境界
境界	Applicationから見える責務	実装
PostgreSQL	契約・未払い・Invoice状態を読む/保存	SQLAlchemy
HolidaySource	祝日一覧を返す	S3 + CSV
PdfStore	PDFを保存/取得する	S3
MailSender	メール送信結果を返す	SendGrid
PdfRenderer	InvoiceDataからbytesを作る	ReportLab

HolidaySource と PdfStore は両方S3でも分けます。

変更理由が違うからです。

祝日CSVの保存先だけ将来Google Driveになることもあり得ますし、PDF保存先だけ別ストレージになる可能性もあります。

一方、PDF Rendererにまで

Python
実行する
class PdfRenderer(Protocol):
    ...

を作る必要は今のところありません。

Python
実行する
render_invoice_pdf(invoice_data) -> bytes

という普通の関数で十分です。

SendGrid/S3については、本番アカウントしかないため、

Python
実行する
class MailSender(Protocol):
    ...

class HolidaySource(Protocol):
    ...

程度の小さな境界には意味があります。

Fakeを差し込めるからです。

10. 設定と秘密情報

現在の

Python
実行する
SENDGRID_API_KEY = "SG...."
DB_URL = "postgresql://..."

は最優先でコード外へ出します。

秘密情報
SENDGRID_API_KEY
DB_URL password
    ↓
環境変数 / Secret Store

環境依存設定
S3_BUCKET
    ↓
環境変数

業務ルール
税率・適用日
    ↓
Domain code

と分類します。

特に、

Python
実行する
TAX_RATE = 0.10

を .env に移すだけではありません。

税率は「環境ごとに違う値」ではなく、日付と品目によって決まる仕様だからです。

またsettingsのimportだけで、

S3へ接続
DBへ接続
秘密情報検証

などを実行しないようにします。

外部I/Oはユースケース開始後に明示的に行います。

11. 現在コードで修正したい依存

示されたコードだけでは実際の循環importが発生しているとは断定できません。

ただし依存方向は既に崩れています。

models.invoice
    ↓
services.pdf

はORMが出力サービスを知っています。

さらに、

services.mailer
    ↓
models.invoice
    ↓
services.pdf

となっており、Mailerが暗黙的にPDF生成まで起動します。

これを、

Application
 ├─→ Domain
 ├─→ PdfRenderer
 ├─→ PdfStore
 └─→ MailSender

にすると、依存が明示されます。

Mailerは、

Python
実行する
send_invoice(
    recipient,
    subject,
    body,
    pdf_bytes,
)

程度でよく、Invoice ORMモデルを知る必要がありません。

12. 3つのテスト点
テスト点	何を保証するか	外部サービス
Domain regression test	税率境界、0円除外、未払い表示、土日祝締め日を固定	不要
Monthly integration test	PostgreSQL＋Fake S3/SendGridで同月2回実行して請求書が増えないこと、SENTを再送しないこと	PostgreSQLのみローカル
Resend test	既存Invoice/PDFを使い、再計算・再生成せず、明示的な再送だけ1回記録すること	Fake SendGrid/S3

特にDomainテストでは、2026年10月1日の前後を必ず境界値として固定します。

また③の「未払い残高を併記」は、提示情報だけでは当月請求総額に加算するのか、表示だけなのかまでは分かりません。そこは推測して実装せず、「current amount」と「unpaid balance」を別項目としてDomainデータに保持し、最終的な合計表示仕様は既存帳票仕様を確認して固定するのが安全です。

このアプリで一番大切なのは、層を増やすことではありません。

監査対象の業務判断
        ↓
pure domain functions

処理順序
        ↓
application use case

SQLAlchemy / boto3 / SendGrid / ReportLab
        ↓
外側

という依存方向に戻すことです。

これだけでも、現在の400行バッチをかなり安全に分解でき、将来の英語版PDFは主にRenderer側、入金消込自動化は新しいUseCaseとして追加できるため、今の段階でそれらを見越した大規模な抽象化を入れる必要はありません。