# How to Add a Custom Storage Backend Implementation in NekoImageGallery

> Learn how to add a custom storage backend implementation in NekoImageGallery. Inherit from BaseStorage, implement methods, and register your class in StorageService for seamless integration.

- Repository: [EdgeNeko/nekoimagegallery](https://github.com/hv0905/nekoimagegallery)
- Tags: how-to-guide
- Published: 2026-03-03

---

**To add a custom storage backend implementation in nekoimagegallery, create a class inheriting from `BaseStorage` in [`app/Services/storage/base.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/base.py), implement all abstract async methods, and register it in `StorageService` at [`app/Services/storage/__init__.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/__init__.py) by importing the class and adding a Match case for your storage mode.**

NekoImageGallery uses a pluggable storage layer abstracted by `BaseStorage` to handle image files across different backends. Whether you need to integrate a cloud object store, a database blob column, or an in-memory cache, you can extend the storage system without modifying core application logic. This guide walks you through implementing the interface and wiring your custom backend into the service container.

## Understand the Storage Architecture

The storage system consists of three primary components working together to provide a unified async API for file operations.

**BaseStorage** ([`app/Services/storage/base.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/base.py)) defines the abstract contract that every backend must fulfill. It specifies coroutines for existence checks, URL generation, file transfers, and metadata management.

**StorageService** ([`app/Services/storage/__init__.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/__init__.py)) acts as a `LifespanService` that instantiates the active storage implementation based on `config.storage.method`. It imports concrete classes and uses Python Match statements to initialize the correct backend.

**StorageMode** ([`app/config.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/config.py)) is the configuration enum that determines which backend loads at runtime, with built-in options for `LOCAL`, `S3`, and `DISABLED`.

Reference implementations provide working patterns:
- `LocalStorage` ([`app/Services/storage/local_storage.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/local_storage.py)) for on-disk storage
- `S3Storage` ([`app/Services/storage/s3_compatible_storage.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/s3_compatible_storage.py)) for S3-compatible APIs
- `DisabledStorage` ([`app/Services/storage/disabled_storage.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/disabled_storage.py)) for no-op behavior

## Implement the BaseStorage Interface

Create a new module under `app/Services/storage/`, such as [`custom_storage.py`](https://github.com/hv0905/nekoimagegallery/blob/main/custom_storage.py), and inherit from `BaseStorage[FileMetaDataT]`.

### Required Abstract Methods

Your class must implement these async methods defined in `BaseStorage` (lines 24-140 of [`base.py`](https://github.com/hv0905/nekoimagegallery/blob/main/base.py)):

- `on_load()`: Lifecycle hook for initialization (connections, buckets, etc.)
- `is_exist(remote_file) -> bool`: Check if a path exists
- `size(remote_file) -> int`: Return file size in bytes, raising `RemoteFileNotFoundError` if missing
- `url(remote_file) -> str`: Generate public URL for the file
- `presign_url(remote_file, expire_second=3600) -> str`: Generate temporary signed URL
- `fetch(remote_file) -> bytes`: Retrieve file content, raising `RemoteFileNotFoundError` on missing keys
- `upload(local_file, remote_file) -> None`: Store file content, raising `RemoteFileExistsError` on conflicts
- `copy(old_remote_file, new_remote_file) -> None`: Duplicate files with existence validation
- `move(old_remote_file, new_remote_file) -> None`: Relocate files (typically copy then delete)
- `delete(remote_file) -> None`: Remove files with `RemoteFileNotFoundError` if absent
- `list_files(path, pattern, batch_max_files, valid_extensions) -> AsyncGenerator`: Yield batches of paths matching filters
- `update_metadata(local_file_metadata, remote_file_metadata) -> None`: Sync metadata between local and remote

You must also define these class attributes in `__init__`:
- `static_dir`: Base path for static assets
- `thumbnails_dir`: Path for generated thumbnails
- `deleted_dir`: Archive path for soft deletes
- `file_metadata`: Type hint for metadata handling

### Exception Handling Contract

All backends must raise specific exceptions from [`app/Services/storage/exception.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/exception.py):
- `RemoteFileNotFoundError` when files are missing
- `RemoteFilePermissionError` for access violations
- `RemoteFileExistsError` for duplicate uploads

This ensures `StorageService` callers handle errors consistently regardless of the underlying backend.

### Example Implementation

Here is a minimal in-memory storage implementation demonstrating the required signatures:

```python

# app/Services/storage/in_memory_storage.py

import asyncio
from typing import AsyncGenerator, Optional
from loguru import logger
import aiofiles
import fnmatch

from app.Services.storage.base import BaseStorage, FileMetaDataT, RemoteFilePathType, LocalFilePathType
from app.Services.storage.exception import RemoteFileNotFoundError, RemoteFileExistsError

class InMemoryStorage(BaseStorage[FileMetaDataT: None]):
    def __init__(self):
        super().__init__()
        self._store: dict[RemoteFilePathType, bytes] = {}
        self.static_dir = "/in_memory"
        self.thumbnails_dir = "/in_memory/thumbnails"
        self.deleted_dir = "/in_memory/_deleted"
        self.file_metadata = None

    async def on_load(self):
        logger.info("InMemoryStorage initialized")

    async def is_exist(self, remote_file: RemoteFilePathType) -> bool:
        return remote_file in self._store

    async def size(self, remote_file: RemoteFilePathType) -> int:
        if remote_file not in self._store:
            raise RemoteFileNotFoundError
        return len(self._store[remote_file])

    async def url(self, remote_file: RemoteFilePathType) -> str:
        return f"memory://{remote_file}"

    async def presign_url(self, remote_file: RemoteFilePathType, expire_second: int = 3600) -> str:
        return await self.url(remote_file)

    async def fetch(self, remote_file: RemoteFilePathType) -> bytes:
        try:
            return self._store[remote_file]
        except KeyError:
            raise RemoteFileNotFoundError

    async def upload(self, local_file: LocalFilePathType, remote_file: RemoteFilePathType) -> None:
        if isinstance(local_file, bytes):
            data = local_file
        else:
            async with aiofiles.open(local_file, "rb") as f:
                data = await f.read()
        if remote_file in self._store:
            raise RemoteFileExistsError
        self._store[remote_file] = data
        logger.success(f"Uploaded {remote_file}")

    async def copy(self, old_remote_file: RemoteFilePathType, new_remote_file: RemoteFilePathType) -> None:
        if old_remote_file not in self._store:
            raise RemoteFileNotFoundError
        if new_remote_file in self._store:
            raise RemoteFileExistsError
        self._store[new_remote_file] = self._store[old_remote_file]

    async def move(self, old_remote_file: RemoteFilePathType, new_remote_file: RemoteFilePathType) -> None:
        await self.copy(old_remote_file, new_remote_file)
        await self.delete(old_remote_file)

    async def delete(self, remote_file: RemoteFilePathType) -> None:
        try:
            del self._store[remote_file]
        except KeyError:
            raise RemoteFileNotFoundError

    async def list_files(
        self,
        path: RemoteFilePathType,
        pattern: Optional[str] = "*",
        batch_max_files: Optional[int] = None,
        valid_extensions: Optional[set[str]] = None,
    ) -> AsyncGenerator[list[RemoteFilePathType], None]:
        matched = [p for p in self._store.keys() if p.startswith(path) and fnmatch.fnmatch(p, pattern)]
        batch_size = batch_max_files or len(matched)
        for i in range(0, len(matched), batch_size):
            yield matched[i:i + batch_size]

    async def update_metadata(self, local_file_metadata, remote_file_metadata) -> None:
        raise NotImplementedError

```

Replace the dictionary logic with your SDK calls while maintaining the same method signatures.

## Register Your Custom Backend

After implementing the class, wire it into the application container.

### Import in StorageService

Edit [`app/Services/storage/__init__.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/__init__.py) to import your class and add a match case:

```python

# app/Services/storage/__init__.py

from app.Services.storage.in_memory_storage import InMemoryStorage  # New import

class StorageService(LifespanService):
    def __init__(self):
        self.active_storage = None
        match config.storage.method:
            case StorageMode.LOCAL:
                self.active_storage = LocalStorage()
            case StorageMode.S3:
                self.active_storage = S3Storage()
            case StorageMode.DISABLED:
                self.active_storage = DisabledStorage()
            case StorageMode.MEMORY:  # New case

                self.active_storage = InMemoryStorage()
            case _:
                raise NotImplementedError(f"Storage mode {config.storage.method} not supported")

```

### Configure StorageMode (Optional)

To make the backend selectable via configuration, extend the enum in [`app/config.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/config.py):

```python

# app/config.py

class StorageMode(str, Enum):
    LOCAL = 'local'
    S3 = 's3'
    DISABLED = 'disabled'
    MEMORY = 'memory'  # New option

```

Set the environment variable `APP_STORAGE__METHOD=memory` to activate your backend at runtime.

## Verify Your Implementation

Test your custom storage backend implementation using the existing test suite and a manual sanity check:

```bash
pytest -k storage

```

Or programmatically verify the wiring:

```python
from app.Services.storage import StorageService
from app.config import config, StorageMode

config.storage.method = StorageMode.MEMORY
service = StorageService()
await service.on_load()
assert isinstance(service.active_storage, InMemoryStorage)
await service.active_storage.upload(b"test data", "test.txt")
assert await service.active_storage.is_exist("test.txt")

```

## Summary

- **Inherit from `BaseStorage`** in [`app/Services/storage/base.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/base.py) and implement all abstract async methods including `upload`, `fetch`, `delete`, and `list_files`.
- **Raise specific exceptions** (`RemoteFileNotFoundError`, `RemoteFileExistsError`, `RemoteFilePermissionError`) to maintain contract consistency with other backends.
- **Register in `StorageService`** by importing your class in [`app/Services/storage/__init__.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/__init__.py) and adding a match case for your storage mode.
- **Optionally extend `StorageMode`** in [`app/config.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/config.py) to expose the backend through configuration files or environment variables.

## Frequently Asked Questions

### What methods are mandatory when adding a custom storage backend implementation?

You must implement all abstract methods defined in `BaseStorage` from [`app/Services/storage/base.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/base.py): `on_load`, `is_exist`, `size`, `url`, `presign_url`, `fetch`, `upload`, `copy`, `move`, `delete`, `list_files`, and `update_metadata`. All methods must be async and accept the parameter types specified in the base class (e.g., `RemoteFilePathType`, `LocalFilePathType`).

### Do I need to modify the StorageMode enum to use my custom backend?

No. While adding a `StorageMode` value in [`app/config.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/config.py) allows runtime selection via configuration files, you can instantiate your backend directly in `StorageService` for hardcoded deployments or testing scenarios. The enum extension is only necessary if operators need to switch backends without code changes.

### How should error handling work in a custom storage backend implementation?

Your backend must raise the library-specific exceptions defined in [`app/Services/storage/exception.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/exception.py). Use `RemoteFileNotFoundError` when files are missing, `RemoteFileExistsError` for duplicate uploads, and `RemoteFilePermissionError` for access denials. This ensures that upstream services in nekoimagegallery handle errors predictably regardless of whether you are using local disk, S3, or a custom provider.

### Can I use synchronous libraries like boto3 in my storage implementation?

Yes, but you must wrap synchronous calls in `asyncio.to_thread()` or similar executor patterns because `BaseStorage` defines all methods as async. The `StorageService` expects awaitable coroutines, and blocking the event loop will degrade application performance. Reference `S3Storage` in [`app/Services/storage/s3_compatible_storage.py`](https://github.com/hv0905/nekoimagegallery/blob/main/app/Services/storage/s3_compatible_storage.py) for patterns on bridging sync SDKs to async interfaces.