orionis.storage.drivers.LocalStorageDriver

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 LocalStorageDriver(IStorageDriver): (source)

Constructor: LocalStorageDriver(root, base_url)

View In Hierarchy

Storage driver backed by the local filesystem.

Every path is resolved inside the configured root directory and all blocking I/O runs on worker threads via asyncio.to_thread, keeping the event loop responsive. Visibility is mapped onto POSIX permission bits (0o644/0o600 for files), which degrades gracefully on platforms without a full POSIX mode implementation.

Static Method	`__iterEntries`	Yield the direct children of the requested kind under base.
Static Method	`__tempPath`	Build a unique staging path for an atomic write.
Static Method	`__walkEntries`	Yield every nested entry of the requested kind under base.
Method	`__absolute`	Return the absolute path for an already normalized path.
Method	`__buildUrl`	Build the public URL for normalized.
Method	`__commitStream`	Promote a finished temp file to its final destination.
Method	`__copySync`	Copy source to target synchronously.
Method	`__downloadSync`	Copy normalized to a local destination synchronously.
Method	`__fileMode`	Return the POSIX file mode for visibility.
Method	`__hashSync`	Compute the content hash of normalized synchronously.
Method	`__infoSync`	Build a metadata snapshot for normalized synchronously.
Method	`__init__`	Initialize the driver rooted at root.
Method	`__listSync`	List entries under normalized synchronously.
Method	`__moveSync`	Move source to target synchronously.
Method	`__openRead`	Open the file at normalized synchronously.
Method	`__openTemp`	Open a temporary sibling file for writing synchronously.
Method	`__readSync`	Read the file at normalized synchronously.
Method	`__relative`	Return the canonical relative path for absolute.
Method	`__statOrFail`	Stat the file at normalized or raise when it is absent.
Method	`__visibilityFromMode`	Derive the visibility level from POSIX permission bits.
Method	`__writeSync`	Atomically write data to normalized synchronously.
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	`_root`	Undocumented

@staticmethod

def __iterEntries(base: Path, *, want_dirs: bool) -> Iterator[Path]: (source) ¶

Yield the direct children of the requested kind under base.

Parameters
base:`Path`	Absolute directory to inspect.
want_dirs:`bool`	When `True`, yield directories; otherwise files.

Returns
`Iterator[Path]`	Undocumented

Yields
`Path`	Absolute path of each matching entry.

@staticmethod

def __tempPath(absolute: Path) -> Path: (source) ¶

Build a unique staging path for an atomic write.

Parameters
absolute:`Path`	Final absolute destination path.

Returns
`Path`	Sibling path carrying a random infix, so concurrent writers never share a staging file, and still ending in `.tmp` so listings keep filtering it out.

@staticmethod

def __walkEntries(base: Path, *, want_dirs: bool) -> Iterator[Path]: (source) ¶

Yield every nested entry of the requested kind under base.

Parameters
base:`Path`	Absolute directory used as the traversal root.
want_dirs:`bool`	When `True`, yield directories; otherwise files.

Returns
`Iterator[Path]`	Undocumented

Yields
`Path`	Absolute path of each matching entry.

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

Return the absolute path for an already normalized path.

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

Returns
`Path`	Absolute path inside the disk root.

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 __commitStream(self, tmp: Path, absolute: Path, visibility: str | None): (source) ¶

Promote a finished temp file to its final destination.

Parameters
tmp:`Path`	Fully written temporary file.
absolute:`Path`	Final absolute destination path.
visibility:`str \| None`	Visibility to apply, or `None` to keep the OS default.

Returns
`None`	Undocumented

def __copySync(self, source: str, target: str): (source) ¶

Copy source to target synchronously.

Parameters
source:`str`	Canonical root-relative source path.
target:`str`	Canonical root-relative destination path.

Returns
`None`	Undocumented

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

def __downloadSync(self, normalized: str, destination: str | Path) -> Path: (source) ¶

Copy normalized to a local destination synchronously.

Parameters
normalized:`str`	Canonical root-relative file path.
destination:`str \| Path`	Local target file or existing directory.

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

Raises
`StorageFileNotFoundException`	If the file does not exist.

def __fileMode(self, visibility: str) -> int: (source) ¶

Return the POSIX file mode for visibility.

Parameters
visibility:`str`	Visibility level (`'public'` or `'private'`).

Returns
`int`	Permission bits to apply to the file.

Raises
`UnsupportedStorageOperationException`	If visibility is not a supported level.

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

Compute the content hash of normalized synchronously.

Parameters
normalized:`str`	Canonical 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.

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

Build a metadata snapshot for normalized synchronously.

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

Returns
`FileInfo`	Immutable metadata snapshot of the file.

Raises
`StorageFileNotFoundException`	If the file does not exist.

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

Initialize the driver rooted at root.

Parameters
root:`Path`	Directory acting as the disk root. It is created when missing.
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 synchronously.

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 __moveSync(self, source: str, target: str): (source) ¶

Move source to target synchronously.

Parameters
source:`str`	Canonical root-relative source path.
target:`str`	Canonical root-relative destination path.

Returns
`None`	Undocumented

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

def __openRead(self, normalized: str, mode: str) -> BinaryIO: (source) ¶

Open the file at normalized synchronously.

Parameters
normalized:`str`	Canonical root-relative file path.
mode:`str`	Binary mode to open the file with.

Returns
`BinaryIO`	Open binary handle.

Raises
`StorageFileNotFoundException`	If the file does not exist.

def __openTemp(self, tmp: Path) -> BinaryIO: (source) ¶

Open a temporary sibling file for writing synchronously.

Parameters
tmp:`Path`	Absolute path of the temporary file.

Returns
`BinaryIO`	Open binary handle in write mode.

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

Read the file at normalized synchronously.

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

Returns
`bytes`	Complete file contents.

Raises
`StorageFileNotFoundException`	If the file does not exist.

def __relative(self, absolute: Path) -> str: (source) ¶

Return the canonical relative path for absolute.

Parameters
absolute:`Path`	Absolute path inside the disk root.

Returns
`str`	Root-relative path using `/` as separator.

def __statOrFail(self, normalized: str) -> tuple[Path, object]: (source) ¶

Stat the file at normalized or raise when it is absent.

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

Returns
`tuple[Path, object]`	The absolute path and its `os.stat_result`.

Raises
`StorageFileNotFoundException`	If the path does not reference an existing file.

def __visibilityFromMode(self, mode: int) -> str: (source) ¶

Derive the visibility level from POSIX permission bits.

Parameters
mode:`int`	Raw `st_mode` value of the file.

Returns
`str`	`'public'` when group or others can read the file, otherwise `'private'`.

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

Atomically write data to normalized synchronously.

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 OS default.

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.

The guess is derived from the file extension and requires no disk access.

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 local 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 local 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.

The write is atomic: data lands in a sibling temporary file that is renamed over the destination once complete.

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 OS default.

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.

Chunks are appended to a temporary file that atomically replaces the destination once the stream is exhausted, so a failed transfer never leaves a partial file behind.

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 OS default.

Returns
`None`	Undocumented

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

overrides orionis.storage.contracts.IStorageDriver.__slots__

Undocumented

_base_url: str | None = (source) ¶

Undocumented

_root: Path = (source) ¶

Undocumented