Storage

The storage mixin and the local filesystem storage with its mount system.

Storage capability: filesystem storage as a mixin over the base server (§4, D6).

StorageMixin is a capability composed over BaseServer (class S(..., StorageMixin, BaseServer)). Unlike AuthMixin/SessionMixin it arms NO middleware — storage is survival infrastructure, not a request-chain concern (§4: PUBLIC = base + storage). Its cooperative __init__ peels storage= and storage_key= and forwards everything else down the D16 chain.

storage= shapes the LocalStorage the server owns:

• None → a fresh LocalStorage() (default base_dir), only the predefined site/secure mounts;

• a LocalStorage instance → adopted as-is;

• a dict {mount_code: {"path": ..., "encrypted": ...}} → a fresh LocalStorage with one add_mount per entry.

storage_key= installs at-rest encryption key material (set_encryption_keys — comma-separated Fernet keys, first encrypts, all decrypt). Configured but resolved empty is an explicit boot error (old-repo semantics: no silent degradation); omitted, the encrypted mounts stay dormant until keys are set.

A composition WITHOUT the mixin has NO storage attribute at all.

class genro_asgi.storage_mixin.StorageMixin(**kwargs)

Bases: object

Storage capability mixin, composed over the base server. Arms no middleware.

Constructor kwargs peeled here: storage — None (a fresh LocalStorage), a LocalStorage instance (adopted), or a {code: {path, encrypted}} dict (mounts materialized); storage_key — the encryption key material (configured-but-empty raises).

Parameters:

kwargs (Any)

property storage: LocalStorage

The filesystem storage this server owns (mounts + at-rest encryption).

LocalStorage — filesystem-only storage with a genro-storage compatible API.

This module provides a minimal storage implementation that uses the same API as genro-storage, but only supports the local filesystem. When genro-storage becomes available, simply change the import:

# Before (local only) from genro_asgi.storage import LocalStorage

# After (full genro-storage) from genro_storage import StorageManager as LocalStorage

The API is entirely SYNCHRONOUS (SPECIFICATION.md D22, core 1b ratified): the node read/write methods do blocking I/O and return values directly. Async callers on the server dispatch paths wrap them in server.run_sync() (the Macro 1 dispatch protocol); there is no dual sync/async mode.

Each mount is a security boundary: a node path with an absolute segment, a .. component, or resolving outside the mount base raises ValueError on access — no read/write/delete may ever touch a location outside its mount root (D5 — explicit error, never a silent wrong-file access).

At-rest encryption

A mount can be encrypted: reads/writes on its nodes are transparently decrypted/encrypted, so store contracts and clients above stay crypto-unaware.

storage.set_encryption_keys(“<key>[,<key2>,…]”) # install key material node = storage.node(“secure:plugin_config.json”) # encrypted mount node.write_text(‘{“k”: 1}’) # ciphertext on disk node.read_text() # plaintext back

Key material is one or more comma-separated Fernet keys wrapped in a MultiFernet: the FIRST key encrypts, ALL keys decrypt (key rotation without bulk migration). The keys live only in memory on the instance and are never exposed; encryption_active reports whether they are installed.

The predefined secure mount (<base_dir>/secure/) is encrypted by definition. A mount declared encrypted: True in add_mount is encrypted too. There is no silent degradation (D5): using an encrypted mount without installed keys, or finding a non-Fernet payload on one, raises explicitly — never a plain-text fallback. Encryption is opt-in: without installed keys the plain mounts behave exactly as before.

class genro_asgi.storage.LocalStorage(base_dir=None)

Bases: object

Filesystem-only storage manager. API compatible with genro_storage.StorageManager.

Mount resolution order (see _resolve_mount): 1. Method mount_{prefix}() → dynamic, overridable via subclass 2. Dict _mounts → registered via add_mount()/configure() 3. ValueError if not found

Parameters:

base_dir (str | Path | None)

__init__(base_dir=None)

Create a storage manager without configured mounts.

Parameters:

base_dir (str | Path | None) – Base directory for resolving relative paths. Defaults to cwd.

Return type:

None

mount_site()

Predefined mount: server base directory.

Return type:

Path

mount_secure()

Predefined mount: server secure directory (encrypted at rest).

The secure mount is encrypted by definition; using it requires installed key material (see set_encryption_keys), else read/write raise explicitly (D5 — no plain-text fallback).

Return type:

Path

set_encryption_keys(keys)

Install key material for encrypted mounts.

keys is one or more comma-separated Fernet keys. They are wrapped in a MultiFernet: the FIRST key encrypts, ALL keys decrypt (key rotation without bulk migration). The keys are held only in memory and never exposed.

Parameters:

keys (str) – Comma-separated Fernet key(s), e.g. “<key>” or “<new>,<old>”.

Raises:

ValueError – if no non-empty key is given.

Return type:

None

property encryption_active: bool

True when key material is installed (read-only; keys stay hidden).

mount_is_encrypted(name)

True if the mount encrypts its content at rest.

Return type:

bool

Parameters:

name (str)

encrypt(data)

Encrypt bytes with the installed cipher (first key encrypts).

Raises:

RuntimeError – if no key material is installed (D5).

Return type:

bytes

Parameters:

data (bytes)

decrypt(token)

Decrypt bytes with the installed cipher (any key decrypts).

Raises:

• RuntimeError – if no key material is installed (D5).

• cryptography.fernet.InvalidToken – on a non-Fernet payload (D5 — no plain-text fallback).

Return type:

bytes

Parameters:

token (bytes)

property mounts: dict[str, Path]

The live dict of configured mounts (code → absolute Path).

Predefined mounts (mount_* methods) are resolved dynamically and are NOT listed here; this is the mapping populated by add_mount.

configure(source)

Configure mount points from a list of dicts.

Parameters:

source (str | list[dict[str, Any]]) – List of mount configurations

Return type:

None

Format:

[{‘name’: ‘site’, ‘type’: ‘local’, ‘path’: ‘/path/to/dir’}]

Note

Only type=’local’ is supported. Other types raise ValueError.

add_mount(config)

Add a single mount point.

Parameters:

config (dict[str, Any]) – {‘name’: str, ‘type’: ‘local’, ‘path’: str, ‘encrypted’: bool} encrypted (default False) makes the mount encrypt at rest; accessing it then requires installed keys (set_encryption_keys).

Raises:

• ValueError – if type != ‘local’

• ValueError – if name already exists

Return type:

None

delete_mount(name)

Remove a mount point.

Return type:

None

Parameters:

name (str)

get_mount_names()

List configured mount names.

Return type:

list[str]

has_mount(name)

True if mount exists (predefined method or configured).

Return type:

bool

Parameters:

name (str)

node(mount_or_path=None, *path_parts)

Create a storage node.

Parameters:

• mount_or_path (str | None) – “mount:path” or just “mount”

• *path_parts (str) – Additional path parts

Return type:

LocalStorageNode

Returns:

LocalStorageNode for the specified path

Examples

storage.node(‘site:resources/logo.png’) storage.node(‘site’, ‘resources’, ‘logo.png’) storage.node(‘site:resources’, ‘images’, ‘logo.png’)

Raises:

ValueError – if mount doesn’t exist

Parameters:

• mount_or_path (str | None)

• path_parts (str)

Return type:

LocalStorageNode

class genro_asgi.storage.LocalStorageNode(storage, mount, path)

Bases: object

Storage node for the local filesystem. API compatible with genro_storage.StorageNode.

Parameters:

• storage (LocalStorage)

• mount (str)

• path (str)

property fullpath: str

path” complete.

Type:

Return “mount

property path: str

Return path without mount.

property exists: bool

True if file/directory exists.

property isfile: bool

True if it’s a file.

property isdir: bool

True if it’s a directory.

property size: int

Size in bytes. 0 if doesn’t exist.

property basename: str

Filename with extension.

property suffix: str

Extension with dot.

property ext: str

Extension without dot.

property mimetype: str

MIME type based on extension.

property parent: LocalStorageNode

Return parent directory node.

read_bytes()

Read content as bytes (decrypted on encrypted mounts).

Return type:

bytes

read_text(encoding='utf-8')

Read content as text (decrypted on encrypted mounts).

Return type:

str

Parameters:

encoding (str)

read(mode='r', encoding='utf-8')

Read content. mode=’r’ for text, mode=’rb’ for binary.

Return type:

str | bytes

Parameters:

• mode (str)

• encoding (str)

write_bytes(data)

Write bytes (encrypted on encrypted mounts). Returns True if written.

Return type:

bool

Parameters:

data (bytes)

write_text(text, encoding='utf-8')

Write text (encrypted on encrypted mounts). Returns True if written.

Return type:

bool

Parameters:

• text (str)

• encoding (str)

write(data, mode='w', encoding='utf-8')

Write content. mode=’w’ for text, mode=’wb’ for binary.

Return type:

bool

Parameters:

• data (str | bytes)

• mode (str)

• encoding (str)

delete()

Remove this file. True if it was removed, False if it was absent.

Only files are in scope; pointing at a directory is an explicit error (this API is for single-file stores, not tree removal).

Raises:

IsADirectoryError – if the node is an existing directory.

Return type:

bool

move_to(dest)

Move this node (file OR directory) to dest, atomically on one mount.

The rename is a plain Path.rename — atomic when source and dest are on the same filesystem (the same mount), which is the spool’s claim/settle guarantee. Encryption is at rest per byte: a move relocates the stored bytes untouched, so it is valid across same-encryption mounts. Creates the destination parent if missing.

Raises:

FileNotFoundError – if this node does not exist.

Return type:

bool

Parameters:

dest (LocalStorageNode)

remove_tree()

Remove this node recursively (a directory and everything under it).

The tree-removal complement of delete() (which is file-only): removes a whole spool task folder in one call. True if something was removed, False if the node was absent. A plain file is removed too.

Return type:

bool

child(*parts)

Return a child node.

Return type:

LocalStorageNode

Parameters:

parts (str)

children()

List children if it’s a directory.

Return type:

list[LocalStorageNode]

class genro_asgi.storage.StorageNode(*args, **kwargs)

Bases: Protocol

Abstract interface for storage nodes.

Any storage backend (local, S3, HTTP) must implement this protocol. LocalStorageNode is the filesystem implementation.

property fullpath: str

path” complete.

Type:

Return “mount

property path: str

Return path without mount.

property exists: bool

True if file/directory exists.

property isfile: bool

True if it’s a file.

property isdir: bool

True if it’s a directory.

property basename: str

Filename with extension.

property mimetype: str

MIME type based on extension.

read_bytes()

Read content as bytes.

Return type:

bytes

read_text(encoding='utf-8')

Read content as text.

Return type:

str

Parameters:

encoding (str)

child(*parts)

Return a child node.

Return type:

StorageNode

Parameters:

parts (str)

children()

List children if it’s a directory.

Return type:

list[StorageNode]