orionis.storage.drivers.GoogleStorageDriver

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

Constructor: GoogleStorageDriver(config)

View In Hierarchy

Storage driver backed by Google Cloud Storage.

Uses the official Google Cloud client library (google-cloud-storage), which is an optional dependency: it is not installed with the framework. Install it before using this driver:

uv add 'orionis[gcs]'

The SDK is imported lazily on first operation, and every blocking call runs on a worker thread via asyncio.to_thread. Directories are virtual: prefixes are inferred from object names and explicit directories are stored as zero-byte path/ markers. Authentication uses the configured service-account key file or Application Default Credentials.

Method	`__aclFor`	Map a visibility level onto a GCS predefined ACL.
Method	`__blob`	Return a lightweight blob handle for normalized.
Method	`__blobOrFail`	Fetch a blob with loaded metadata or raise when absent.
Method	`__bucket`	Return the bucket handle, bootstrapping the client on first use.
Method	`__init__`	Initialize the driver from a GCS disk configuration entity.
Method	`__isPublicSync`	Check whether a blob grants read access to anonymous users.
Method	`__listKeysSync`	List every object name under prefix.
Method	`__missing`	Build the framework exception for a missing object.
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.
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 names 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 V4 signed 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	`_bucket_name`	Undocumented
Instance Variable	`_cloud_error`	Undocumented
Instance Variable	`_key_file`	Undocumented
Instance Variable	`_not_found`	Undocumented
Instance Variable	`_project`	Undocumented

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

Map a visibility level onto a GCS predefined ACL.

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

Returns
`str`	Predefined ACL name.

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

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

Return a lightweight blob handle for normalized.

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

Returns
`Any`	`Blob` handle bound to the bucket.

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

Fetch a blob with loaded metadata or raise when absent.

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

Returns
`Any`	`Blob` with populated properties.

Raises
`StorageFileNotFoundException`	If the object does not exist.

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

Return the bucket handle, bootstrapping the client on first use.

Returns
`Any`	Configured `Bucket` instance.

Raises
`MissingStorageDependencyException`	If `google-cloud-storage` is not installed.

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

Initialize the driver from a GCS 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 `project_id`, `key_file`, `bucket`, and `url`.

Returns
`None`	Undocumented

def __isPublicSync(self, blob: Any) -> bool: (source) ¶

Check whether a blob grants read access to anonymous users.

Parameters
blob:`Any`	`Blob` with loaded metadata.

Returns
`bool`	`True` when `allUsers` holds a reader or owner role. `False` when ACLs cannot be inspected (e.g. uniform bucket-level access).

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

List every object name under prefix.

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

Returns
`list[str]`	All matching object names, 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.

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`	URL derived from the configured base URL or the canonical `storage.googleapis.com` address.

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

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

overrides orionis.storage.contracts.IStorageDriver.files

List the object names 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; checksum holds the MD5 hash stored by GCS when available.

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

Not available on buckets with uniform bucket-level access, where the SDK raises the corresponding API error.

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 V4 signed URL for the object at path.

Requires credentials with a private key (a service-account key file); plain Application Default Credentials without a key cannot sign URLs.

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

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

Raises
`MissingStorageDependencyException`	If `google-cloud-storage` 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 or the canonical `storage.googleapis.com` 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'` when anonymous users can read the object, otherwise `'private'`.

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 applied through a predefined ACL, or `None` for the bucket default. Ignored by GCS when the bucket enforces uniform bucket-level access.

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 resumable transfer.

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

Returns
`None`	Undocumented

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

overrides orionis.storage.contracts.IStorageDriver.__slots__

Undocumented

_base_url: str | None = (source) ¶

Undocumented

_bucket: Any = (source) ¶

Undocumented

_bucket_name: str = (source) ¶

Undocumented

_cloud_error: type[Exception] = (source) ¶

Undocumented

_key_file: str | None = (source) ¶

Undocumented

_not_found: type[Exception] = (source) ¶

Undocumented

_project: str = (source) ¶

Undocumented