コーディングルール

ディレクトリ構成は ディレクトリ構成 を参照してください。 ここでは実装規約（命名・型・層間の呼び出し方・データアクセス・エラーハンドリング 等）を定義します。

---

共通規約（全アプリ共通）

言語・ツール

• Python 3.12
• フォーマッタ / リンタ : ruff（line-length 88、double quote）
• 型チェック : mypy（strict = True）
• 依存管理 : uv

命名・スタイル

• ファイル名・モジュール名・変数名・関数名は snake_case
• 定数は UPPER_SNAKE_CASE（例: DEFAULT_PAGE、DEFAULT_SIZE）
• クラス名は PascalCase（例: ArticleResponse、CreateArticleRequest）
• Pydantic / pydantic-settings の モデル定義以外で class を使わない
• ビジネスロジックは関数で表現し、状態はモジュールスコープに置く
• すべての関数・メソッドに 型アノテーション を付ける（mypy strict 準拠）
• DB に保存する日時フィールドは UNIX タイムスタンプ（ミリ秒） で int(time.time() * 1000) を使う
• 論理削除フィールドは deleted_at（None = 未削除）

import の書き方

• 同名衝突を避けるため、レイヤをまたぐ import は エイリアス付き で行う
• services → repositories : from ...repositories import article as article_repository
• services → libs : from ...libs import sqs as sqs_lib
• routes → services : from ...services import article as article_service
• import 順序は ruff の I（isort）に従う（標準ライブラリ / サードパーティ / 自プロジェクト）

設定（環境変数）

• 環境変数は src/config/env.py の Settings(BaseSettings) に集約する
• 各レイヤからは settings 経由 でのみアクセスし、os.environ を直接読まない

# src/config/env.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    environment: str = "local"
    aws_region: str = "ap-northeast-1"
    # ...

    class Config:
        env_file = ".env"


settings = Settings()

---

@apps/odds_poc_app（FastAPI）

API 設計規約

• URI・JSON ノード名は snake_case
• リクエスト / レスポンスボディは JSON
• メタ情報（ページング・件数など）は基本レスポンスボディに含めるが、必要に応じて HTTP ヘッダー を使う
• API バージョンはパスプレフィックスで管理 : /api/v1/...（load_router() で prefix="/api/v1" を付与）
• 認証は API Gateway の API Key

レイヤ構成と呼び出し規則

routes/  ─►  services/  ─►  repositories/
                  │              │
                  └──►  libs/  ◄─┘

• routes は services のみ を呼び出す
• services は repositories と libs を呼び出す
• repositories は libs と packages/models/documentDB を参照してよい
• 同じレイヤ内で互いを呼び出さない（例 : services の関数から別の services の関数を呼ばない）
• 下位レイヤから上位レイヤ（repositories → services、services → routes 等）への呼び出しは禁止

routes layer（src/routes/v1/）

• ファイルはリソース単位で 1 つ作成する（例 : article.py、media.py）
• 各ファイルでモジュールスコープに router = APIRouter(tags=[...]) を定義する
• ルーティングは FastAPI のデコレータを使用する
• ハンドラ関数は 必ず async def で、シグネチャに引数・戻り値の型アノテーションを付ける
• ハンドラ関数の中身は service 関数の呼び出しのみ とし、ビジネスロジックを書かない
• リクエスト / レスポンスのバリデーションは src/schemas/ の Pydantic モデルで行う
• 関数名は HTTP メソッド + 対象に対応する以下の固定名を用いる

用途	関数名
作成（POST）	create()
一覧取得（GET）	get()
1 件取得（GET, id 指定）	getById()
1 件取得（GET, alias 指定）	getByAlias()
1 件取得（GET, id か alias）	getByIdOrAlias()
更新（PUT）	update()
削除（DELETE）	delete()

camelCase の関数名は ruff の N802 を除外することで許可している（ruff.toml）。新規ルールを追加する場合もこの命名に揃えること。

実装例 :

# src/routes/v1/article.py
from fastapi import APIRouter

from apps.odds_poc_app.src.schemas.article import (
    ArticleListResponse,
    ArticleResponse,
    CreateArticleRequest,
    UpdateArticleRequest,
)
from apps.odds_poc_app.src.services import article as article_service

router = APIRouter(tags=["article"])


@router.post("/articles", response_model=ArticleResponse, status_code=201)
async def create(req: CreateArticleRequest) -> ArticleResponse:
    return await article_service.create(req)


@router.get("/articles/{article_id}", response_model=ArticleResponse)
async def getByIdOrAlias(article_id: str) -> ArticleResponse:
    return await article_service.getByIdOrAlias(article_id)


@router.delete("/articles/{article_id}", status_code=204)
async def delete(article_id: str) -> None:
    await article_service.delete(article_id)

新しい router を作成したら src/routes/__init__.py の load_router() に追加する。

services layer（src/services/）

• ファイル名は対応する routes と揃える（例 : article.py ↔ routes/v1/article.py）
• 関数名は routes と同じ命名規則（create() / get() / getById() / update() / delete() ...）を使う
• ビジネスロジック・バリデーション・複数 repository の連携をここで実装する
• リクエスト DTO（Pydantic）を引数で受け取り、レスポンス DTO（Pydantic）を返す
• 例外は fastapi.HTTPException を raise する（ステータスコードを明示する）

# src/services/article.py（抜粋）
from fastapi import HTTPException

from apps.odds_poc_app.src.repositories import article as article_repository
from apps.odds_poc_app.src.schemas.article import (
    ArticleResponse,
    CreateArticleRequest,
)


async def create(req: CreateArticleRequest) -> ArticleResponse:
    existing = await article_repository.getByAlias(req.alias)
    if existing is not None:
        raise HTTPException(status_code=409, detail="aliasが既に存在します")

    article = await article_repository.create(
        {
            "race_id": req.race_id,
            "alias": req.alias,
            "title": req.title,
            # ...
        }
    )
    return _to_response(article)

• ファイル内ヘルパーは _ プレフィックスを付ける（例 : _to_response()、_validate_and_resolve_media()）
• 同レイヤの services を import / 呼び出ししてはいけない（共通化したい処理は utils/ か libs/ に切り出す）

repositories layer（src/repositories/）

• ファイル名はコレクション単位で作成する（例 : article.py、media.py）
• packages/models/documentDB の Beanie モデルを使用し、DB 操作は必ずこの層に閉じ込める
• 関数名は routes / services と揃える（create() / get() / getById() / getByAlias() / update() / delete()）
• 戻り値は Beanie の Document モデル、もしくはそれを含むタプルとする（Pydantic レスポンス DTO へは services 層で変換する）
• created_at / updated_at / deleted_at の代入は repository が担当する

# src/repositories/article.py
import time

from beanie import PydanticObjectId

from packages.models.documentDB import Article


async def create(data: dict) -> Article:
    now = int(time.time() * 1000)
    article = Article(created_at=now, updated_at=now, **data)
    await article.insert()
    return article


async def getById(article_id: str) -> Article | None:
    return await Article.find_one(
        {"_id": PydanticObjectId(article_id), "deleted_at": None}
    )


async def delete(article: Article) -> None:
    await article.set({"deleted_at": int(time.time() * 1000)})

• 物理削除は禁止。delete() は deleted_at を設定する論理削除のみ
• 全ての参照系クエリで "deleted_at": None を条件に含める
• services を呼び出してはいけない

schemas layer（src/schemas/）

• ファイル名は routes / services と揃える
• Request / Response の Pydantic モデルをここに集約する
• 命名規則 :
• 作成リクエスト : Create{Resource}Request
• 更新リクエスト : Update{Resource}Request（任意項目は T | None = None）
• 1 件レスポンス : {Resource}Response
• 一覧レスポンス : {Resource}ListResponse（list、current_page、total_count を持つ）
• フィールドは snake_case

# src/schemas/article.py（抜粋）
from pydantic import BaseModel


class CreateArticleRequest(BaseModel):
    race_id: str
    alias: str
    title: str
    # ...


class ArticleResponse(BaseModel):
    id: str
    race_id: str
    # ...
    created_at: int
    updated_at: int


class ArticleListResponse(BaseModel):
    list: list[ArticleResponse]
    current_page: int
    total_count: int

libs layer（src/libs/）

• 外部サービスごとに 1 ファイル（documentdb.py、s3.py、sqs.py 等）
• AWS SDK クライアントは モジュールスコープで初期化 する（_client: Any = boto3.client(...)）
• クライアントの戻り値型は Any で型注釈する（boto3 の型情報の都合）
• 公開関数は薄いラッパーとし、services から呼ばれる前提でシグネチャを揃える

# src/libs/sqs.py
import json
from typing import Any

import boto3

from apps.odds_poc_app.src.config.env import settings

_client: Any = boto3.client("sqs", region_name=settings.aws_region)


def send_message(queue_url: str, body: dict) -> None:
    _client.send_message(QueueUrl=queue_url, MessageBody=json.dumps(body))

constants layer（src/constants/）

• ページング等のマジックナンバーは定数として集約する
• 命名は UPPER_SNAKE_CASE、型注釈必須（例 : DEFAULT_PAGE: int = 1）

エラーハンドリング

• HTTP のエラーレスポンスは fastapi.HTTPException を使う
• 例 : 重複は 409、未発見は 404、入力エラーは 400
• グローバルなフォーマット統一が必要な場合は app.add_exception_handler() で対応する
• repositories / libs では原則例外を握りつぶさず、必要なら raise して services 層で HTTPException に変換する

起動・初期化

• アプリのエントリーポイントは src/app.py
• DB 初期化は lifespan（@asynccontextmanager）内で init_db() を呼ぶ
• AWS Lambda 上では Mangum(app) を handler として公開する

---

@apps/article_generation, @apps/blog_generation, @apps/blog_rewriter, @apps/thumbnail_generation（Python Lambda）

FastAPI を使わない、SQS トリガーのイベント駆動型 Lambda。

HTML 構築の Lambda は TypeScript

article_builder / blog_builder は React コンポーネントで HTML を構築するため TypeScript（Node.js 22）で実装します。本章の規約は Python Lambda を対象とします。

共通規約

• Pydantic / pydantic-settings のモデル以外で class を使わない
• すべての関数に型アノテーションを付ける
• 非同期（async/await）は DB 操作（Beanie / motor）など必要な場合のみ 用い、handler.py で asyncio.run() で起動する
• Lambda の同期実行モデルに合わせ、services の主処理関数のシグネチャは async def でも def でもよいが 戻り値は None を基本とする

レイヤ構成と呼び出し規則

handler.py  ─►  services/  ─►  repositories/
                    │               │
                    └──► libs/  ◄───┘

• handler.py は services のみを呼び出す
• services は repositories と libs を呼び出す
• repositories は libs と packages/models/documentDB を参照してよい
• 上位レイヤ（handler.py・services）を下位から呼び出さない

handler.py（プロジェクトルート）

• Lambda のエントリーポイント。lambda_handler(event: dict[str, Any], context: object) -> None
• SQS イベントのループ・ペイロードの Pydantic バリデーション・service 関数の呼び出しのみを行う
• ビジネスロジックを書かない
• 非同期サービスは asyncio.run(...) で起動する
• 例外は そのまま raise し、SQS の再試行 / DLQ 転送に委ねる（捕捉して握りつぶさない）

# apps/article_generation/handler.py
import asyncio
from typing import Any

from apps.article_generation.src.schemas.event import ArticleGenerationPayload
from apps.article_generation.src.services import article as article_service


def lambda_handler(event: dict[str, Any], context: object) -> None:
    for record in event["Records"]:
        payload = ArticleGenerationPayload.model_validate_json(record["body"])
        asyncio.run(article_service.create_generation_job(payload))

schemas layer（src/schemas/event.py）

• SQS メッセージのペイロードを Pydantic モデルで型定義する
• フィールドは snake_case
• モデル名は処理内容に合わせる（例 : ArticleGenerationPayload、MediaAnalysisPayload）

# apps/article_generation/src/schemas/event.py
from pydantic import BaseModel


class ArticleGenerationPayload(BaseModel):
    article_id: str
    race_id: str
    racing_type: Literal["auto_racing", "bicycle_racing", "horse_racing"]
    title: str
    race_result: dict[str, Any]

services layer（src/services/）

• 処理ドメイン単位でファイルを作成する
• ペイロードを受け取り、libs / repositories を協調させて処理を完結させる
• handler.py から呼ばれる public 関数の名前は 処理内容を表す動詞（例 : create_generation_job()、analyze()、censor()）
• 同レイヤの services 同士を呼び出さない
• 例外は raise する（Lambda に委ねる）

# apps/article_generation/src/services/article.py（抜粋）
from apps.article_generation.src.libs import bedrock as bedrock_lib
from apps.article_generation.src.libs import documentdb as db_lib
from apps.article_generation.src.repositories import (
    article_generation as article_generation_repository,
)
from apps.article_generation.src.schemas.event import ArticleGenerationPayload


async def create_generation_job(payload: ArticleGenerationPayload) -> None:
    await db_lib.init_db()
    # ... ビジネスロジック ...
    await article_generation_repository.update_body_html_and_status(
        payload.article_id, body_html
    )

repositories layer（src/repositories/）

• ファイル名はコレクション単位（例 : article_generation.py、media.py、category.py）
• 関数名は処理内容を 動詞 + 対象 で表現する（例 : update_body_html_and_status()、get_by_ids()、get_by_id()）
• services を呼び出してはいけない（libs は呼び出してよい）
• packages/models/documentDB のモデルをここで操作する
• 更新時の updated_at 設定は repository の責務

# apps/article_generation/src/repositories/article_generation.py
import time

from beanie import PydanticObjectId

from packages.models.documentDB import Article, ArticleStatus


async def update_body_html_and_status(article_id: str, body_html: str) -> None:
    article = await Article.find_one({"_id": PydanticObjectId(article_id)})
    if article is not None:
        await article.set(
            {
                "body_html": body_html,
                "status": ArticleStatus.PUBLISHED,
                "updated_at": int(time.time() * 1000),
            }
        )

libs layer（src/libs/）

• 外部サービス単位でファイルを作成する（bedrock.py、s3.py、documentdb.py 等）
• boto3.client(...) などはモジュールスコープで初期化し、client または _client として公開する
• services / repositories を呼び出してはいけない

# apps/article_generation/src/libs/bedrock.py
from typing import Any

import boto3

from apps.article_generation.src.config.env import settings

client: Any = boto3.client("bedrock-runtime", region_name=settings.aws_region)

config layer（src/config/env.py）

• Lambda の環境変数は pydantic-settings で受け取り、settings として公開する
• 他の層からは settings 経由でのみアクセスする

---

レイヤ別 import / 呼び出しの早見表

from \ to	routes	services	repositories	libs	schemas	constants	packages/models
routes	✕	◯	✕	✕	◯	◯	△ (Enum のみ可)
services	✕	✕	◯	◯	◯	◯	△ (Enum のみ可)
repositories	✕	✕	✕	◯	✕	◯	◯
libs	✕	✕	✕	✕	✕	◯	✕
handler.py（Lambda）	–	◯	✕	✕	◯	✕	✕

Beanie の Document モデル（packages/models/documentDB の Article 等）は repositories からのみ操作 し、routes / services では Enum（ArticleStatus、MediaType 等）の参照に留める。