aerospike-py-fastapi

1. App Structure (Lifespan + DI)

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, Depends
from aerospike_py import AsyncClient
import aerospike_py

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncClient({
        "hosts": [("127.0.0.1", 3000)],
        "max_concurrent_operations": 64,      # backpressure
        "operation_queue_timeout_ms": 5000,
    })
    aerospike_py.init_tracing()                # OpenTelemetry (optional)
    # aerospike_py.start_metrics_server(port=9464)  # built-in /metrics; pick this OR §6 endpoint, not both
    await client.connect()
    app.state.aerospike = client
    yield
    await client.close()
    # aerospike_py.stop_metrics_server()
    aerospike_py.shutdown_tracing()

app = FastAPI(lifespan=lifespan)

def get_client(request: Request) -> AsyncClient:
    return request.app.state.aerospike

2. CRUD Endpoints

from fastapi.responses import JSONResponse

NS, SET = "app", "records"

@app.post("/records/{pk}", status_code=201)
async def create(pk: str, body: dict, client: AsyncClient = Depends(get_client)):
    try:
        await client.put((NS, SET, pk), body,
                         policy={"exists": aerospike_py.POLICY_EXISTS_CREATE_ONLY})
        return {"key": pk}
    except aerospike_py.RecordExistsError:
        return JSONResponse(status_code=409, content={"error": "already exists"})

@app.get("/records/{pk}")
async def read(pk: str, client: AsyncClient = Depends(get_client)):
    try:
        record = await client.get((NS, SET, pk))
        return record.bins                     # NamedTuple attribute access
    except aerospike_py.RecordNotFound:
        return JSONResponse(status_code=404, content={"error": "not found"})

@app.put("/records/{pk}")
async def update(pk: str, body: dict, client: AsyncClient = Depends(get_client)):
    try:
        await client.put((NS, SET, pk), body,
                         policy={"exists": aerospike_py.POLICY_EXISTS_UPDATE_ONLY})
        return {"key": pk}
    except aerospike_py.RecordNotFound:
        return JSONResponse(status_code=404, content={"error": "not found"})

@app.delete("/records/{pk}", status_code=204)
async def delete(pk: str, client: AsyncClient = Depends(get_client)):
    try:
        await client.remove((NS, SET, pk))
    except aerospike_py.RecordNotFound:
        return JSONResponse(status_code=404, content={"error": "not found"})

2b. Batch Endpoints

batch_read returns dict[UserKey, AerospikeRecord] -- iterate with .items(). Missing keys are absent from the dict.

from pydantic import BaseModel

class BatchReadReq(BaseModel):
    keys: list[str]

@app.post("/records:batchRead")
async def batch_read(req: BatchReadReq, client: AsyncClient = Depends(get_client)):
    keys = [(NS, SET, k) for k in req.keys]
    records = await client.batch_read(keys)            # dict[str, dict[str, Any]]
    return {"found": records, "missing": [k for k in req.keys if k not in records]}

class BatchWriteItem(BaseModel):
    key: str
    bins: dict
    ttl: int | None = None

@app.post("/records:batchWrite")
async def batch_write(items: list[BatchWriteItem], client: AsyncClient = Depends(get_client)):
    records = [
        ((NS, SET, item.key), item.bins) if item.ttl is None
        else ((NS, SET, item.key), item.bins, {"ttl": item.ttl})
        for item in items
    ]
    result = await client.batch_write(records)         # BatchWriteResult
    # br.key may be None for some failure paths -- guard before dereferencing
    def _user_key(br):
        return str(br.key.user_key) if br.key is not None else None
    in_doubt = [_user_key(br) for br in result.batch_records if br.result != 0 and br.in_doubt]
    failed   = [_user_key(br) for br in result.batch_records if br.result != 0 and not br.in_doubt]
    if in_doubt:
        # Some writes may have applied -- caller should reconcile via batch_read, not blind retry
        return JSONResponse(status_code=503, content={"in_doubt": in_doubt, "failed": failed})
    return {"failed": failed}

3. Global Error Handler

@app.exception_handler(aerospike_py.BackpressureError)
async def backpressure_handler(request, exc):
    return JSONResponse(status_code=503, content={"error": "server busy, retry later"})

@app.exception_handler(aerospike_py.AerospikeError)
async def aerospike_error_handler(request, exc):
    return JSONResponse(status_code=500, content={"error": str(exc)})

4. Exception → HTTP Status Mapping

Exception	HTTP Status	Meaning
RecordNotFound	404	Record does not exist
RecordExistsError	409	Record already exists (CREATE_ONLY)
RecordGenerationError	409	Optimistic lock conflict
BackpressureError	503	Too many concurrent operations
AerospikeTimeoutError	504	Operation timed out (canonical name; TimeoutError alias removed)
AerospikeIndexError	400/500	Secondary index error (400 if user supplied bad query, else 500)
AerospikeError	500	Catch-all server error

5. Health Check

@app.get("/health/ready")
async def ready(client: AsyncClient = Depends(get_client)):
    # ping() does an info("build") round-trip; never raises -- returns False on failure.
    return {"status": "ok"} if await client.ping() else JSONResponse(503, {"status": "unhealthy"})

@app.get("/health/live")
async def live():
    return {"status": "ok"}  # liveness should NOT depend on Aerospike (avoids restart loops on transient blip)

6. Metrics Endpoint

from fastapi import Response

@app.get("/metrics")
async def metrics():
    return Response(
        content=aerospike_py.get_metrics(),
        media_type="text/plain; version=0.0.4",
    )

Or use the built-in server: aerospike_py.start_metrics_server(port=9464) in lifespan.

7. Key Patterns

• Lifespan: Create AsyncClient in lifespan, store on app.state, close on shutdown
• DI: Use Depends(get_client) for every endpoint — never create client per-request
• Exceptions on module: aerospike_py.RecordNotFound (NOT aerospike_py.exception.RecordNotFound)
• NamedTuple returns: record.bins, record.meta.gen (NOT tuple unpacking)
• batch_read: returns dict[UserKey, AerospikeRecord] — iterate with .items(), missing keys are absent
• batch_write: inspect BatchRecord.in_doubt before retrying non-idempotent writes
• Health probe: await client.ping() for readiness; trivial 200 OK for liveness
• Backpressure: Set max_concurrent_operations to prevent connection pool exhaustion
• Exception name: AerospikeTimeoutError (legacy TimeoutError alias removed)

Detail: ../aerospike-py-api/reference/client-config.md | ../aerospike-py-api/reference/admin.md | ../aerospike-py-api/reference/health.md | ../aerospike-py-api/reference/write.md