Services

class MyService(BaseCrudService[
    ModelType,          # SQLAlchemy model     → e.g. User
    CreateSchemaType,   # Pydantic create       → e.g. UserCreate
    UpdateSchemaType,   # Pydantic update       → e.g. UserUpdate
    ResponseSchemaType  # Pydantic response     → e.g. UserResponse
]):
    pass

# ── Create ────────────────────────────────────
product = service.create(ProductCreate(name="Widget", price=9.99))
products = service.create_many([ProductCreate(...), ProductCreate(...)])

# ── Read ──────────────────────────────────────
product  = service.find(1)                      # None if missing
product  = service.find_or_fail(1)              # raises NotFoundException
products = service.get_all()
products = service.get_all(limit=50)
products = service.filter(status='active', price__lt=100)
product  = service.filter_one(sku='ABC123')
exists   = service.exists(sku='ABC123')         # bool
count    = service.count(status='active')       # int

# ── Paginate ──────────────────────────────────
products, meta = service.paginate(
    page=1, per_page=20,
    _order_by='-created_at',
    status='active'
)

# ── Update ────────────────────────────────────
updated = service.update(1, ProductUpdate(price=12.99))
count   = service.update_many(filters={'status':'pending'}, data={'status':'active'})

# ── Delete ────────────────────────────────────
service.delete(1)                  # soft delete if SoftDeleteMixin
service.delete(1, force=True)      # always hard delete
count = service.delete_many({'status': 'inactive'})

# ── Transaction control ───────────────────────
service.commit()
service.rollback()
service.flush()

# All three work identically
service.create(ProductCreate(name="Widget", price=9.99))
service.create({"name": "Widget", "price": 9.99})

# Partial updates — only set fields are sent to the database
update = ProductUpdate(price=12.99)  # name is not set
service.update(1, update)            # only price column updated

from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from fastkit_core.database import get_async_db, AsyncRepository

async def get_service(session: AsyncSession = Depends(get_async_db)):
    return ProductService(AsyncRepository(Product, session))

@app.post("/products", status_code=201)
async def create_product(data: ProductCreate, svc = Depends(get_service)):
    product = await svc.create(data)
    return success_response(data=product.model_dump())

@app.get("/products")
async def list_products(page: int = 1, svc = Depends(get_service)):
    products, meta = await svc.paginate(page=page, per_page=20)
    return {'items': [p.model_dump() for p in products], 'pagination': meta}

class UserResponse(BaseModel):
    id:    int
    email: str
    name:  str
    # password is NOT included — secure by design
    model_config = {"from_attributes": True}

class UserService(BaseCrudService[User, UserCreate, UserUpdate, UserResponse]):
    def __init__(self, repository: Repository):
        super().__init__(repository, response_schema=UserResponse)

# Every method returns UserResponse — not the raw User model
user: UserResponse = service.create(user_data)
user: UserResponse = service.find(1)
users: list[UserResponse] = service.get_all()

class UserService(BaseCrudService[User, UserCreate, UserUpdate, UserResponse]):

    def before_create(self, data: dict) -> dict:
        data['password']   = hash_password(data['password'])
        data['created_by'] = get_current_user_id()
        return data  # must return modified data

    def after_create(self, instance: User) -> None:
        send_welcome_email(instance.email)
        create_user_profile(instance.id)

    def before_update(self, id: int, data: dict) -> dict:
        data['updated_by'] = get_current_user_id()
        return data

    def after_update(self, instance: User) -> None:
        cache.delete(f'user:{instance.id}')
        notify_watchers(instance.id)

    def before_delete(self, id: int) -> None:
        user = self.find_or_fail(id)
        if user.has_active_subscriptions():
            raise ValueError("Cannot delete user with active subscriptions")

    def after_delete(self, id: int) -> None:
        delete_user_files(id)
        revoke_user_tokens(id)

class ProductService(BaseCrudService[Product, ProductCreate, ProductUpdate, ProductResponse]):

    def validate_create(self, data: ProductCreate) -> None:
        # Uniqueness
        if self.exists(sku=data.sku):
            raise ValueError("Product with this SKU already exists")

        # Business rules
        if data.price <= 0:
            raise ValueError("Price must be greater than zero")

        if data.stock < 0:
            raise ValueError("Stock cannot be negative")

    def validate_update(self, id: int, data: ProductUpdate) -> None:
        product = self.find_or_fail(id)

        # Permission check
        if not can_edit_product(product):
            raise ValueError("You don't have permission to edit this product")

        # Business rule
        if data.price and data.price < product.cost:
            raise ValueError("Price cannot be lower than cost")

class ProductService(AsyncBaseCrudService[Product, ProductCreate, ProductUpdate, ProductResponse]):

    async def validate_create(self, data: ProductCreate) -> None:
        if await self.exists(sku=data.sku):
            raise ValueError("SKU already exists")

        # Call external services in validation
        if not await external_inventory_check(data.sku):
            raise ValueError("SKU not found in inventory system")

from fastkit_core.services import AsyncBaseCrudService, SlugServiceMixin

class ArticleService(
    SlugServiceMixin,
    AsyncBaseCrudService[Article, ArticleCreate, ArticleUpdate, ArticleResponse]
):
    def __init__(self, repository: AsyncRepository):
        super().__init__(repository, response_schema=ArticleResponse)

    async def before_create(self, data: dict) -> dict:
        # Generates unique slug, appends -1, -2 etc. if needed
        data['slug'] = await self.async_generate_slug(data['title'])
        return data

    async def before_update(self, id: int, data: dict) -> dict:
        if 'title' in data:
            # exclude_id ensures current record isn't seen as a conflict
            data['slug'] = await self.async_generate_slug(
                data['title'],
                exclude_id=id
            )
        return data

# Single field
users = service.filter(_order_by='-created_at')       # newest first
users = service.filter(_order_by='name')              # A-Z

# Multiple fields
products = service.filter(
    status='active',
    _order_by=['-rating', 'price']                    # highest rated, cheapest first
)

# With pagination
users, meta = service.paginate(
    page=1, per_page=20,
    _order_by=['-created_at', 'name']
)

# Async
products = await service.filter(
    category='electronics',
    _order_by=['-published_at', 'title']
)

# First page
users, next_cursor = service.cursor_paginate(per_page=20)

# Next page
users, next_cursor = service.cursor_paginate(
    per_page=20,
    cursor=next_cursor,
    status='active'             # filters work as normal
)

# Async — identical API with await
users, next_cursor = await service.cursor_paginate(
    per_page=20,
    cursor_field='created_at',
    direction='desc',
    cursor=next_cursor
)

class UserService(AsyncBaseCrudService[User, UserCreate, UserUpdate, UserResponse]):
    def __init__(self, repository):
        super().__init__(repository, response_schema=UserResponse)

# Returns list[UserResponse], not list[User]
users: list[UserResponse]
users, next_cursor = await service.cursor_paginate(per_page=20)

from fastapi import Query
from fastkit_core.http import success_response

@app.get('/users')
async def list_users(
    cursor: str | None = Query(default=None),
    per_page: int = Query(default=20, ge=1, le=100),
    svc: UserService = Depends(get_service)
):
    users, next_cursor = await svc.cursor_paginate(
        per_page=per_page,
        cursor=cursor
    )
    return {
        'data':        [u.model_dump() for u in users],
        'next_cursor': next_cursor
    }