How to Add a Custom Storage Backend Implementation in NekoImageGallery

To add a custom storage backend implementation in nekoimagegallery, create a class inheriting from BaseStorage in app/Services/storage/base.py, implement all abstract async methods, and register it in StorageService at 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) 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) 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) 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) for on-disk storage
• S3Storage (app/Services/storage/s3_compatible_storage.py) for S3-compatible APIs
• DisabledStorage (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, and inherit from BaseStorage[FileMetaDataT].

Required Abstract Methods

Your class must implement these async methods defined in BaseStorage (lines 24-140 of 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:

• 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:

# 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 to import your class and add a match case:

# 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:

# 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:

pytest -k storage

Or programmatically verify the wiring:

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 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 and adding a match case for your storage mode.
• Optionally extend StorageMode in 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: 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 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. 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 for patterns on bridging sync SDKs to async interfaces.

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