98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import List, Optional
|
|
|
|
import aiosqlite
|
|
from fastapi import APIRouter, Depends, Query, Response
|
|
|
|
import persons
|
|
from common import Page
|
|
from main import get_db
|
|
|
|
router = APIRouter(prefix="/persons", tags=["persons"])
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
operation_id="listPersons",
|
|
response_model=Page[persons.Person],
|
|
summary="List persons (paginated)",
|
|
responses={
|
|
200: {
|
|
"description": "A page of persons",
|
|
"content": {
|
|
"application/json": {
|
|
"example": {
|
|
"items": [
|
|
{
|
|
"id": 1,
|
|
"name": "Ada Lovelace"
|
|
}
|
|
],
|
|
"nextCursor": "2",
|
|
"prevCursor": "0",
|
|
"total": 1
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
)
|
|
async def list_persons(
|
|
q: Optional[str] = Query(
|
|
default=None,
|
|
description="Optional case-insensitive name filter (applied after page read; may be pushed to SQL in the future).",
|
|
),
|
|
cursor: Optional[str] = Query(
|
|
default=None,
|
|
description="Opaque cursor for pagination. Pass the value returned in nextCursor to fetch the next page.",
|
|
),
|
|
limit: int = Query(
|
|
50,
|
|
ge=1,
|
|
le=200,
|
|
description="Maximum number of items to return (1-200).",
|
|
),
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
) -> Page[persons.Person]:
|
|
# v1: DB-backed pagination
|
|
last_id = None
|
|
if cursor:
|
|
try:
|
|
last_id = int(cursor)
|
|
except ValueError:
|
|
last_id = None
|
|
|
|
fetch_limit = limit + 1
|
|
paged: List[persons.Person] = []
|
|
if q:
|
|
async for p in persons.search_by_name_paged(conn, q, last_id, fetch_limit):
|
|
paged.append(p)
|
|
else:
|
|
async for p in persons.get_all_paged(conn, last_id, fetch_limit):
|
|
paged.append(p)
|
|
|
|
has_more = len(paged) > limit
|
|
items = paged[:limit]
|
|
next_cursor = str(items[-1].id) if has_more and items else None
|
|
# Compute prevCursor via DB helper
|
|
prev_cursor: Optional[str] = None
|
|
if items:
|
|
first_id = items[0].id
|
|
prev_cursor = await persons.compute_prev_cursor(conn, first_id, limit, q)
|
|
total = await (persons.count_by_name(conn, q) if q else persons.count_all(conn))
|
|
return Page(items=items, nextCursor=next_cursor, prevCursor=prev_cursor, total=total)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
operation_id="createPerson",
|
|
summary="Create a person",
|
|
response_model=persons.Person,
|
|
)
|
|
async def create_person(
|
|
person: persons.Person, response: Response, conn: aiosqlite.Connection = Depends(get_db)
|
|
) -> persons.Person:
|
|
await persons.insert_person(conn, person)
|
|
response.headers["Location"] = f"/api/v1/persons/{person.id}"
|
|
return person
|