orionis.storage.drivers.MemoryStorageDriver

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class MemoryStorageDriver(IStorageDriver): (source)

Constructor: MemoryStorageDriver(base_url)

View In Hierarchy

Storage driver keeping every object in process memory.

Designed for testing and ephemeral workloads: it implements the full driver contract over plain dictionaries, enabling fakes such as a future Storage.fake() without touching the rest of the component. Directories exist implicitly through file prefixes and explicitly through createDirectory.

Concurrency

The store is a plain dictionary mutated without locks. Every operation completes without awaiting midway, so concurrent tasks on a single event loop never observe a partial mutation. No guarantee is offered when the same path is mutated from several threads at once, which streams opened with open do because they flush their buffer on a worker thread.

Static Method	`__ancestors`	Return every ancestor prefix of path.
Method	`__allDirectories`	Return every explicit and implicit directory path.
Method	`__buildUrl`	Build the public URL for normalized.
Method	`__digest`	Compute the hex digest of data using algorithm.
Method	`__entryOrFail`	Return the entry stored at normalized or raise when absent.
Method	`__init__`	Initialize an empty in-memory disk.
Method	`__listSync`	List entries under normalized.
Method	`__storeSync`	Persist data under normalized, preserving creation time.
Async Method	`copy`	Copy the file at source to target.
Async Method	`createDirectory`	Create the directory at path, including missing parents.
Async Method	`delete`	Delete the file at path.
Async Method	`deleteDirectory`	Recursively delete the directory at path.
Async Method	`directories`	List the directory paths contained in the directory at path.
Async Method	`directoryExists`	Check whether a directory exists at path.
Async Method	`download`	Copy the file at path to a location on the local filesystem.
Async Method	`exists`	Check whether a file exists at path.
Async Method	`files`	List the file paths contained in the directory at path.
Async Method	`hash`	Compute the content hash of the file at path.
Async Method	`info`	Collect a metadata snapshot for the file at path.
Async Method	`lastModified`	Return the last-modification timestamp of the file at path.
Async Method	`mimeType`	Guess the MIME type of the file at path.
Async Method	`move`	Move the file at source to target.
Method	`open`	Open an asynchronous binary stream for the file at path.
Async Method	`read`	Read the full contents of the file at path.
Async Method	`readStream`	Stream the contents of the file at path in chunks.
Async Method	`setVisibility`	Change the visibility of the file at path.
Async Method	`size`	Return the size in bytes of the file at path.
Async Method	`temporaryUrl`	Build a signed, time-limited URL for the file at path.
Async Method	`url`	Build the public URL for the file at path.
Async Method	`visibility`	Return the visibility of the file at path.
Async Method	`write`	Write contents to path, replacing any existing file.
Async Method	`writeStream`	Write the chunks produced by stream to path.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_base_url`	Undocumented
Instance Variable	`_directories`	Undocumented
Instance Variable	`_files`	Undocumented

@staticmethod

def __ancestors(path: str) -> set[str]: (source) ¶

Return every ancestor prefix of path.

Parameters
path:`str`	Canonical root-relative path.

Returns
`set[str]`	Ancestor directory prefixes, excluding the path itself.

def __allDirectories(self) -> set[str]: (source) ¶

Return every explicit and implicit directory path.

Returns
`set[str]`	Union of explicitly created directories and the ancestor prefixes implied by stored file paths.

def __buildUrl(self, normalized: str) -> str: (source) ¶

Build the public URL for normalized.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`str`	URL composed from the configured base URL.

def __digest(self, data: bytes, algorithm: str) -> str: (source) ¶

Compute the hex digest of data using algorithm.

Parameters
data:`bytes`	Raw bytes to hash.
algorithm:`str`	Any algorithm name accepted by `hashlib.new`.

Returns
`str`	Hexadecimal digest of the data.

Raises
`UnsupportedStorageOperationException`	If algorithm is not available.

def __entryOrFail(self, normalized: str) -> _MemoryEntry: (source) ¶

Return the entry stored at normalized or raise when absent.

Parameters
normalized:`str`	Canonical root-relative file path.

Returns
`_MemoryEntry`	Stored entry for the path.

Raises
`StorageFileNotFoundException`	If no file exists at the path.

def __init__(self, base_url: str | None = None): (source) ¶

Initialize an empty in-memory disk.

Parameters
base_url:`str \| None`	Base URL used to build public file URLs, or `None` when the disk does not expose URLs.

Returns
`None`	Undocumented

def __listSync(self, normalized: str, *, recursive: bool, want_dirs: bool) -> list[str]: (source) ¶

List entries under normalized.

Parameters
normalized:`str`	Canonical root-relative directory path.
recursive:`bool`	When `True`, traverse the whole subtree.
want_dirs:`bool`	When `True`, collect directories; otherwise files.

Returns
`list[str]`	Sorted root-relative entry paths.

def __storeSync(self, normalized: str, data: bytes, visibility: str | None): (source) ¶

Persist data under normalized, preserving creation time.

Parameters
normalized:`str`	Canonical root-relative file path.
data:`bytes`	Raw bytes to persist.
visibility:`str \| None`	Visibility to apply, or `None` to keep the current one.

Returns
`None`	Undocumented

async def copy(self, source: str, target: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.copy

Copy the file at source to target.

Parameters
source:`str`	Root-relative path of the existing file.
target:`str`	Root-relative destination path.

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the source file does not exist.

async def createDirectory(self, path: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.createDirectory

Create the directory at path, including missing parents.

Parameters
path:`str`	Root-relative directory path.

Returns
`None`	Undocumented

async def delete(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.delete

Delete the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bool`	`True` if the file existed and was removed.

async def deleteDirectory(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.deleteDirectory

Recursively delete the directory at path.

Parameters
path:`str`	Root-relative directory path.

Returns
`bool`	`True` if the directory existed and was removed.

async def directories(self, path: str = '', *, recursive: bool = False) -> list[str]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.directories

List the directory paths contained in the directory at path.

Parameters
path:`str`	Root-relative directory path. Empty string for the root.
recursive:`bool`	When `True`, include all nested directories.

Returns
`list[str]`	Sorted root-relative directory paths.

async def directoryExists(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.directoryExists

Check whether a directory exists at path.

Parameters
path:`str`	Root-relative directory path. The empty string denotes the disk root.

Returns
`bool`	`True` if a directory exists at the given path.

async def download(self, path: str, destination: str | Path) -> Path: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.download

Copy the file at path to a location on the local filesystem.

Parameters
path:`str`	Root-relative file path on the disk.
destination:`str \| Path`	Local target. When it points to an existing directory the file keeps its original name inside that directory.

Returns
`Path`	Absolute local path of the downloaded file.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def exists(self, path: str) -> bool: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.exists

Check whether a file exists at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bool`	`True` if a file exists at the given path.

async def files(self, path: str = '', *, recursive: bool = False) -> list[str]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.files

List the file paths contained in the directory at path.

Parameters
path:`str`	Root-relative directory path. Empty string for the root.
recursive:`bool`	When `True`, include files from all nested directories.

Returns
`list[str]`	Sorted root-relative file paths.

async def hash(self, path: str, algorithm: str = 'sha256') -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.hash

Compute the content hash of the file at path.

Parameters
path:`str`	Root-relative file path.
algorithm:`str`	Any algorithm name accepted by `hashlib.new`.

Returns
`str`	Hexadecimal digest of the file contents.

Raises
`StorageFileNotFoundException`	If the file does not exist.
`UnsupportedStorageOperationException`	If algorithm is not available.

async def info(self, path: str) -> FileInfo: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.info

Collect a metadata snapshot for the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`FileInfo`	Immutable entity with size, MIME type, timestamps, hashes, visibility, and URL when available.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def lastModified(self, path: str) -> datetime: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.lastModified

Return the last-modification timestamp of the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`datetime`	Timezone-aware modification timestamp (UTC).

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def mimeType(self, path: str) -> str | None: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.mimeType

Guess the MIME type of the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`str \| None`	MIME type, or `None` when it cannot be determined.

async def move(self, source: str, target: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.move

Move the file at source to target.

Parameters
source:`str`	Root-relative path of the existing file.
target:`str`	Root-relative destination path.

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the source file does not exist.

def open(self, path: str, mode: str = 'rb') -> AsyncStream: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.open

Open an asynchronous binary stream for the file at path.

Parameters
path:`str`	Root-relative file path.
mode:`str`	Binary mode: `'rb'`, `'wb'`, `'ab'`, `'rb+'`, `'wb+'`, or `'ab+'`.

Returns
`AsyncStream`	Lazily opened stream; use it as an async context manager.

Raises
`UnsupportedStorageOperationException`	If mode is not a supported binary mode.

async def read(self, path: str) -> bytes: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.read

Read the full contents of the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`bytes`	Complete file contents.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def readStream(self, path: str, chunk_size: int = _CHUNK_SIZE) -> AsyncIterator[bytes]: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.readStream

Stream the contents of the file at path in chunks.

Parameters
path:`str`	Root-relative file path.
chunk_size:`int`	Maximum number of bytes per yielded chunk.

Returns
`AsyncIterator[bytes]`	Undocumented

Yields
`bytes`	Consecutive chunks of the file contents.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def setVisibility(self, path: str, visibility: str): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.setVisibility

Change the visibility of the file at path.

Parameters
path:`str`	Root-relative file path.
visibility:`str`	Target visibility (`'public'` or `'private'`).

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the file does not exist.
`UnsupportedStorageOperationException`	If visibility is not a supported level.

async def size(self, path: str) -> int: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.size

Return the size in bytes of the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`int`	File size in bytes.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def temporaryUrl(self, path: str, expires_in: int) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.temporaryUrl

Build a signed, time-limited URL for the file at path.

The memory driver cannot sign URLs, so this operation always fails.

Parameters
path:`str`	Root-relative file path.
expires_in:`int`	Lifetime of the URL in seconds.

Returns
`str`	Never returned by this driver.

Raises
`UnsupportedStorageOperationException`	Always, since memory disks cannot sign URLs.

async def url(self, path: str) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.url

Build the public URL for the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`str`	Publicly accessible URL for the file.

Raises
`UnsupportedStorageOperationException`	If the disk has no base URL configured.

async def visibility(self, path: str) -> str: (source) ¶

overrides orionis.storage.contracts.IStorageDriver.visibility

Return the visibility of the file at path.

Parameters
path:`str`	Root-relative file path.

Returns
`str`	`'public'` or `'private'`.

Raises
`StorageFileNotFoundException`	If the file does not exist.

async def write(self, path: str, contents: bytes | str, visibility: str | None = None): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.write

Write contents to path, replacing any existing file.

Parameters
path:`str`	Root-relative file path.
contents:`bytes \| str`	Data to persist. Strings are encoded as UTF-8.
visibility:`str \| None`	Visibility to apply, or `None` to keep the current one.

Returns
`None`	Undocumented

async def writeStream(self, path: str, stream: AsyncIterable[bytes], visibility: str | None = None): (source) ¶

overrides orionis.storage.contracts.IStorageDriver.writeStream

Write the chunks produced by stream to path.

Parameters
path:`str`	Root-relative file path.
stream:`AsyncIterable[bytes]`	Asynchronous byte-chunk producer.
visibility:`str \| None`	Visibility to apply, or `None` to keep the current one.

Returns
`None`	Undocumented

__slots__: tuple[str, ...] = (source) ¶

overrides orionis.storage.contracts.IStorageDriver.__slots__

Undocumented

_base_url: str | None = (source) ¶

Undocumented

_directories: set[str] = (source) ¶

Undocumented

_files: dict[str, _MemoryEntry] = (source) ¶

Undocumented