Coding Rules

For the directory layout, see Directory Architecture. This document defines implementation rules (naming, typing, inter-layer calls, data access, error handling, etc.).

---

Common Rules (All Apps)

Language & Tooling

• Python 3.12
• Formatter / Linter: ruff (line-length 88, double quotes)
• Type checker: mypy (strict = True)
• Dependency manager: uv

Naming & Style

• File names, module names, variable names, and function names use snake_case
• Constants use UPPER_SNAKE_CASE (e.g. DEFAULT_PAGE, DEFAULT_SIZE)
• Class names use PascalCase (e.g. ArticleResponse, CreateArticleRequest)
• Do not use class outside of Pydantic / pydantic-settings model definitions
• Business logic is expressed as functions; state lives at module scope
• All functions and methods must have type annotations (per mypy strict)
• Date/time fields stored in the DB use UNIX milliseconds: int(time.time() * 1000)
• Logical-delete field is deleted_at (None means not deleted)

Imports

• To avoid name collisions, cross-layer imports use aliases:
• 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 order follows ruff's I (isort): stdlib / third-party / first-party

Configuration (Environment Variables)

• Environment variables are centralized in Settings(BaseSettings) in src/config/env.py
• All layers access them via settings — never read os.environ directly

# 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 Design Rules

• URI segments and JSON node names use snake_case
• Request / response bodies are JSON
• Meta information (paging, counts, etc.) goes in the response body by default; use HTTP headers only when required
• API version is managed via a path prefix: /api/v1/... (load_router() applies prefix="/api/v1")
• Authentication uses an API Key at API Gateway

Layer Layout & Call Rules

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

• routes only calls services
• services calls repositories and libs
• repositories may reference libs and packages/models/documentDB
• Functions in the same layer do not call each other (e.g. a service must not call another service)
• Lower layers must not call upper layers (no repositories → services, no services → routes, etc.)

routes layer (src/routes/v1/)

• Create one file per resource (e.g. article.py, media.py)
• Define router = APIRouter(tags=[...]) at module scope in each file
• Use FastAPI decorators for routing
• Handler functions must be async def with full type annotations on parameters and return values
• Handler bodies only call a service function — no business logic in routes
• Request / response validation uses Pydantic models from src/schemas/
• Function names follow the fixed convention below, based on HTTP method + target:

Purpose	Function name
Create (POST)	create()
List (GET)	get()
Get one by id (GET)	getById()
Get one by alias (GET)	getByAlias()
Get one by id-or-alias (GET)	getByIdOrAlias()
Update (PUT)	update()
Delete (DELETE)	delete()

The camelCase function names are explicitly allowed by excluding ruff's N802 rule (see ruff.toml). New endpoints must follow the same naming.

Example:

# 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)

Register a new router in load_router() inside src/routes/__init__.py.

services layer (src/services/)

• Filenames mirror the routes (e.g. article.py ↔ routes/v1/article.py)
• Functions use the same naming convention as routes (create() / get() / getById() / update() / delete(), etc.)
• Business logic, validation, and orchestration across multiple repositories live here
• Functions take a request DTO (Pydantic) and return a response DTO (Pydantic)
• Raise fastapi.HTTPException for errors, with an explicit status code

# src/services/article.py (excerpt)
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 already exists")

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

• File-local helpers are prefixed with _ (e.g. _to_response(), _validate_and_resolve_media())
• Services must not import or call other services. Extract shared logic into utils/ or libs/ instead.

repositories layer (src/repositories/)

• One file per collection (e.g. article.py, media.py)
• Use the Beanie models from packages/models/documentDB. All DB operations must be confined to this layer.
• Function names match routes / services (create() / get() / getById() / getByAlias() / update() / delete())
• Return Beanie Document models (or tuples containing them). Convert to Pydantic response DTOs in the services layer.
• Setting created_at / updated_at / deleted_at is the repository's responsibility

# 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)})

• Physical delete is forbidden. delete() performs a logical delete by setting deleted_at.
• Every read query must include "deleted_at": None in its filter
• Repositories must not call services

schemas layer (src/schemas/)

• Filenames mirror routes / services
• This is where Request / Response Pydantic models are centralized
• Naming convention:
• Create request: Create{Resource}Request
• Update request: Update{Resource}Request (optional fields are T | None = None)
• Single response: {Resource}Response
• List response: {Resource}ListResponse (fields: list, current_page, total_count)
• Field names use snake_case

# src/schemas/article.py (excerpt)
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/)

• One file per external service (documentdb.py, s3.py, sqs.py, etc.)
• AWS SDK clients are initialized at module scope (_client: Any = boto3.client(...))
• Annotate the client as Any (boto3 has no first-class type info)
• Public functions are thin wrappers with signatures shaped for the services callers

# 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/)

• Centralize magic numbers (paging defaults, etc.) as constants
• Use UPPER_SNAKE_CASE and require type annotations (e.g. DEFAULT_PAGE: int = 1)

Error Handling

• HTTP error responses use fastapi.HTTPException
• e.g. duplicate → 409, not found → 404, invalid input → 400
• For a globally unified error format, register handlers via app.add_exception_handler()
• Repositories / libs must not swallow exceptions. Re-raise so services can translate them into HTTPException.

Startup & Initialization

• The app entry point is src/app.py
• DB initialization happens inside the lifespan (@asynccontextmanager) by calling init_db()
• On AWS Lambda, Mangum(app) is exposed as handler

---

@apps/article_generation, @apps/blog_generation, @apps/blog_rewriter, @apps/thumbnail_generation (Python Lambdas)

Event-driven Lambda functions triggered by SQS — no FastAPI.

The HTML builders are TypeScript

article_builder / blog_builder build HTML from React components and are written in TypeScript (Node.js 22). The rules in this section apply to the Python Lambdas.

Common Rules

• Do not use class outside of Pydantic / pydantic-settings models
• All functions require type annotations
• Use async/await only where it is required (e.g. Beanie / motor DB operations). The handler.py launches it with asyncio.run().
• To match Lambda's synchronous execution model, a service's main function can be either async def or def, but its return type should be None by default

Layer Layout & Call Rules

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

• handler.py only calls services
• services calls repositories and libs
• repositories may reference libs and packages/models/documentDB
• Lower layers must not call upper layers (no calls back into handler.py or services)

handler.py (project root)

• Lambda entry point: lambda_handler(event: dict[str, Any], context: object) -> None
• Only iterates SQS records, validates the payload with Pydantic, and calls a service function
• No business logic here
• Launch async services with asyncio.run(...)
• Re-raise exceptions — do not swallow them. Let SQS handle retries / DLQ forwarding.

# 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)

• Define the SQS message payload as a Pydantic model
• Fields use snake_case
• Model names reflect the operation (e.g. 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/)

• One file per processing domain
• Receives a payload and orchestrates libs / repositories to complete the work
• The public function called from handler.py is named with a descriptive verb (e.g. create_generation_job(), analyze(), censor())
• Services must not call other services in the same layer
• Raise exceptions on failure — let Lambda handle them

# apps/article_generation/src/services/article.py (excerpt)
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()
    # ... business logic ...
    await article_generation_repository.update_body_html_and_status(
        payload.article_id, body_html
    )

repositories layer (src/repositories/)

• One file per collection (e.g. article_generation.py, media.py, category.py)
• Function names follow the verb + target pattern (e.g. update_body_html_and_status(), get_by_ids(), get_by_id())
• Repositories must not call services (calling libs is allowed)
• DocumentDB models from packages/models/documentDB are manipulated here
• Setting updated_at on updates is the repository's responsibility

# 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/)

• One file per external service (bedrock.py, s3.py, documentdb.py, etc.)
• Initialize boto3.client(...) at module scope and expose it as client or _client
• Libs must not call services or 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)

• Receive Lambda environment variables via pydantic-settings and expose them as settings
• Other layers access them via settings only

---

Inter-Layer Import / Call Cheatsheet

from \ to	routes	services	repositories	libs	schemas	constants	packages/models
routes	✕	◯	✕	✕	◯	◯	△ (Enums only)
services	✕	✕	◯	◯	◯	◯	△ (Enums only)
repositories	✕	✕	✕	◯	✕	◯	◯
libs	✕	✕	✕	✕	✕	◯	✕
handler.py (Lambda)	–	◯	✕	✕	◯	✕	✕

Beanie Document models (e.g. Article in packages/models/documentDB) are manipulated only in repositories. In routes / services, restrict references to enums (ArticleStatus, MediaType, etc.).