Paginación
Built-in en BaseService.list() + BaseRepository.list_paginated(). El controller solo declara los Query params.
Endpoint paginado
@router.get("/", response_model=BasePaginationResponse[ThingResponseSchema])
async def list_things(
self,
page: int = Query(1, ge=1),
count: int = Query(10, ge=1, le=100),
search: Optional[str] = Query(None),
):
return await self.list()
self.list() lee los params del request, llama service.list(), calcula total_pages, retorna BasePaginationResponse.
Request
Response shape
{
"data": [
{ "id": "...", "name": "Thing 1" },
{ "id": "...", "name": "Thing 2" }
],
"pagination": {
"page": 2,
"count": 20,
"total": 153,
"total_pages": 8
},
"message": "Operación exitosa",
"status": "success"
}
Search
?search=foo aplica ILIKE %foo% (case-insensitive) sobre los campos listados, combinados con OR.
Soporta paths anidados:
list_paginated agrega los JOINs necesarios automáticamente.
Ordenamiento — order_by
GET /api/v1/things/?order_by=created_at # ASC
GET /api/v1/things/?order_by=-created_at # DESC
GET /api/v1/things/?order_by=category__name # JOIN + ASC
GET /api/v1/things/?order_by=-owner__email # JOIN + DESC
Default por servicio:
Filtros desde query string
Cualquier query param que NO sea page/count/search/order_by se trata como filtro:
Equivale a WHERE status='active' AND category_id='abc-123'.
Filtros con relaciones:
Filtros forzados desde el service
Para multi-tenant, sobreescribe get_filters:
def get_filters(self, filters: dict | None = None) -> dict:
filters = filters or {}
user = getattr(self.request.state, "user", None)
if user and not user.is_platform_admin:
filters["company_id"] = user.company_id # forzado, no overrideable por client
return filters
El cliente NO puede saltarse estos filtros desde query string.
OR en lugar de AND
Default es AND entre filtros. Para OR, llama explícito desde controller:
Eager loading — joins
Por acción:
def get_kwargs_query(self) -> dict:
if self.action in ("list_things", "retrieve"):
return {"joins": ["category", "tags"]}
return {}
O explícito:
joins aplica selectinload (relación 1:N) o joinedload (relación N:1) automáticamente según el tipo.
Enriquecer los items de la página — post_process_list (0.5.0)
Para agregar un campo derivado o un contador a cada fila DESPUÉS de paginar, sin
reescribir la paginación ni overridear list():
class ThingService(BaseService):
async def post_process_list(self, items):
ids = [t.id for t in items]
counts = await self.child_repo.count_by_parent_ids(ids)
for t in items:
t.child_count = counts.get(t.id, 0)
return items
Corre sobre los items de la página actual; no cambia total ni filtra (para
filtrar usá get_filters/build_list_queryset).
El motor de paginación NO se reescribe
El count/skip/offset/limit/$facet vive en el repo base
(list_paginated SQL, paginate/paginate_pipeline Beanie) y no se copia.
Toda personalización baja por un hook. Mapa completo caso→hook:
Puntos de Extensión (Listados).