Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
Create base service class that wraps repository operations and provides business logic hooks
FastAPI Core Service
Overview
This skill covers creating the base service class that acts as an intermediary between routers and repositories. Services contain business logic and orchestrate repository calls.
Create core/service.py
Create src/app/core/service.py:
from collections.abc importSequencefrom typing importGenericfrom uuid import UUID
from fastapi_filter.contrib.sqlalchemy import Filter
from fastapi_pagination import Params
from fastapi_pagination.bases import AbstractPage
from app.core.repository import (
AbstractRepository,
CreateSchemaType,
ModelType,
UpdateSchemaType,
)
classBaseService(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
"""
Base service class providing standard CRUD operations.
Services:
- Wrap repository operations
- Contain business logic
- Orchestrate multiple repository calls
- Handle cross-cutting concerns (logging, events, etc.)
Services should NOT:
- Write SQL queries (delegate to repository)
- Handle HTTP concerns (that's the router's job)
- Access the database session directly
Type Parameters:
ModelType: SQLAlchemy model class
CreateSchemaType: Pydantic schema for create operations
UpdateSchemaType: Pydantic schema for update operations
Usage:
class ItemService(BaseService[Item, ItemCreate, ItemUpdate]):
pass
"""
():
._repository = repository
() -> ModelType:
._repository.create(obj_in)
() -> ModelType | :
._repository.get_by_id()
() -> [ModelType]:
._repository.get_all()
() -> ModelType | :
._repository.update(, obj_in, exclude_unset)
() -> :
._repository.delete()
() -> AbstractPage[ModelType]:
._repository.get_paginated(params, filter_spec)
() -> [ModelType]:
._repository.get_filtered(filter_spec)
() -> :
._repository.count(filter_spec)
() -> [ModelType]:
._repository.bulk_create(objs_in)
() -> [ModelType]:
._repository.bulk_upsert(
objs_in, index_elements, update_fields
)
() -> :
._repository.bulk_delete(ids)
() -> :
._repository.soft_delete()
() -> :
._repository.restore()
() -> ModelType | :
._repository.get_by_id_with_deleted()
() -> :
._repository.exists()
() -> [ModelType]:
._repository.get_by_ids(ids)
"""
Update an existing record.
Override to add business logic before/after update.
Args:
id: UUID of record to update
obj_in: Update data
exclude_unset: Only update explicitly set fields
Returns:
Updated model instance if found, None otherwise
"""
return
await
self
id
async
def
delete
self, id: UUID
bool
"""
Permanently delete a record.
Args:
id: UUID of record to delete
Returns:
True if deleted, False if not found
"""
"""
Get paginated records with optional filtering.
This method receives the filter from the router layer and
passes it to the repository.
Args:
params: Pagination parameters
filter_spec: Optional filter specification
Returns:
Paginated result
"""
return
await
self
async
def
get_filtered
self,
filter_spec: Filter,
Sequence
"""
Get all records matching filter.
Args:
filter_spec: Filter specification
Returns:
Matching records
"""
return
await
self
async
def
count
self, filter_spec: Filter | None = None
int
"""
Count records.
Args:
filter_spec: Optional filter specification
Returns:
Number of matching records
"""
"""
Upsert multiple records.
Args:
objs_in: Records to upsert
index_elements: Unique constraint columns
update_fields: Fields to update on conflict
Returns:
Upserted model instances
"""
return
await
self
async
def
bulk_delete
self, ids: Sequence[UUID]
int
"""
Delete multiple records.
Args:
ids: UUIDs to delete
Returns:
Number deleted
"""
"""
Check if record exists.
Args:
id: UUID to check
Returns:
True if exists
"""
return
await
self
id
async
def
get_by_ids
self, ids: Sequence[UUID]
Sequence
"""
Get multiple records by IDs.
Args:
ids: UUIDs to fetch
Returns:
Found model instances
"""
return
await
self
Usage: Creating Entity Services
# src/app/items/service.pyfrom 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 with business logic."""def__init__(self, repository: ItemRepository):
super().__init__(repository)
# Type hint for IDE supportself._repository: ItemRepository = repository
asyncdefcreate(self, obj_in: ItemCreate) -> Item:
"""
Create item with duplicate name check.
Raises:
ConflictError: If item with same name exists
"""# Business logic: check for duplicate name
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.
Useful when you need to ensure the item exists.
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 | None:
"""
Update item with duplicate name check.
Raises:
ConflictError: If new name conflicts with existing item
"""# Business logic: check name uniqueness on updateif 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,
)
returnawaitsuper().update(id, obj_in, exclude_unset)
Service Patterns
1. Get or Raise Pattern
asyncdefget_by_id_or_raise(self, id: UUID) -> ModelType:
"""Get record or raise NotFoundError."""
instance = awaitself.get_by_id(id)
ifnot instance:
raise NotFoundError(resource=self._model_name, id=id)
return instance
2. Validation Before Create
asyncdefcreate(self, obj_in: CreateSchemaType) -> ModelType:
"""Create with pre-validation."""awaitself._validate_create(obj_in)
returnawaitsuper().create(obj_in)
asyncdef_validate_create(self, obj_in: CreateSchemaType) -> None:
"""Override in subclass to add validation logic."""pass
By default, each repository method commits its transaction. For complex operations spanning multiple writes, consider transaction management:
# In repository, add a method that doesn't commit:asyncdefcreate_no_commit(self, obj_in: CreateSchemaType) -> ModelType:
data = obj_in.model_dump()
instance = self._model(**data)
self._session.add(instance)
awaitself._session.flush() # Get ID without commitreturn instance
# In service:asyncdefcreate_order_with_items(self, order: OrderCreate) -> Order:
asyncwithself._session.begin(): # Transaction
order = awaitself._order_repo.create_no_commit(order)
for item in order.items:
awaitself._order_item_repo.create_no_commit(item)
# Commits on exitreturn order
Key Principles
Services contain business logic - validation, orchestration, rules
Services delegate data access - never write SQL in services
Services are stateless - no instance state between calls
Services raise domain exceptions - NotFoundError, ConflictError, etc.
Services are testable - mock the repository for unit tests
