fastapi-entity

name	fastapi-entity
description	Create a new entity with model, schemas, repository, service, router, dependencies, and filters

FastAPI Entity Creation

Overview

This skill covers creating a complete entity with all necessary files following the entity-based folder structure. Each entity is self-contained with its own model, schemas, repository, service, router, dependencies, and filters.

Entity Folder Structure

For an entity named items:

src/app/items/
├── __init__.py
├── models.py          # SQLAlchemy model
├── schemas.py         # Pydantic schemas
├── repository.py      # Data access layer
├── service.py         # Business logic layer
├── router.py          # API endpoints
├── dependencies.py    # FastAPI dependencies
└── filters.py         # fastapi-filter definitions

Step 1: Create init.py

Create src/app/{entity}/__init__.py:

from app.items.models import Item
from app.items.repository import ItemRepository
from app.items.router import router
from app.items.schemas import ItemCreate, ItemResponse, ItemUpdate
from app.items.service import ItemService

__all__ = [
    "Item",
    "ItemCreate",
    "ItemUpdate",
    "ItemResponse",
    "ItemRepository",
    "ItemService",
    "router",
]

Step 2: Create models.py

Create src/app/{entity}/models.py:

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column

from app.core.models import Base, SoftDeleteMixin, TimestampMixin, UUIDMixin


class Item(UUIDMixin, TimestampMixin, SoftDeleteMixin, Base):
    """
    Item model.
    
    Attributes:
        id: UUID primary key
        name: Item name (required)
        description: Item description (optional)
        created_at: Creation timestamp
        updated_at: Last update timestamp
        deleted_at: Soft delete timestamp
    """

    __tablename__ = "items"

    name: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)

Step 3: Create schemas.py

Create src/app/{entity}/schemas.py:

from pydantic import Field

from app.core.schemas import (
    BaseCreateSchema,
    BaseResponseSchema,
    BaseResponseWithDeletedSchema,
    BaseUpdateSchema,
)


class ItemCreate(BaseCreateSchema):
    """Schema for creating an item."""

    name: str = Field(
        ...,
        min_length=1,
        max_length=255,
        description="Item name",
        examples=["My Item"],
    )
    description: str | None = Field(
        default=None,
        max_length=5000,
        description="Item description",
        examples=["A detailed description of the item"],
    )


class ItemUpdate(BaseUpdateSchema):
    """Schema for updating an item. All fields optional for PATCH."""

    name: str | None = Field(
        default=None,
        min_length=1,
        max_length=255,
        description="Item name",
    )
    description: str | None = Field(
        default=None,
        max_length=5000,
        description="Item description",
    )


class ItemResponse(BaseResponseSchema):
    """Schema for item responses."""

    name: str
    description: str | None


class ():
    

    name: 
    description:  | 

Step 4: Create repository.py

Create src/app/{entity}/repository.py:

from sqlalchemy.ext.asyncio import AsyncSession

from app.common.postgres_repository import PostgresRepository
from app.items.models import Item
from app.items.schemas import ItemCreate, ItemUpdate


class ItemRepository(PostgresRepository[Item, ItemCreate, ItemUpdate]):
    """
    Repository for Item entity.
    
    Inherits all CRUD, pagination, filtering, bulk operations,
    and soft delete methods from PostgresRepository.
    
    Add entity-specific query methods here.
    """

    def __init__(self, session: AsyncSession):
        super().__init__(session, Item)

    async def get_by_name(self, name: str) -> Item | None:
        """
        Get item by name.
        
        Args:
            name: Item name to search for
            
        Returns:
            Item if found, None otherwise
        """
        return await self.get_by_field("name", name)

Step 5: Create service.py

Create src/app/{entity}/service.py:

from uuid import UUID

from app.core.service import BaseService
from app.exceptions import ConflictError, NotFoundError
from app.items.models import Item
from app.items.repository import ItemRepository
from app.items.schemas import ItemCreate, ItemUpdate


class ItemService(BaseService[Item, ItemCreate, ItemUpdate]):
    """
    Service for Item entity.
    
    Contains business logic and validation rules.
    Delegates data access to the repository.
    """

    def __init__(self, repository: ItemRepository):
        super().__init__(repository)
        self._repository: ItemRepository = repository

    async def create(self, obj_in: ItemCreate) -> Item:
        """
        Create a new item.
        
        Validates that the name is unique before creation.
        
        Args:
            obj_in: Item creation data
            
        Returns:
            Created item
            
        Raises:
            ConflictError: If item with same name exists
        """
        existing = await self._repository.get_by_name(obj_in.name)
        if existing:
            raise ConflictError(
                resource="Item",
                field="name",
                value=obj_in.name,
            )
        return await super().create(obj_in)

    async def get_by_id_or_raise(self, : UUID) -> Item:
        
        item =  .get_by_id()
          item:
             NotFoundError(resource=, =)
         item

      () -> Item:
        
        
         .get_by_id_or_raise()

        
         obj_in.name   :
            existing =  ._repository.get_by_name(obj_in.name)
             existing  existing. != :
                 ConflictError(
                    resource=,
                    field=,
                    value=obj_in.name,
                )

        item =  ().update(, obj_in, exclude_unset)
         item  :
             NotFoundError(resource=, =)
         item

      () -> :
        
         .get_by_id_or_raise()
          .delete()

      () -> :
        
         .get_by_id_or_raise()
          .soft_delete()

Step 6: Create filters.py

Create src/app/{entity}/filters.py:

from typing import Optional

from fastapi_filter.contrib.sqlalchemy import Filter

from app.items.models import Item


class ItemFilter(Filter):
    """
    Filter specification for Item queries.
    
    Supports filtering by:
    - name: Exact match
    - name__ilike: Case-insensitive partial match
    - description__ilike: Case-insensitive partial match
    
    Supports ordering by any field.
    """

    # Exact match
    name: Optional[str] = None

    # Case-insensitive contains
    name__ilike: Optional[str] = None
    description__ilike: Optional[str] = None

    # Ordering
    order_by: Optional[list[str]] = None

    class Constants(Filter.Constants):
        model = Item
        ordering_field_name = "order_by"

Step 7: Create dependencies.py

Create src/app/{entity}/dependencies.py:

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.dependencies import get_db
from app.items.repository import ItemRepository
from app.items.service import ItemService


def get_item_repository(
    session: AsyncSession = Depends(get_db),
) -> ItemRepository:
    """
    Dependency that provides an ItemRepository instance.
    
    Args:
        session: Database session from get_db dependency
        
    Returns:
        ItemRepository instance
    """
    return ItemRepository(session)


def get_item_service(
    repository: ItemRepository = Depends(get_item_repository),
) -> ItemService:
    """
    Dependency that provides an ItemService instance.
    
    Args:
        repository: ItemRepository from get_item_repository dependency
        
    Returns:
        ItemService instance
    """
    return ItemService(repository)

Step 8: Create router.py

Create src/app/{entity}/router.py:

from uuid import UUID

from fastapi import APIRouter, Depends, status
from fastapi_filter import FilterDepends
from fastapi_pagination import Page, Params

from app.items.dependencies import get_item_service
from app.items.filters import ItemFilter
from app.items.models import Item
from app.items.schemas import ItemCreate, ItemResponse, ItemUpdate
from app.items.service import ItemService

router = APIRouter(prefix="/items", tags=["items"])


@router.post(
    "",
    response_model=ItemResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Create item",
    description="Create a new item with the provided data.",
)
async def create_item(
    obj_in: ItemCreate,
    service: ItemService = Depends(get_item_service),
) -> Item:
    """Create a new item."""
    return await service.create(obj_in)


@router.get(
    "",
    response_model=Page[ItemResponse],
    summary="List items",
    description="Get a paginated list of items with optional filtering.",
)
async def list_items(
    params: Params = Depends(),
    filter_spec: ItemFilter = FilterDepends(ItemFilter),
    service: ItemService = Depends(),
) -> Page[Item]:
    
      service.get_paginated(params, filter_spec)



  () -> Item:
    
      service.get_by_id_or_raise(item_id)



  () -> Item:
    
      service.update(item_id, obj_in)



  () -> :
    
     service.soft_delete_or_raise(item_id)



  () -> Item:
    
     service.restore(item_id)
      service.get_by_id_or_raise(item_id)

Step 9: Register Router

Update src/app/api/v1/__init__.py:

from fastapi import APIRouter

from app.items.router import router as items_router

router = APIRouter()

router.include_router(items_router)

Step 10: Register Model for Alembic

Update alembic/env.py:

# Import models for autogenerate
from app.items.models import Item  # noqa: F401

Step 11: Create Migration

uv run alembic revision --autogenerate -m "add items table"
uv run alembic upgrade head

Generated API Endpoints

Method	Path	Description
POST	/api/v1/items	Create item
GET	/api/v1/items	List items (paginated)
GET	/api/v1/items/{id}	Get item by ID
PATCH	/api/v1/items/{id}	Update item
DELETE	/api/v1/items/{id}	Soft delete item
POST	/api/v1/items/{id}/restore	Restore item

Query Parameters for List Endpoint

Pagination:

• page: Page number (default: 1)
• size: Items per page (default: 50)

Filtering:

• name: Exact name match
• name__ilike: Case-insensitive name contains
• description__ilike: Case-insensitive description contains

Ordering:

• order_by: Field to order by (prefix with - for descending)

Example:

GET /api/v1/items?page=1&size=20&name__ilike=widget&order_by=-created_at

Adding Relationships

For entities with relationships:

# models.py
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Item(UUIDMixin, TimestampMixin, SoftDeleteMixin, Base):
    __tablename__ = "items"
    
    name: Mapped[str] = mapped_column(String(255))
    category_id: Mapped[UUID] = mapped_column(ForeignKey("categories.id"))
    
    # Relationship
    category: Mapped["Category"] = relationship(back_populates="items")


# schemas.py
class ItemResponse(BaseResponseSchema):
    name: str
    category_id: UUID
    
class ItemWithCategoryResponse(BaseResponseSchema):
    name: str
    category: CategoryResponse  # Nested response