Tutorial: Full CRUD API¶

Prerequisites

Complete the Getting Started guide first. This tutorial assumes you are familiar with @component, @configured, and pico_boot.init().

You can also scaffold this project instantly with pico-initializer — select fastapi, sqlalchemy, pydantic, and auth, then enable the Products CRUD example.

This tutorial walks through building a production-ready Products API using the pico ecosystem. The example is a versioned REST API (/api/v1/) that manages a product catalog — creation, listing, search, pagination, partial updates, soft and hard deletion — secured with JWT and role-based access control.

The goal is to show how the different pico packages fit together as a cohesive stack, and how each one takes responsibility for a distinct layer:

pico-boot wires everything together with zero boilerplate
pico-client-auth protects all routes by default; you opt individual ones out
pico-pydantic guards the service contract regardless of how the service is called
pico-sqlalchemy handles transactions and queries declaratively
pico-fastapi keeps controllers thin — pure HTTP mapping, no business logic

By the end you will have a working API where an invalid JWT, a missing role, a malformed request body, and a domain constraint violation each fail fast at their own layer, with the right HTTP status and without leaking details across layers.

Stack¶

Library	Role
`pico-boot`	Bootstrap, plugin auto-discovery, config loading
`pico-ioc`	IoC container, constructor injection
`pico-fastapi`	HTTP controllers with `@controller`
`pico-sqlalchemy`	Repositories, transactions, declarative queries
`pico-pydantic`	AOP validation in the service layer with `@validate`
`pico-client-auth`	JWT authentication, RBAC, `SecurityContext`
`pydantic`	HTTP schemas (input/output DTOs)

Installation¶

pip install \
  pico-boot \
  pico-fastapi \
  pico-sqlalchemy \
  pico-pydantic \
  pico-client-auth \
  aiosqlite \
  pydantic \
  uvicorn

Replace aiosqlite with asyncpg for PostgreSQL.

Database model¶

The example uses a single products table. active enables soft-deletion — deactivated products disappear from all queries without being physically removed.

Layer architecture¶

Each pico package owns exactly one layer. The IoC container resolves constructor dependencies automatically — no manual wiring between layers.

HTTP layer — pico-fastapi · pico-client-auth

ProductController @controller · /api/v1/products JWT middleware validates Bearer token RoleResolver extracts roles from claims

Service layer — pico-ioc · pico-pydantic ProductService @validate · @transactional SecurityContext request-scoped ContextVar ValidationInterceptor AOP — pico-pydantic

Data layer — pico-sqlalchemy

ProductRepository @repository · @query (declarative) SessionManager async engine · implicit commit

Infrastructure Database SQLite · PostgreSQL JWKS endpoint RSA public keys · TTL cache

constructor injection reads / wraps

Auth server separation¶

The Products API and the Auth server run as independent processes. The API never stores credentials — it validates JWTs using only the public keys from the Auth server's JWKS endpoint, cached locally with a configurable TTL.

CLIENT Browser / App 1. POST /login 2. Receives JWT 3. Calls API with Bearer

AUTH SERVER separate process Login endpoint JWT issuer JWKS endpoint GET /jwks.json RSA key pair private signs · public verifies

PRODUCTS API separate process JWT middleware JWKS cache TTL 300s SecurityContext ProductController Database

POST /login JWT token

Bearer <jwt>

GET /jwks.json cached · TTL 300s

CREDENTIALS NEVER REACH THE API only the public key (JWKS) is shared

Project structure¶

├── application.yaml         # all configuration — one file for the whole stack
├── main.py                  # 5 lines — full bootstrap
└── myapp/
    ├── __init__.py
    ├── database.py          # DatabaseConfigurer — auto-creates tables on startup
    ├── models.py            # SQLAlchemy entity — tables only, no logic
    ├── schemas.py           # Pydantic schemas — HTTP DTOs + domain contract
    ├── repositories.py      # database access — declarative queries, no session boilerplate
    ├── services.py          # business logic — transactions, validation, domain rules
    ├── controllers.py       # HTTP routes — thin mapping layer, no business logic
    └── auth.py              # custom RoleResolver (optional — only if your IdP is non-standard)

The pico-ioc scanner is recursive: declaring "myapp" in init() is enough to discover everything in any subpackage. No explicit imports between layers are needed for wiring — the IoC container resolves constructor dependencies by type.

Configuration — `application.yaml`¶

All configuration for all plugins in a single file:

# Database — read by pico-sqlalchemy
database:
  url: sqlite+aiosqlite:///./products.db
  echo: false

# FastAPI — read by pico-fastapi
fastapi:
  title: Products API
  version: 1.0.0

# JWT auth — read by pico-client-auth
auth_client:
  issuer: https://auth.example.com   # required — fails at startup if missing
  audience: my-api                   # required — fails at startup if missing
  jwks_ttl_seconds: 300
  accepted_algorithms:
    - RS256

Each package reads its own section. There is no configuration code anywhere in the application components.

Environment variables override any YAML value:

DATABASE__URL=postgresql+asyncpg://user:pass@db/prod \
AUTH_CLIENT__ISSUER=https://auth.prod.com \
uvicorn main:app

Bootstrap — `main.py`¶

from pico_boot import init
from pico_ioc import configuration, YamlTreeSource, EnvSource
from fastapi import FastAPI

container = init(
    modules=["myapp"],
    config=configuration(YamlTreeSource("application.yaml"), EnvSource()),
)

app = container.get(FastAPI)

That's it. pico-boot discovers all installed plugins via entry points (pico-fastapi, pico-sqlalchemy, pico-pydantic, pico-client-auth) and initializes them in the correct order. There is no explicit mention of those packages anywhere in the code.

uvicorn main:app --reload
# Swagger at http://localhost:8000/docs

Request flow¶

POST /api/v1/products/  Authorization: Bearer <jwt>
         │
         ▼
pico-client-auth (automatic middleware)
  ├─ validates JWT signature against JWKS
  ├─ checks iss, aud, exp
  ├─ resolves roles via RoleResolver
  └─ populates SecurityContext          →  401 if token invalid/missing
         │
         ▼
@requires_role("product-manager")      →  403 if insufficient permissions
         │
         ▼
ProductController.create_product(body: ProductCreate)
  └─ FastAPI validates the body         →  422 if HTTP data invalid
         │
         ▼
ProductService.create(data: ProductData)
  └─ pico-pydantic @validate            →  ValidationFailedError if contract broken
         │
         ▼
@transactional opens async session
         │
         ▼
ProductRepository.save(product)
  └─ flush() + automatic commit
         │
         ▼
HTTP 201  {"id": 1, "name": "Laptop Pro", ...}

Three independent validation layers, each in the right place: auth before touching any code, HTTP data at the HTTP boundary, domain contract inside the service.

1. Entity — `myapp/models.py`¶

from sqlalchemy import Integer, String, Float, Boolean
from pico_sqlalchemy import AppBase, Mapped, mapped_column


class Product(AppBase):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
    price: Mapped[float] = mapped_column(Float, nullable=False)
    stock: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)

AppBase is the declarative base from pico-sqlalchemy. Entity subclasses are discovered automatically, but table creation requires a DatabaseConfigurer component.

1b. Table setup — `myapp/database.py`¶

import asyncio

from pico_ioc import component
from pico_sqlalchemy import DatabaseConfigurer, AppBase


@component
class SchemaSetup(DatabaseConfigurer):
    def __init__(self, base: AppBase):
        self.base = base

    @property
    def priority(self) -> int:
        return 0

    def configure_database(self, engine) -> None:
        async def _create():
            async with engine.begin() as conn:
                await conn.run_sync(self.base.metadata.create_all)
        asyncio.run(_create())

This component is auto-discovered by the IoC scanner. On startup, pico-sqlalchemy calls configure_database() and the tables are created.

2. Schemas — `myapp/schemas.py`¶

Two types of schemas with distinct purposes:

HTTP schemas (ProductCreate, ProductUpdate, ProductResponse): define the REST API contract. FastAPI uses them to validate and serialize at the HTTP boundary.
Domain schema (ProductData): defines the internal service contract. pico-pydantic enforces it on every call to the service, regardless of origin — HTTP, Celery, tests, scripts, anything.

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Self


# --- HTTP schemas ---

class ProductCreate(BaseModel):
    name: str = Field(min_length=2, max_length=100)
    description: str | None = Field(default=None, max_length=500)
    price: float = Field(gt=0)
    stock: int = Field(ge=0, default=0)

    @field_validator("name")
    @classmethod
    def name_no_spaces_only(cls, v: str) -> str:
        if v.strip() == "":
            raise ValueError("Name cannot be whitespace only")
        return v.strip()

    @field_validator("price")
    @classmethod
    def price_max_two_decimals(cls, v: float) -> float:
        if round(v, 2) != v:
            raise ValueError("Price cannot have more than 2 decimal places")
        return v


class ProductUpdate(BaseModel):
    name: str | None = Field(default=None, min_length=2, max_length=100)
    description: str | None = Field(default=None, max_length=500)
    price: float | None = Field(default=None, gt=0)
    stock: int | None = Field(default=None, ge=0)
    active: bool | None = Field(default=None)

    @model_validator(mode="after")
    def at_least_one_field(self) -> Self:
        if all(v is None for v in self.model_dump().values()):
            raise ValueError("At least one field must be provided for update")
        return self


class ProductResponse(BaseModel):
    model_config = {"from_attributes": True}  # builds from ORM without manual mapping

    id: int
    name: str
    description: str | None
    price: float
    stock: int
    active: bool


class ProductListResponse(BaseModel):
    items: list[ProductResponse]
    total: int
    page: int
    page_size: int


# --- Domain schema (for pico-pydantic) ---

class ProductData(BaseModel):
    """
    Internal service contract. pico-pydantic enforces this on every
    call to the service, regardless of where the call comes from.
    """
    name: str = Field(min_length=2, max_length=100)
    description: str | None = Field(default=None, max_length=500)
    price: float = Field(gt=0)
    stock: int = Field(ge=0, default=0)

    @field_validator("name")
    @classmethod
    def strip_name(cls, v: str) -> str:
        return v.strip()

ProductData is independent of FastAPI. The fact that the HTTP schema and the domain schema look similar here is coincidental — in real systems they often diverge.

3. Repository — `myapp/repositories.py`¶

from pico_sqlalchemy import repository, query, SessionManager, get_session
from .models import Product


@repository(entity=Product)
class ProductRepository:
    def __init__(self, manager: SessionManager):
        self.manager = manager

    # Write methods — implicitly Read-Write by @repository

    async def save(self, product: Product) -> Product:
        session = get_session(self.manager)
        session.add(product)
        await session.flush()     # INSERT/UPDATE without commit
        await session.refresh(product)
        return product

    async def delete(self, product: Product) -> None:
        session = get_session(self.manager)
        await session.delete(product)

    # Declarative queries — implicitly Read-Only by @query.
    # The "..." body is ignored; pico-sqlalchemy generates and
    # executes the SQL from the expr.

    @query(expr="id = :id", unique=True)
    async def find_by_id(self, id: int) -> Product | None: ...

    @query(expr="active = true")
    async def find_all_active(self) -> list[Product]: ...

    @query(expr="active = true", paged=True)
    async def find_active_paged(self, page) -> ...: ...

    @query(expr="name like :pattern")
    async def search_by_name(self, pattern: str) -> list[Product]: ...

flush() sends the operation to the engine without committing. The commit happens automatically when the @transactional method in the service returns.

@query generates SQL equivalent to:

SELECT * FROM products WHERE id = :id
SELECT * FROM products WHERE active = true
SELECT * FROM products WHERE name like :pattern

With paged=True it accepts a PageRequest and returns Page[Product] with items (the list) and total (the unpaginated count).

4. Service — `myapp/services.py`¶

from pico_ioc import component
from pico_sqlalchemy import transactional, PageRequest, Page
from pico_pydantic import validate
from pico_client_auth import SecurityContext
from fastapi import HTTPException, status

from .models import Product
from .repositories import ProductRepository
from .schemas import ProductData


@component
class ProductService:
    def __init__(self, repo: ProductRepository):
        self.repo = repo

    @validate          # pico-pydantic: validates ProductData before executing
    @transactional     # pico-sqlalchemy: opens the async session
    async def create(self, data: ProductData) -> Product:
        # If we get here, data is a valid ProductData.
        # pico-pydantic accepts both a dict and a ProductData — converts automatically.
        product = Product(
            name=data.name,
            description=data.description,
            price=data.price,
            stock=data.stock,
        )
        return await self.repo.save(product)

    async def get_by_id(self, product_id: int) -> Product:
        product = await self.repo.find_by_id(id=product_id)
        if product is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"Product {product_id} not found",
            )
        return product

    async def list_active(self) -> list[Product]:
        return await self.repo.find_all_active()

    async def list_paged(self, page: int, page_size: int) -> Page:
        return await self.repo.find_active_paged(
            page=PageRequest(page=page, size=page_size)
        )

    async def search(self, name: str) -> list[Product]:
        return await self.repo.search_by_name(pattern=f"%{name}%")

    @validate
    @transactional
    async def update(self, product_id: int, data: ProductData) -> Product:
        product = await self.get_by_id(product_id)
        for field, value in data.model_dump(exclude_none=True).items():
            setattr(product, field, value)
        return await self.repo.save(product)

    @transactional
    async def deactivate(self, product_id: int) -> Product:
        product = await self.get_by_id(product_id)
        product.active = False
        return await self.repo.save(product)

    @transactional
    async def delete(self, product_id: int) -> None:
        product = await self.get_by_id(product_id)
        await self.repo.delete(product)

@validate always goes before @transactional: if the data is invalid, no transaction is opened unnecessarily.

SecurityContext.get() is available anywhere in the call stack during a request — without passing it as a parameter. Useful for audit logs, per-user filtering, etc.

5. Controller — `myapp/controllers.py`¶

With pico-client-auth installed, all routes are protected by default. Only those decorated with @allow_anonymous are public.

from pico_fastapi import controller, get, post, put, patch, delete
from pico_client_auth import allow_anonymous, requires_role, SecurityContext
from fastapi import Query, status

from .services import ProductService
from .schemas import (
    ProductCreate, ProductUpdate,
    ProductResponse, ProductListResponse,
)


@controller(prefix="/api/v1/products", tags=["Products"])
class ProductController:
    def __init__(self, service: ProductService):
        self.service = service

    # Authenticated — any user with a valid token
    @get("/", response_model=list[ProductResponse])
    async def list_products(self):
        products = await self.service.list_active()
        return [ProductResponse.model_validate(p) for p in products]

    @get("/paged", response_model=ProductListResponse)
    async def list_paged(
        self,
        page: int = Query(default=1, ge=1),
        page_size: int = Query(default=10, ge=1, le=100),
    ):
        result = await self.service.list_paged(page=page, page_size=page_size)
        return ProductListResponse(
            items=[ProductResponse.model_validate(p) for p in result.items],
            total=result.total,
            page=page,
            page_size=page_size,
        )

    @get("/search", response_model=list[ProductResponse])
    async def search(self, q: str = Query(min_length=1)):
        products = await self.service.search(name=q)
        return [ProductResponse.model_validate(p) for p in products]

    @get("/{product_id}", response_model=ProductResponse)
    async def get_product(self, product_id: int):
        product = await self.service.get_by_id(product_id)
        return ProductResponse.model_validate(product)

    # Role-based access control
    @post("/", response_model=ProductResponse, status_code=status.HTTP_201_CREATED)
    @requires_role("product-manager")
    async def create_product(self, body: ProductCreate):
        # model_dump() → dict; pico-pydantic converts it to ProductData in the service
        product = await self.service.create(body.model_dump())
        return ProductResponse.model_validate(product)

    @put("/{product_id}", response_model=ProductResponse)
    @requires_role("product-manager")
    async def update_product(self, product_id: int, body: ProductUpdate):
        product = await self.service.update(
            product_id, body.model_dump(exclude_none=True)
        )
        return ProductResponse.model_validate(product)

    @patch("/{product_id}/deactivate", response_model=ProductResponse)
    @requires_role("product-manager")
    async def deactivate_product(self, product_id: int):
        product = await self.service.deactivate(product_id)
        return ProductResponse.model_validate(product)

    @delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
    @requires_role("admin")
    async def delete_product(self, product_id: int):
        await self.service.delete(product_id)

The health endpoint lives in its own controller at /api/v1/health — keeping it separate avoids route collisions with path parameters like /{product_id}:

@controller(prefix="/api/v1", tags=["Health"])
class HealthController:
    @get("/health")
    @allow_anonymous
    async def health(self):
        return {"status": "ok"}

The role check happens before the method executes. If the token is valid but the user lacks the required role, a 403 is returned without reaching the controller body.

6. Custom RoleResolver — `myapp/auth.py`¶

Optional. Only needed if your IdP puts roles in a non-standard claim. Example for Keycloak:

from pico_ioc import component
from pico_client_auth import RoleResolver, TokenClaims


@component
class KeycloakRoleResolver:
    async def resolve(self, claims: TokenClaims, raw_claims: dict) -> list[str]:
        realm_roles = raw_claims.get("realm_access", {}).get("roles", [])
        client_roles = (
            raw_claims
            .get("resource_access", {})
            .get("my-api", {})
            .get("roles", [])
        )
        return list(set(realm_roles + client_roles))

Declaring @component is all it takes — the framework detects it and uses it instead of the default RoleResolver. No additional registration needed.

Validation summary¶

Layer	What is validated	Tool	Error
Auth	JWT signature, `iss`, `aud`, `exp`	pico-client-auth	401
Auth	Required role / group	pico-client-auth	403
HTTP	Body and query params	Pydantic + FastAPI	422
Domain	Service contract	pico-pydantic `@validate`	`ValidationFailedError`

Endpoints¶

Method	Path	Access
GET	`/api/v1/health`	public
GET	`/api/v1/products/`	authenticated
GET	`/api/v1/products/paged`	authenticated
GET	`/api/v1/products/search?q=`	authenticated
GET	`/api/v1/products/{id}`	authenticated
POST	`/api/v1/products/`	`product-manager`
PUT	`/api/v1/products/{id}`	`product-manager`
PATCH	`/api/v1/products/{id}/deactivate`	`product-manager`
DELETE	`/api/v1/products/{id}`	`admin`

Tutorial: Full CRUD API¶

Stack¶

Installation¶

Database model¶

Layer architecture¶

Auth server separation¶

Project structure¶

Configuration — application.yaml¶

Bootstrap — main.py¶

Request flow¶

1. Entity — myapp/models.py¶

1b. Table setup — myapp/database.py¶

2. Schemas — myapp/schemas.py¶

3. Repository — myapp/repositories.py¶

4. Service — myapp/services.py¶

5. Controller — myapp/controllers.py¶

6. Custom RoleResolver — myapp/auth.py¶