Storage

Filesystem storage with mount system.

LocalStorage - Filesystem-only storage with genro-storage compatible API.

This module provides a minimal storage implementation that uses the same API as genro-storage, but only supports 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

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)[source]

Bases: object

Storage manager filesystem-only. 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

__init__(base_dir=None)[source]

Create storage manager without configured mounts.

Parameters:

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

add_mount(config)[source]

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:
Return type:

None

configure(source)[source]

Configure mount points from 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.

decrypt(token)[source]

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

delete_mount(name)[source]

Remove a mount point.

Return type:

None

encrypt(data)[source]

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

Raises:

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

Return type:

bytes

property encryption_active: bool

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

get_mount_names()[source]

List configured mount names.

Return type:

list[str]

has_mount(name)[source]

True if mount exists (predefined method or configured).

Return type:

bool

mount_is_encrypted(name)[source]

True if the mount encrypts its content at rest.

Return type:

bool

mount_secure()[source]

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

mount_site()[source]

Predefined mount: server base directory.

Return type:

Path

node(mount_or_path=None, *path_parts)[source]

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

set_encryption_keys(keys)[source]

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

class genro_asgi.storage.LocalStorageNode(storage, mount, path)[source]

Bases: object

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

property basename: str

Filename with extension.

child(*parts)[source]

Return a child node.

Return type:

LocalStorageNode

children()[source]

List children if it’s a directory.

Return type:

list[LocalStorageNode]

delete()[source]

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

property exists: bool

True if file/directory exists.

property ext: str

Extension without dot.

property fullpath: str

path” complete.

Type:

Return “mount

property isdir: bool

True if it’s a directory.

property isfile: bool

True if it’s a file.

property mimetype: str

MIME type based on extension.

property parent: LocalStorageNode

Return parent directory node.

property path: str

Return path without mount.

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

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

Return type:

str | bytes

read_bytes()[source]

Read content as bytes (decrypted on encrypted mounts).

Return type:

bytes

read_text(encoding='utf-8')[source]

Read content as text (decrypted on encrypted mounts).

Return type:

str

property size: int

Size in bytes. 0 if doesn’t exist.

property suffix: str

Extension with dot.

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

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

Return type:

bool

write_bytes(data)[source]

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

Return type:

bool

write_text(text, encoding='utf-8')[source]

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

Return type:

bool

class genro_asgi.storage.StorageNode(*args, **kwargs)[source]

Bases: Protocol

Abstract interface for storage nodes.

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

property basename: str

Filename with extension.

child(*parts)[source]

Return a child node.

Return type:

StorageNode

children()[source]

List children if it’s a directory.

Return type:

list[StorageNode]

property exists: bool

True if file/directory exists.

property fullpath: str

path” complete.

Type:

Return “mount

property isdir: bool

True if it’s a directory.

property isfile: bool

True if it’s a file.

property mimetype: str

MIME type based on extension.

property path: str

Return path without mount.

read_bytes()[source]

Read content as bytes.

Return type:

bytes

read_text(encoding='utf-8')[source]

Read content as text.

Return type:

str