코딩 컨벤션

truloop-ai-server에서 사용하는 Python/FastAPI 코딩 컨벤션과 프로젝트 패턴을 정리합니다.

---

코드 품질 도구

도구	버전	용도	설정 파일
Ruff	>=0.12.8	포맷팅 + 린팅 (isort 포함)	pyproject.toml [tool.ruff]
mypy	>=1.17.1	타입 체크	pyproject.toml [tool.mypy]
pre-commit	>=4.3.0	Git 훅 기반 자동 검사	.pre-commit-config.yaml

Ruff 설정

[tool.ruff]
line-length = 120
target-version = "py311"

설정	값	설명
line-length	120	최대 줄 길이
quote-style	"double"	큰따옴표 사용
indent-style	"space"	스페이스 들여쓰기

활성화된 lint 규칙: E (pycodestyle), W (warnings), F (pyflakes), I (isort), B (bugbear), C4 (comprehensions), UP (pyupgrade), SIM (simplify).

mypy 설정

• 점진적 도입 전략: allow_untyped_defs = true (초기 단계)
• 플러그인: pydantic.mypy, sqlalchemy.ext.mypy.plugin
• tasks/ 모듈은 ignore_errors = true (Celery 타입 이슈)

---

프로젝트 구조 컨벤션

디렉토리 구조

app/
├── api/v1/{도메인}/     # API 엔드포인트 (라우터)
├── common/              # 공통 Enum, 타입 정의
├── config/              # 설정 (로깅, 메트릭 등)
├── core/                # 핵심 인프라 (DB, Redis, Sentry 등)
├── models/              # SQLAlchemy ORM 모델
├── schemas/             # Pydantic 입출력 스키마
├── services/            # 비즈니스 로직
│   ├── {도메인}/        # 도메인별 서비스 클래스
│   ├── interfaces/      # ABC 인터페이스
│   └── common/          # 공통 유틸리티
├── middleware/           # FastAPI 미들웨어
├── templates/           # Jinja2 HTML 템플릿
└── static/              # 정적 파일
tasks/                   # Celery 태스크 (app/ 외부)

파일 명명 규칙

유형	패턴	예시
API 라우터	{기능}.py	content_generation.py, image_analysis.py
서비스	{기능}_service.py	content_generation_service.py
모델	{테이블명_snake}.py	loop_media.py, loop_generated_content.py
스키마	{도메인}.py	content_template.py, common.py
Celery 태스크	{기능}_tasks.py	content_generation_tasks.py
인터페이스	{역할}.py	content_generator.py, notification_service.py

---

아키텍처 패턴

Service Layer Pattern

비즈니스 로직은 app/services/ 아래의 서비스 클래스에 캡슐화합니다. API 라우터에는 요청 파싱, 유효성 검사, 응답 구성만 위치합니다.

# API 라우터 - 요청 처리와 서비스 호출만
@router.post("/highlights")
async def generate_highlight_content(
    request_data: GenerateHighlightContentRequest,
    db: Session = Depends(get_db),
):
    service = ContentGenerationService(db)
    generated_content, estimated_time = await service.generate_highlight_content_with_task(...)
    return ApiResponse.success(data=..., extras={"estimated_completion_time": estimated_time})

DB 세션 주입

FastAPI의 Depends(get_db)를 통해 DB 세션을 주입합니다. 서비스 클래스는 생성자에서 Session을 받습니다.

class ContentGenerationService:
    def __init__(self, db: Session):
        self.db = db
        self.media_service = MediaService(db)

Celery 태스크에서의 DB 세션

Celery Worker는 FastAPI DI 컨테이너를 사용하지 않으므로, SessionLocal()로 직접 세션을 생성하고 try/finally로 반드시 닫습니다.

@current_app.task(bind=True, max_retries=0)
def generate_content_task(self, ...):
    db = SessionLocal()
    try:
        service = ContentGenerationService(db)
        async_to_sync(service._process_content_generation)(...)
    finally:
        db.close()

동기/비동기 브릿지

Celery Worker(동기)에서 async 서비스 메서드를 호출할 때는 asgiref.sync.async_to_sync를 사용합니다.

인터페이스 패턴

외부 서비스 연동은 app/services/interfaces/에 ABC 인터페이스를 정의하고 구현합니다.

class ContentGeneratorInterface(ABC):
    @abstractmethod
    async def generate_content(self, ...) -> dict[str, Any]:
        pass

class JunisContentGenerator(ContentGeneratorInterface):
    async def generate_content(self, ...) -> dict[str, Any]:
        ...

---

API 컨벤션

응답 포맷

모든 API는 ApiResponse 유틸리티 클래스를 사용하여 일관된 응답 형식을 유지합니다.

# 성공 응답
return ApiResponse.success(data=result, extras={"key": "value"}, message="성공")
# → {"success": True, "message": "성공", "data": ..., "extras": ...}

# 에러 응답
return ApiResponse.error(message="실패", data={"error": "상세 정보"})
# → {"success": False, "message": "실패", "data": {"error": "상세 정보"}}

경로 prefix

모든 라우터는 API_BASE_PATH 환경변수 prefix가 적용됩니다. 라우터 등록 시:

app.include_router(
    content_generation.router,
    prefix=API_BASE_PATH + "/api/content-generation",
    tags=["content-generation"],
)

Pydantic 스키마

• 입력 스키마: app/schemas/ 아래에 Pydantic BaseModel로 정의
• 출력 스키마: model_validate()로 SQLAlchemy 모델을 Pydantic 모델로 변환
• 제네릭 응답 모델: ApiResponseModel[T] 사용

---

모델 컨벤션

SQLAlchemy 모델

• DeclarativeBase (app/core/database.Base) 상속
• Mapped + mapped_column 스타일 사용 (SQLAlchemy 2.0)
• DB 스키마: truloop (SET search_path TO truloop)
• 모든 컬럼에 comment 파라미터로 설명 기재

class LoopGeneratedContent(Base):
    __tablename__ = "loop_generated_content"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True, comment="생성된 콘텐츠 고유 ID")
    status: Mapped[ContentStatus] = mapped_column(
        Enum(ContentStatus, name="loop_generated_content_status_enum", native_enum=False),
        nullable=False,
        default=ContentStatus.PENDING,
        comment="콘텐츠 생성 상태",
    )

Enum 정의

• 비즈니스 Enum은 app/common/types.py에 정의 (TemplateType, MediaContentType)
• 모델 전용 Enum은 해당 모델 파일에 정의 (ContentStatus, PromptCategory)
• 모든 Enum은 str을 상속하여 JSON 직렬화 가능하게 구현

class TemplateType(str, Enum):
    HIGHLIGHT_DEFAULT = "highlight_default"
    POSTER_STYLE = "poster_style"
    STORY = "story"

읽기 전용 모델

외부 서비스가 관리하는 테이블의 모델은 docstring에 읽기 전용임을 명시합니다.

class LoopMedia(Base):
    """
    Query-only model for loop_media table.
    All inserts are handled by external Media API.
    """

---

외부 API 호출 컨벤션

HTTP 클라이언트

• 비동기 HTTP 호출: httpx.AsyncClient 사용
• OpenRouter LLM 호출: openai.AsyncOpenAI (OpenRouter base URL 설정)
• 클라이언트는 반드시 finally 블록에서 aclose() 호출

Rate Limiting

ClaudeRateLimiter로 동시 요청과 분당 요청 수를 제어합니다. async with 문으로 사용합니다.

self.rate_limiter = ClaudeRateLimiter(max_concurrent=3, max_requests_per_minute=50)

async with self.rate_limiter:
    result = await call_claude_with_retry(...)

메트릭 데코레이터

외부 API 호출에는 @track_external_api 데코레이터를 적용하여 Prometheus 메트릭을 자동 수집합니다.

@track_external_api("openrouter", "claude")
async def call_claude_with_retry(model, max_tokens, temperature, system, messages, max_retries=3):
    ...

---

로깅 컨벤션

• 모듈별 logging.getLogger(__name__) 사용
• 에러 시 logger.error() + logger.exception() 조합으로 스택트레이스 포함
• 민감 정보(API 키, 비밀번호)는 로그에 포함하지 않음
• Sentry에 에러 보고 시 sentry_sdk.new_scope()로 컨텍스트 추가

---

테스트

주의

현재 pytest 테스트는 진행하지 않습니다. tests/ 디렉토리와 pytest 설정은 존재하지만 실제 테스트 실행은 추후 적용 예정입니다.

---

변경 이력

날짜	내용
2026-03-10	초기 작성: 소스 코드 기반 코딩 컨벤션 문서화