orionis.storage.drivers.S3StorageDriver

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

Constructor: S3StorageDriver(config)

View In Hierarchy

Storage driver backed by Amazon S3 (or S3-compatible services).

Uses the official AWS SDK for Python (boto3), which is an optional dependency: it is not installed with the framework. Install it before using this driver:

uv add 'orionis[s3]'

The SDK is imported lazily on first operation, and every blocking call runs on a worker thread via asyncio.to_thread so the event loop stays responsive. Directories are virtual: prefixes are inferred from object keys, and explicit directories are stored as zero-byte path/ marker objects.

Method	`__aclFor`	Map a visibility level onto an S3 canned ACL.
Method	`__client`	Return the boto3 S3 client, bootstrapping it on first use.
Method	`__headSync`	Fetch object metadata or raise when the object is absent.
Method	`__init__`	Initialize the driver from an S3 disk configuration entity.
Method	`__isMissing`	Check whether an SDK error denotes a missing object.
Method	`__listKeysSync`	List every object key under prefix.
Method	`__missing`	Build the framework exception for a missing object.
Method	`__putArgs`	Build the optional arguments for an object upload.
Method	`__seedStream`	Seed buffer with the current object content for a stream.
Async Method	`__spool`	Buffer an async byte stream into a spooled temporary file.
Method	`__urlFor`	Compose the public URL of an object without SDK involvement.
Method	`__visibilitySync`	Resolve the object visibility from its ACL grants.
Async Method	`copy`	Copy the object at source to target server-side.
Async Method	`createDirectory`	Create a zero-byte directory marker at path.
Async Method	`delete`	Delete the object at path.
Async Method	`deleteDirectory`	Recursively delete every object under path.
Async Method	`directories`	List the directory prefixes contained under path.
Async Method	`directoryExists`	Check whether any object exists under path.
Async Method	`download`	Download the object at path to the local filesystem.
Async Method	`exists`	Check whether an object exists at path.
Async Method	`files`	List the object keys that represent files under path.
Async Method	`hash`	Compute the content hash of the object at path.
Async Method	`info`	Collect a metadata snapshot for the object at path.
Async Method	`lastModified`	Return the last-modification timestamp of the object at path.
Async Method	`mimeType`	Return the MIME type of the object at path.
Async Method	`move`	Move the object at source to target.
Method	`open`	Open an asynchronous binary stream for the object at path.
Async Method	`read`	Read the full contents of the object at path.
Async Method	`readStream`	Stream the contents of the object at path in chunks.
Async Method	`setVisibility`	Change the visibility of the object at path.
Async Method	`size`	Return the size in bytes of the object at path.
Async Method	`temporaryUrl`	Build a presigned URL for the object at path.
Async Method	`url`	Build the public URL for the object at path.
Async Method	`visibility`	Return the visibility of the object at path.
Async Method	`write`	Write contents to path, replacing any existing object.
Async Method	`writeStream`	Write the chunks produced by stream to path.
Class Variable	`__slots__`	Undocumented
Instance Variable	`_base_url`	Undocumented
Instance Variable	`_bucket`	Undocumented
Instance Variable	`_client`	Undocumented
Instance Variable	`_client_error`	Undocumented
Instance Variable	`_endpoint`	Undocumented
Instance Variable	`_key`	Undocumented
Instance Variable	`_region`	Undocumented
Instance Variable	`_secret`	Undocumented
Instance Variable	`_use_path_style`	Undocumented

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

Map a visibility level onto an S3 canned ACL.

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

Returns
`str`	Canned ACL name.

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

def __client(self) -> Any: (source) ¶

Return the boto3 S3 client, bootstrapping it on first use.

Returns
`Any`	Configured `boto3` S3 client.

Raises
`MissingStorageDependencyException`	If `boto3` is not installed.

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

Fetch object metadata or raise when the object is absent.

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

Returns
`dict`	Raw `head_object` response.

Raises
`StorageFileNotFoundException`	If the object does not exist.

def __init__(self, config: object): (source) ¶

Initialize the driver from an S3 disk configuration entity.

No SDK import or network activity happens here; the client is bootstrapped lazily on first use.

Parameters
config:`object`	Disk configuration exposing `bucket`, `region`, `key`, `secret`, `url`, `endpoint`, and `use_path_style_endpoint`.

Returns
`None`	Undocumented

def __isMissing(self, exc: Exception) -> bool: (source) ¶

Check whether an SDK error denotes a missing object.

Parameters
exc:`Exception`	Exception raised by the boto3 client.

Returns
`bool`	`True` when the error code maps to a 404 condition.

def __listKeysSync(self, prefix: str) -> list[str]: (source) ¶

List every object key under prefix.

Parameters
prefix:`str`	Key prefix to filter by; empty string lists the bucket.

Returns
`list[str]`	All matching object keys, including directory markers.

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

Build the framework exception for a missing object.

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

Returns
`StorageFileNotFoundException`	Exception instance ready to be raised.

def __putArgs(self, normalized: str, visibility: str | None) -> dict[str, str]: (source) ¶

Build the optional arguments for an object upload.

Parameters
normalized:`str`	Canonical root-relative file path.
visibility:`str \| None`	Visibility to apply, or `None` to omit the ACL.

Returns
`dict[str, str]`	Extra arguments with `ContentType` and optional `ACL`.

def __seedStream(self, buffer: BinaryIO, normalized: str, mode: str): (source) ¶

Seed buffer with the current object content for a stream.

Downloads the object into the buffer and positions the cursor according to mode. Missing objects are tolerated for append modes and rejected for read modes.

Parameters
buffer:`BinaryIO`	Spooled buffer backing the stream.
normalized:`str`	Canonical root-relative file path.
mode:`str`	Binary mode requested by the caller.

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the object does not exist and mode requires it.

async def __spool(self, stream: AsyncIterable[bytes]) -> BinaryIO: (source) ¶

Buffer an async byte stream into a spooled temporary file.

Parameters
stream:`AsyncIterable[bytes]`	Asynchronous byte-chunk producer.

Returns
`BinaryIO`	Rewound buffer holding the full payload.

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

Compose the public URL of an object without SDK involvement.

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

Returns
`str`	Public URL derived from the configured base URL, custom endpoint, or the canonical virtual-host address.

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

Resolve the object visibility from its ACL grants.

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

Returns
`str`	`'public'` when anonymous users hold a read grant, otherwise `'private'`.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.copy

Copy the object at source to target server-side.

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

Returns
`None`	Undocumented

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

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

overrides orionis.storage.contracts.IStorageDriver.createDirectory

Create a zero-byte directory marker at path.

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 object at path.

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

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

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

overrides orionis.storage.contracts.IStorageDriver.deleteDirectory

Recursively delete every object under path.

Parameters
path:`str`	Root-relative directory path. The empty string clears the whole bucket prefix space.

Returns
`bool`	`True` if at least one object was removed.

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

overrides orionis.storage.contracts.IStorageDriver.directories

List the directory prefixes contained under path.

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

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

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

overrides orionis.storage.contracts.IStorageDriver.directoryExists

Check whether any object exists under path.

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

Returns
`bool`	`True` if the prefix contains at least one object.

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

overrides orionis.storage.contracts.IStorageDriver.download

Download the object at path to 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 object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.exists

Check whether an object exists at path.

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

Returns
`bool`	`True` if an object exists at the given key.

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

overrides orionis.storage.contracts.IStorageDriver.files

List the object keys that represent files under path.

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

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 object at path.

The object is streamed in chunks, so large files never load fully into memory.

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

Returns
`str`	Hexadecimal digest of the object contents.

Raises
`StorageFileNotFoundException`	If the object 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 object at path.

The snapshot is built from object metadata only; the checksum field is None because computing it would require a full download (use hash instead).

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

Returns
`FileInfo`	Immutable entity with size, MIME type, timestamps, ETag, visibility, and URL.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.lastModified

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

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

Returns
`datetime`	Timezone-aware modification timestamp.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.mimeType

Return the MIME type of the object at path.

Prefers the Content-Type stored in S3 and falls back to a guess based on the file extension.

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

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

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.move

Move the object at source to target.

Implemented as a server-side copy followed by a delete of the source object.

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

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the source object 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 object at path.

Read-oriented modes download the object into a spooled temporary buffer; writable modes upload the buffered content back to S3 when the stream is closed.

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 object at path.

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

Returns
`bytes`	Complete object contents.

Raises
`StorageFileNotFoundException`	If the object 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 object 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 object contents.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.setVisibility

Change the visibility of the object at path.

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

Returns
`None`	Undocumented

Raises
`StorageFileNotFoundException`	If the object 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 object at path.

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

Returns
`int`	Object size in bytes.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

overrides orionis.storage.contracts.IStorageDriver.temporaryUrl

Build a presigned URL for the object at path.

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

Returns
`str`	Presigned GET URL valid for expires_in seconds.

Raises
`MissingStorageDependencyException`	If `boto3` is not installed.

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

overrides orionis.storage.contracts.IStorageDriver.url

Build the public URL for the object at path.

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

Returns
`str`	URL derived from the configured base URL, custom endpoint, or the canonical virtual-host address.

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

overrides orionis.storage.contracts.IStorageDriver.visibility

Return the visibility of the object at path.

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

Returns
`str`	`'public'` or `'private'` based on the object ACL.

Raises
`StorageFileNotFoundException`	If the object 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 object.

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` for the bucket 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.

The payload is buffered into a spooled temporary file (spilling to disk past 8 MiB) and uploaded with the SDK's managed transfer, which switches to multipart uploads for large objects.

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