aerospike-py-fastapi

name

aerospike-py-fastapi

description

MUST USE for building FastAPI/REST API applications with Aerospike database (aerospike-py Rust/PyO3 async client). Covers AsyncClient lifespan management, FastAPI Depends injection, aerospike_py exception-to-HTTP-status mapping (RecordNotFound→404, RecordExistsError→409, BackpressureError→503, AerospikeTimeoutError→504), POLICY_EXISTS_CREATE_ONLY/UPDATE_ONLY for CRUD semantics, NamedTuple attribute access (record.bins not tuple unpacking), client.ping() readiness probe, batch_read returning dict, batch_write with in_doubt retry signal, and global AerospikeError handler. Triggers on: FastAPI + Aerospike, REST API + aerospike-py, CRUD API with Aerospike, client.ping() health probe, batch_read/batch_write endpoint, web server/HTTP service backed by Aerospike NoSQL, uvicorn + Aerospike.

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():
    :
         client.put((NS, SET, pk), body,
                         policy={: aerospike_py.POLICY_EXISTS_UPDATE_ONLY})
         {: pk}
     aerospike_py.RecordNotFound:
         JSONResponse(status_code=, content={: })


  ():
    :
         client.remove((NS, SET, pk))
     aerospike_py.RecordNotFound:
         JSONResponse(status_code=, content={: })

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})
         item  items
    ]
    result =  client.batch_write(records)         
    
     ():
         (br.key.user_key)  br.key     
    in_doubt = [_user_key(br)  br  result.batch_records  br.result !=   br.in_doubt]
    failed   = [_user_key(br)  br  result.batch_records  br.result !=    br.in_doubt]
     in_doubt:
        
         JSONResponse(status_code=, content={: in_doubt, : 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