Skip to content

Storage Adapters

Overview

A source keeps its files in a storage adapter. Two are built in:

Name Storage Settings
local (default) Directory on the server root
s3 AWS S3 or an S3-compatible service s3 block (AWS S3); needs the s3 extra

Anything else (a database, another object store, a remote API) plugs in as a custom adapter registered under a name.

How it works

action → Source (root confinement, name checks, thumbnails)
       → FileStorage (path normalization, error wrapping)
       → StorageAdapter (your backend)

Adapters see normalized relative paths: / separators, no leading slash, no . or .., the empty string for the root. Everything above them (access checks, confinement to the source, safe names, thumbnail folders) is the connector's job, and an adapter only stores bytes.

For non-local adapters root is a virtual prefix (default /); the real location, e.g. the key prefix of a bucket, belongs to the adapter's own options.

Selecting an adapter

{
  "sources": {
    "local": {
      "title": "Local",
      "root": "/var/www/files",
      "baseurl": "https://example.com/files/"
    },
    "media": {
      "title": "Media",
      "baseurl": "https://cdn.example.com/media/",
      "storageAdapter": "s3",
      "s3": {"bucket": "my-bucket", "prefix": "media"}
    },
    "scratch": {
      "title": "Scratch",
      "baseurl": "https://example.com/scratch/",
      "storageAdapter": "memory"
    }
  }
}

A name that is not registered fails the request that uses the source with 400 Unknown storage adapter "...", listing the registered names.

The interface

jcpy.StorageAdapter is a Protocol: any class with these methods works, no base class needed.

Method Contract
async write(path, contents: bytes) Create or replace a file, creating parent directories
async read(path) -> bytes Whole file; raise FileWasNotFoundError(path) when missing
async delete_file(path) Delete a file; missing is not an error
async create_directory(path) Create a directory with its parents
async delete_directory(path) Delete a directory recursively; missing is not an error
async stat(path) -> StatEntry Size and modification time (epoch ms); raise when the path does not exist
list(path, *, deep) -> AsyncIterator[StatEntry] Entries of a directory (every level with deep=True); fill in size and time when the backend returns them, otherwise the connector calls stat for each entry
async file_exists(path) -> bool True for a file, not for a directory
async directory_exists(path) -> bool True for a directory (the root included)
async copy_file(source, destination) Copy a file, creating the destination's parents
async move_file(source, destination) Move a file or a directory

StatEntry(path, is_file, size=None, last_modified_ms=None) is in jcpy.storage, next to StorageError and FileWasNotFoundError. Exceptions raised by an adapter are wrapped with the operation (Unable to write the file. Reason: ...) and answered with 500 or 400 depending on the action.

Blocking libraries should run in threads (anyio.to_thread.run_sync), as the built-in adapters do, to keep the event loop free.

The built-in adapters also guarantee:

  • local writes and copies atomically (a temporary file next to the target, then a rename), so readers never see a partial file; listings skip entries that are neither files nor directories (symlinks, FIFOs) and log a warning.
  • s3 fails a folder deletion when any object could not be deleted instead of reporting success.

Registering an adapter

from jcpy import SourceConfig, create_app, register_storage_adapter


def my_adapter(source: SourceConfig) -> StorageAdapter:
    return MyAdapter(source.name)


register_storage_adapter("mine", my_adapter)
app = create_app("config.json")

The factory gets the source settings and returns an adapter. Adapters of all sources are built once, on the first request that needs sources, and kept for the life of the instance (for dynamic sources: until the tenant leaves the cache). Register before the first request; the registry is global to the process.

Settings for your adapter can be read from the source: unknown keys are rejected, so either keep them in your code (keyed by source.name) or reuse an existing field such as root.

Example: in-memory adapter

A complete adapter, used by examples/custom_storage.py and covered by the test suite:

class MemoryStorageAdapter:
    """Keep files in a dict; paths are relative and use ``/``."""

    def __init__(self) -> None:
        self.files: dict[str, tuple[bytes, float]] = {}
        self.directories: set[str] = {""}

    def _add_parents(self, path: str) -> None:
        parts = path.split("/")
        for index in range(1, len(parts)):
            self.directories.add("/".join(parts[:index]))

    async def write(self, path: str, contents: bytes) -> None:
        """Create or replace a file."""
        self._add_parents(path)
        self.files[path] = (contents, time.time() * 1000)

    async def read(self, path: str) -> bytes:
        """Read a whole file."""
        if path not in self.files:
            raise FileWasNotFoundError(path)
        return self.files[path][0]

    async def delete_file(self, path: str) -> None:
        """Delete a file; a missing file is not an error."""
        self.files.pop(path, None)

    async def create_directory(self, path: str) -> None:
        """Create a directory with its parents."""
        self._add_parents(path)
        self.directories.add(path)

    async def delete_directory(self, path: str) -> None:
        """Delete a directory recursively."""
        prefix = f"{path}/"
        self.files = {
            key: value
            for key, value in self.files.items()
            if not key.startswith(prefix)
        }
        self.directories = {
            item
            for item in self.directories
            if item != path and not item.startswith(prefix)
        }

    async def stat(self, path: str) -> StatEntry:
        """Read size and modification time."""
        if path in self.files:
            contents, modified = self.files[path]
            return StatEntry(path, True, len(contents), modified)
        if path in self.directories:
            return StatEntry(path, False)
        msg = f"Unable to get stat. Reason: {path} does not exist"
        raise StorageError(msg)

    async def list(self, path: str, *, deep: bool) -> AsyncIterator[StatEntry]:
        """List a directory (and its sub-directories when ``deep``)."""
        prefix = f"{path}/" if path else ""
        entries = [
            StatEntry(item, is_file=False)
            for item in sorted(self.directories)
            if item and item.startswith(prefix)
        ] + [
            StatEntry(item, is_file=True)
            for item in sorted(self.files)
            if item.startswith(prefix)
        ]
        for entry in entries:
            if deep or "/" not in entry.path.removeprefix(prefix):
                yield entry

    async def file_exists(self, path: str) -> bool:
        """Tell whether a file exists."""
        return path in self.files

    async def directory_exists(self, path: str) -> bool:
        """Tell whether a directory exists."""
        return path in self.directories

    async def copy_file(self, source: str, destination: str) -> None:
        """Copy a file."""
        await self.write(destination, await self.read(source))

    async def move_file(self, source: str, destination: str) -> None:
        """Move a file or a whole directory."""
        if source in self.files:
            self._add_parents(destination)
            self.files[destination] = self.files.pop(source)
            return
        prefix = f"{source}/"
        for key in [key for key in self.files if key.startswith(prefix)]:
            await self.write(
                f"{destination}/{key.removeprefix(prefix)}",
                self.files.pop(key)[0],
            )
        for item in sorted(self.directories):
            if item == source or item.startswith(prefix):
                self.directories.discard(item)
                await self.create_directory(
                    destination + item.removeprefix(source)
                )


_STORES: dict[str, MemoryStorageAdapter] = {}


def memory_adapter(source: SourceConfig) -> MemoryStorageAdapter:
    """Give every source its own store, kept between requests.

    Args:
        source: Settings of the source using the adapter.

    Returns:
        Store of the source.
    """
    return _STORES.setdefault(source.name, MemoryStorageAdapter())


def build_app() -> FastAPI:
    """Build the application.

    Returns:
        Connector application with in-memory sources.
    """
    register_storage_adapter("memory", memory_adapter)
    return create_app(CONFIG)

Testing an adapter

Run the connector over it and exercise the actions, as the example's test does:

from httpx import ASGITransport, AsyncClient


async def test_round_trip() -> None:
    app = build_app()
    transport = ASGITransport(app=app)
    async with AsyncClient(
        transport=transport, base_url="http://test"
    ) as http:
        await http.post("/fileUpload", files={"files[0]": ("a.txt", b"hi")})
        await http.get("/folderCreate", params={"name": "docs"})
        moved = await http.get(
            "/fileMove", params={"from": "a.txt", "path": "docs"}
        )
        listing = await http.get("/files", params={"path": "docs"})
        removed = await http.get("/folderRemove", params={"name": "docs"})

    assert moved.status_code == 200
    assert [
        f["file"] for f in listing.json()["data"]["sources"][0]["files"]
    ] == ["a.txt"]
    assert removed.status_code == 200

Pitfalls

  1. The root is the empty string. directory_exists("") must be True and list("") must list the top level.
  2. Parents appear implicitly. write("a/b/c.txt", ...) must make a and a/b exist, and list must report them.
  3. deep listings return every level, with paths relative to the root, not to the listed directory.
  4. move_file moves directories too (fileMove and folderMove use it).
  5. Missing is not an error for delete_file and delete_directory.