temporalio.contrib.aws.s3driver._driver

module documentation
(source)

Amazon S3 storage driver for Temporal external storage.

Warning

This API is experimental.

Function _format_client_context Format the client's describe() output as ", k=v, k=v" for error messages. Returns an empty string when the client reports no metadata or describe itself raises (diagnostic output must never mask the real error).
Async Function _gather_with_cancellation Run coroutines concurrently, cancelling all remaining tasks if one fails.
Function _percent_encode Percent-encode a path segment per the S3 key spec.
Constant _S3_SAFE_CHARS Undocumented
Type Variable _T Undocumented
def _format_client_context(client: S3StorageDriverClient) -> str: (source)

Format the client's describe() output as ", k=v, k=v" for error messages. Returns an empty string when the client reports no metadata or describe itself raises (diagnostic output must never mask the real error).

async def _gather_with_cancellation(coros: Sequence[Coroutine[Any, Any, _T]]) -> list[_T]: (source)

Run coroutines concurrently, cancelling all remaining tasks if one fails.

def _percent_encode(val: str | None) -> str | None: (source)

Percent-encode a path segment per the S3 key spec.

Encodes every byte outside _S3_SAFE_CHARS as %XX with upper-case hex digits, operating on the UTF-8 bytes so non-ASCII characters are escaped byte by byte. Returns None for empty or missing values so callers can substitute "null".

_S3_SAFE_CHARS = (source)

Undocumented

Value
frozenset((string.ascii_letters + string.digits + '!-_.*\'()'))
_T = (source)

Undocumented

Value
TypeVar('_T')