Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
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",
]
from pydantic import Field
from app.core.schemas import (
BaseCreateSchema,
BaseResponseSchema,
BaseResponseWithDeletedSchema,
BaseUpdateSchema,
)
classItemCreate(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"],
)
classItemUpdate(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",
)
classItemResponse(BaseResponseSchema):
"""Schema for item responses."""
name: str
description: str | NoneclassItemResponseWithDeleted(BaseResponseWithDeletedSchema):
"""Schema for item responses including soft delete info."""
name: str
description: str | None
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
classItemRepository(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)
asyncdefget_by_name(self, name: str) -> Item | None:
"""
Get item by name.
Args:
name: Item name to search for
Returns:
Item if found, None otherwise
"""returnawaitself.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
classItemService(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
asyncdefcreate(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 = awaitself._repository.get_by_name(obj_in.name)
if existing:
raise ConflictError(
resource="Item",
field="name",
value=obj_in.name,
)
returnawaitsuper().create(obj_in)
asyncdefget_by_id_or_raise(self, id: UUID) -> Item:
"""
Get item by ID or raise NotFoundError.
Args:
id: Item UUID
Returns:
Item instance
Raises:
NotFoundError: If item not found
"""
item = awaitself.get_by_id(id)
ifnot item:
raise NotFoundError(resource="Item", id=id)
return item
asyncdefupdate(
self,
id: UUID,
obj_in: ItemUpdate,
exclude_unset: bool = True,
) -> Item:
"""
Update an item.
Validates name uniqueness if name is being changed.
Args:
id: Item UUID to update
obj_in: Update data
exclude_unset: Only update explicitly set fields
Returns:
Updated item
Raises:
NotFoundError: If item not found
ConflictError: If new name conflicts with existing item
"""# Ensure item existsawaitself.get_by_id_or_raise(id)
# Check name uniqueness if being updatedif obj_in.name isnotNone:
existing = awaitself._repository.get_by_name(obj_in.name)
if existing and existing.id != id:
raise ConflictError(
resource="Item",
field="name",
value=obj_in.name,
)
item = awaitsuper().update(id, obj_in, exclude_unset)
if item isNone:
raise NotFoundError(resource="Item", id=id)
return item
asyncdefdelete_or_raise(self, id: UUID) -> bool:
"""
Delete an item or raise if not found.
Args:
id: Item UUID to delete
Returns:
True if deleted
Raises:
NotFoundError: If item not found
"""awaitself.get_by_id_or_raise(id)
returnawaitself.delete(id)
asyncdefsoft_delete_or_raise(self, id: UUID) -> bool:
"""
Soft delete an item or raise if not found.
Args:
id: Item UUID to soft delete
Returns:
True if soft deleted
Raises:
NotFoundError: If item not found
"""awaitself.get_by_id_or_raise(id)
returnawaitself.soft_delete(id)
Step 6: Create filters.py
Create src/app/{entity}/filters.py:
from typing importOptionalfrom fastapi_filter.contrib.sqlalchemy import Filter
from app.items.models import Item
classItemFilter(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]] = NoneclassConstants(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
defget_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)
defget_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.",
)asyncdefcreate_item(
obj_in: ItemCreate,
service: ItemService = Depends(get_item_service),
) -> Item:
"""Create a new item."""returnawait service.create(obj_in)
@router.get("",
response_model=Page[ItemResponse],
summary="List items",
description="Get a paginated list of items with optional filtering.",
)asyncdeflist_items(
params: Params = Depends(),
filter_spec: ItemFilter = FilterDepends(ItemFilter),
service: ItemService = Depends(get_item_service),
) -> Page[Item]:
"""List items with pagination and filtering."""returnawait service.get_paginated(params, filter_spec)
@router.get("/{item_id}",
response_model=ItemResponse,
summary="Get item",
description="Get a single item by ID.",
)asyncdefget_item(
item_id: UUID,
service: ItemService = Depends(get_item_service),
) -> Item:
"""Get item by ID."""returnawait service.get_by_id_or_raise(item_id)
@router.patch("/{item_id}",
response_model=ItemResponse,
summary="Update item",
description="Update an existing item. Only provided fields will be updated.",
)asyncdefupdate_item(
item_id: UUID,
obj_in: ItemUpdate,
service: ItemService = Depends(get_item_service),
) -> Item:
"""Update item by ID."""returnawait service.update(item_id, obj_in)
@router.delete("/{item_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete item",
description="Soft delete an item by ID.",
)asyncdefdelete_item(
item_id: UUID,
service: ItemService = Depends(get_item_service),
) -> None:
"""Soft delete item by ID."""await service.soft_delete_or_raise(item_id)
@router.post("/{item_id}/restore",
response_model=ItemResponse,
summary="Restore item",
description="Restore a soft-deleted item.",
)asyncdefrestore_item(
item_id: UUID,
service: ItemService = Depends(get_item_service),
) -> Item:
"""Restore a soft-deleted item."""await service.restore(item_id)
returnawait 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)
