Flagship Use Case: Order Platform¶

"""orders/app.py — every pico lib doing its natural job."""

import asyncio

import httpx
from pydantic import BaseModel
from sqlalchemy import String

from pico_caching import cacheable
from pico_celery import task
from pico_fastapi import controller, get, post
from pico_httpx import http_client
from pico_httpx import post as http_post
from pico_ioc import component
from pico_kafka import kafka_consumer, kafka_producer, produce
from pico_pydantic import validate
from pico_rabbitmq import consumer, publish, publisher
from pico_resilience import circuit_breaker, retryable
from pico_scheduling import scheduled
from pico_sqlalchemy import (
    AppBase,
    Mapped,
    SessionManager,
    get_session,
    mapped_column,
    query,
    repository,
    transactional,
)

# ── Domain (pico-sqlalchemy) ─────────────────────────────────────


class Order(AppBase):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str] = mapped_column(String(50))
    amount_cents: Mapped[int] = mapped_column()
    status: Mapped[str] = mapped_column(String(20))


class OrderRequest(BaseModel):
    sku: str
    quantity: int = 1


@repository(entity=Order)
class OrderRepository:
    def __init__(self, session_manager: SessionManager):
        self.sm = session_manager

    async def save(self, sku: str, amount_cents: int) -> Order:
        session = get_session(self.sm)
        order = Order(sku=sku, amount_cents=amount_cents, status="paid")
        session.add(order)
        await session.flush()
        return order

    @query(expr="status = :status")
    async def find_by_status(self, status: str) -> list[Order]: ...


@component
class SchemaSetup:
    """DatabaseConfigurer for dev; production sets database.migrations_path
    and lets Alembic run on startup instead."""

    priority = 0

    def configure_database(self, engine) -> None:
        async def _create():
            async with engine.begin() as conn:
                await conn.run_sync(AppBase.metadata.create_all)
            # asyncpg binds pooled connections to the loop that created
            # them; drop the DDL loop's pool so request loops start fresh.
            await engine.dispose()

        asyncio.run(_create())


# ── External payments (pico-httpx + pico-resilience) ─────────────


@http_client(name="payments")   # base_url from http.clients.payments.base_url
class PaymentsApi:
    @retryable(max_attempts=3, backoff_seconds=0.05, retry_on=(httpx.HTTPError,))
    @circuit_breaker(failure_threshold=5, reset_timeout_seconds=30)
    @http_post("/charges")
    async def charge(self, json: dict) -> dict: ...


# ── Catalog with distributed cache (pico-caching + pico-data-redis) ──


@component
class Catalog:
    @cacheable(ttl_seconds=300)   # Redis-backed the moment pico-data-redis is installed
    def price_cents(self, sku: str) -> int:
        return {"ESPRESSO": 250, "LATTE": 390}.get(sku, 999)


# ── Events (pico-rabbitmq) and analytics stream (pico-kafka) ─────


@publisher
class OrderEvents:
    @publish(exchange="events", routing_key="orders.created")
    def order_created(self, message): ...


@kafka_producer
class OrderAnalytics:
    @produce("orders")
    def order_placed(self, message): ...


@component
class AnalyticsProjection:
    @kafka_consumer("orders", group_id="analytics")
    @validate
    def on_order(self, message: dict): ...


@component
class StockProjection:
    @consumer("stock-projection", exchange="events", routing_key="orders.*")
    async def on_order_event(self, message: dict): ...


# ── Invoicing worker (pico-celery) ───────────────────────────────


@component(scope="prototype")
class InvoiceTasks:
    @task(name="invoices.generate")
    async def generate(self, order_id: int) -> dict:
        return {"order_id": order_id, "pdf": f"invoice-{order_id}.pdf"}


# ── Business service ─────────────────────────────────────────────


@component
class OrderService:
    def __init__(self, repo: OrderRepository, payments: PaymentsApi, catalog: Catalog,
                 events: OrderEvents, analytics: OrderAnalytics):
        self._repo = repo
        self._payments = payments
        self._catalog = catalog
        self._events = events
        self._analytics = analytics

    @transactional
    async def place(self, request: OrderRequest) -> dict:
        amount = self._catalog.price_cents(request.sku) * request.quantity
        await self._payments.charge(json={"amount_cents": amount})
        order = await self._repo.save(request.sku, amount)
        payload = {"id": order.id, "sku": order.sku, "amount_cents": amount}
        self._events.order_created(payload)
        self._analytics.order_placed(payload)
        return {**payload, "status": order.status}


# ── HTTP API (pico-fastapi) ──────────────────────────────────────


@controller(prefix="/api/v1/orders", tags=["Orders"])
class OrderController:
    def __init__(self, service: OrderService, repo: OrderRepository):
        self._service = service
        self._repo = repo

    @post("")
    async def place_order(self, request: OrderRequest):
        return await self._service.place(request)

    @get("/paid")
    async def paid_orders(self):
        return [{"id": o.id, "sku": o.sku} for o in await self._repo.find_by_status("paid")]


# ── Housekeeping (pico-scheduling) ───────────────────────────────


@component
class Reconciliation:
    def __init__(self, repo: OrderRepository):
        self._repo = repo

    @scheduled(cron="0 3 * * *")
    async def reconcile(self):
        await self._repo.find_by_status("paid")

from fastapi import FastAPI
from pico_boot import init


def create_app() -> FastAPI:
    container = init(modules=["orders"])   # installed pico plugins auto-discovered
    return container.get(FastAPI)

def test_place_order(make_container):
    container = make_container("orders", config={...})
    ...