Database Error Handling Patterns in NekoImageGallery: Domain-Driven Exception Design

NekoImageGallery implements a defensive, domain-driven error handling strategy that wraps Qdrant vector database operations in custom exception classes, validates IDs before operations, and translates domain errors into HTTP responses at the API boundary.

NekoImageGallery is an open-source image gallery application built with FastAPI and the Qdrant vector database. Understanding its error handling patterns for database operations reveals a consistent defensive programming approach that separates domain logic from transport concerns. The architecture uses custom exception hierarchies, proactive validation, and automatic retry mechanisms to ensure robust data integrity across all vector DB interactions.

Domain-Specific Exception Hierarchy

The foundation of error handling in NekoImageGallery rests on domain-specific exception classes that carry context-rich metadata. Rather than raising generic exceptions, the code defines precise failure types that higher layers can catch and handle appropriately.

PointNotFoundError for Missing Data

When vector points cannot be located, the application raises PointNotFoundError. This exception is defined directly in the database service layer at [app/Services/vector_db_context.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/vector_db_context.py#L20-L24) (lines 20‑24), making it immediately available to all repository operations.

# Defined in VectorDbContext

class PointNotFoundError(Exception):
    def __init__(self, point_id: str):
        self.point_id = point_id
        super().__init__(f"Point {point_id} not found")

PointDuplicateError for Data Conflicts

For duplicate insertion attempts, the system uses PointDuplicateError, declared in [app/Models/errors.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Models/errors.py). This separation of error types allows calling code to distinguish between "data does not exist" and "data already exists" scenarios, enabling precise HTTP status code mapping (404 vs. 400).

Validate-Before-Operate Pattern

Every destructive or retrieval operation begins with proactive ID validation. The VectorDbContext.validate_ids method (lines 90‑101 in [app/Services/vector_db_context.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/vector_db_context.py#L90-L101)) queries the database to confirm existence before allowing operations to proceed.

async def validate_ids(self, ids: list[str]) -> list[str]:
    """Returns list of existing IDs, raises PointNotFoundError if any missing."""
    existing = await self._client.retrieve(
        collection_name=self.collection_name,
        ids=ids,
        with_payload=False,
        with_vectors=False
    )
    if len(existing) != len(ids):
        missing = set(ids) - {p.id for p in existing}
        raise PointNotFoundError(next(iter(missing)))
    return [p.id for p in existing]

This pattern ensures that retrieve, delete, and update operations fail fast with deterministic error paths rather than silent no-ops.

Explicit Failure Paths in CRUD Operations

NekoImageGallery enforces fail-fast semantics across all data access methods, eliminating ambiguous return states like None or empty lists when data should exist.

Raising on Missing Points

The retrieve_by_id method (lines 55‑68) and retrieve_by_ids method (lines 70‑88) in [app/Services/vector_db_context.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/vector_db_context.py) explicitly raise PointNotFoundError when requested vectors are absent:

async def retrieve_by_id(self, point_id: str) -> ImageData:
    result = await self._client.retrieve(
        collection_name=self.collection_name,
        ids=[point_id]
    )
    if not result:
        raise PointNotFoundError(point_id)  # Line 66

    return ImageData.from_point(result[0])

Preventing Duplicate Inserts

Before any insertion, services validate uniqueness. The IndexService._is_point_duplicate method (lines 36‑41 in [app/Services/index_service.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/index_service.py)) checks existence, raising PointDuplicateError at lines 44‑46:

if await self._is_point_duplicate([image_data]):
    raise PointDuplicateError(
        "The uploaded points are contained in the database!", 
        image_data.id
    )

Similarly, UploadService.assign_image_id (lines 90‑94 in [app/Services/upload_service.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/upload_service.py)) performs the same guard before accepting uploads.

API Layer Translation

The transport layer remains decoupled from database specifics through exception translation. FastAPI controllers catch domain exceptions and convert them to appropriate HTTP status codes.

In [app/Controllers/images.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Controllers/images.py#L27-L33) (lines 27‑33), missing images become 404 responses:

try:
    img = await services.db_context.retrieve_by_id(str(image_id))
except PointNotFoundError:
    raise HTTPException(404, "Cannot find the image with the given ID.")

The admin controller in [app/Controllers/admin.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Controllers/admin.py#L122-L124) (lines 122‑124) handles duplicates with 400 Bad Request:

except PointDuplicateError as e:
    raise HTTPException(400, f"Duplicate point: {e.point_id}")

This centralized translation keeps HTTP concerns out of the database layer while ensuring clients receive semantically correct status codes.

Resilience Patterns

Beyond functional error handling, NekoImageGallery implements infrastructure resilience patterns for transient failures and application lifecycle management.

Automatic Retry for Transient Errors

When the Qdrant client operates in server mode, the VectorDbContext.__init__ method (lines 37‑38) wraps the client with retry_async to handle transient grpc.AioRpcError and httpx.HTTPError:

from app.util.retry_deco_async import retry_async, wrap_object

self._client = AsyncQdrantClient(host=host, port=port)
wrap_object(self._client, retry_async((AioRpcError, HTTPError)))

The decorator in [app/util/retry_deco_async.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/util/retry_deco_async.py) implements exponential backoff without polluting business logic with retry loops.

Graceful Shutdown Handling

The LifespanService base class in [app/Services/lifespan_service.py](https://github.com/hv0905/nekoimagegallery/blob/master/app/Services/lifespan_service.py) defines on_exit hooks used by VectorDbContext, UploadService, and IndexService. These hooks ensure pending database operations complete before the FastAPI application terminates, preventing data loss during deployment or restart events.

Summary

NekoImageGallery demonstrates production-grade error handling patterns for vector database operations through:

• Domain-specific exceptions (PointNotFoundError, PointDuplicateError) that carry contextual IDs
• Validate-before-operate workflows via validate_ids that fail fast on missing data
• Explicit failure paths in retrieval and insertion methods rather than silent null returns
• API boundary translation that converts domain exceptions to semantic HTTP status codes (404, 400)
• Transparent retry logic for transient network errors using decorators
• Lifecycle management ensuring graceful shutdowns with LifespanService hooks

Frequently Asked Questions

What exception classes does NekoImageGallery use for database errors?

NekoImageGallery defines two primary domain exceptions: PointNotFoundError (declared in app/Services/vector_db_context.py lines 20‑24) for missing vector points, and PointDuplicateError (declared in app/Models/errors.py) for duplicate insertion attempts. Both classes carry the offending point ID as an attribute, enabling precise error reporting.

How does NekoImageGallery prevent duplicate vector inserts?

The application prevents duplicates through pre-insertion validation. The IndexService._is_point_duplicate method checks Qdrant for existing IDs before insertion, raising PointDuplicateError if found (lines 44‑46). The UploadService performs similar validation in assign_image_id (lines 90‑94), ensuring no overwrites occur during concurrent uploads.

Where are database exceptions converted to HTTP responses?

Exception translation occurs in the FastAPI controller layer. The images_router catches PointNotFoundError and raises HTTPException(404) (lines 27‑33), while the admin controller catches PointDuplicateError and raises HTTPException(400) (lines 122‑124). This separation keeps HTTP status codes out of the database service layer.

How does the application handle transient Qdrant connection failures?

NekoImageGallery wraps the AsyncQdrantClient with a retry decorator during initialization (lines 37‑38 of vector_db_context.py). The retry_async utility from app/util/retry_deco_async.py automatically retries operations that raise grpc.AioRpcError or httpx.HTTPError, implementing exponential backoff without modifying business logic code.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}