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

> Discover NekoImageGallery's domain-driven error handling for database operations. Learn how custom exceptions and ID validation ensure robust error management.

- Repository: [EdgeNeko/nekoimagegallery](https://github.com/hv0905/nekoimagegallery)
- Tags: best-practices
- Published: 2026-03-03

---

**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/main/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.

```python

# 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/main/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/main/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.

```python
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/main/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:

```python
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/main/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:

```python
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/main/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/main/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:

```python
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/main/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:

```python
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`:

```python
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/main/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/main/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`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/vector_db_context.py) lines 20‑24) for missing vector points, and **`PointDuplicateError`** (declared in [`app/Models/errors.py`](https://github.com/hv0905/nekoimagegallery/blob/main/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`](https://github.com/hv0905/nekoimagegallery/blob/main/vector_db_context.py)). The `retry_async` utility from [`app/util/retry_deco_async.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/util/retry_deco_async.py) automatically retries operations that raise `grpc.AioRpcError` or `httpx.HTTPError`, implementing exponential backoff without modifying business logic code.