module

Command /fastapi:module

Create a complete FastAPI module with best practices.

Module Structure

src/modules/{name}/
├── __init__.py
├── models.py       # SQLAlchemy model
├── dto.py          # Pydantic DTOs (Create, Read, Update)
├── repositories.py # Repository with generics
├── services.py     # Business logic
└── routers.py      # FastAPI endpoints

tests/modules/{name}/
├── __init__.py
├── conftest.py     # Module fixtures
├── test_services.py
└── test_routers.py

Instructions

Step 1: Create directories

mkdir -p src/modules/{{ name }}
mkdir -p tests/modules/{{ name }}

Step 2: Create module files

models.py

"""{{ name | capitalize }} models."""

from sqlalchemy import String, Boolean, Integer
from sqlalchemy.orm import Mapped, mapped_column

from src.core.database import Base


class {{ name | capitalize }}(Base):
    """{{ name | capitalize }} model."""

    __tablename__ = "{{ name }}"

    id: Mapped[int] = mapped_column(primary_key=True)
    # Add fields based on {{ fields }}

    def __repr__(self) -> str:
        return f"<{{ name | capitalize }}(id={self.id})>"

dto.py

"""{{ name | capitalize }} DTOs."""

from pydantic import BaseModel, ConfigDict


class {{ name | capitalize }}BaseDTO(BaseModel):
    """Base DTO with common fields."""

    model_config = ConfigDict(from_attributes=True, populate_by_name=True)

    # Common fields
    pass


class {{ name | capitalize }}CreateDTO({{ name | capitalize }}BaseDTO):
    """DTO for creating {{ name }}."""

    # Fields for creation (without id)
    pass


class {{ name | capitalize }}ReadDTO({{ name | capitalize }}BaseDTO):
    """DTO for reading {{ name }}."""

    id: int
    # All fields for reading


class {{ name | capitalize }}UpdateDTO(BaseModel):
    """DTO for updating {{ name }}."""

    model_config = ConfigDict(from_attributes=True)

    # Optional fields for update
    pass

repositories.py

"""{{ name | capitalize }} repository."""

from collections.abc import Sequence

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.core.repositories import BaseRepository
from .models import {{ name | capitalize }}
from .dto import {{ name | capitalize }}CreateDTO, {{ name | capitalize }}ReadDTO


class {{ name | capitalize }}Repository(
    BaseRepository[{{ name | capitalize }}, {{ name | capitalize }}CreateDTO, {{ name | capitalize }}ReadDTO]
):
    """Repository for {{ name }} operations."""

    def __init__(self, session: AsyncSession) -> None:
        super().__init__(session, {{ name | capitalize }})

    async def get_by_id(self, id: int) -> {{ name | capitalize }}ReadDTO | None:
        """Get {{ name }} by ID."""
        result = await self._session.execute(
            select({{ name | capitalize }}).where({{ name | capitalize }}.id == id)
        )
        entity = result.scalar_one_or_none()
        return {{ name | capitalize }}ReadDTO.model_validate(entity) if entity else None

    # Add specific methods