roboto.storage.upload_transaction#

Module Contents#

class roboto.storage.upload_transaction.TransactionFile#

Bases: TypedDict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:

d = {} for k, v in iterable:

d[k] = v

dict(**kwargs) -> new dictionary initialized with the name=value pairs

in the keyword argument list. For example: dict(one=1, two=2)

destination_path: str#
file_size: int#
local_path: pathlib.Path#
class roboto.storage.upload_transaction.UploadTransaction(items, association, device_id=None, origination=None, batch_size=None, roboto_client=None, caller_org_id=None)#
Parameters:
await_uploads()#

Wait for all registered uploads to complete, then report progress to API.

Must be called while still inside the object_store context manager, since the transfer manager may be shut down when that context exits.

Raises:

  • ExceptionGroup – If any uploads fail, an ExceptionGroup containing all upload errors is raised after reporting successfully completed uploads.

  • RobotoInternalException – If the server’s progress response reports (uri, file_id) pairs that do not correspond one-to-one with the uploads reported to it.

Return type:

None

property completed_uploads: dict[pathlib.Path, str]#

Mapping from each uploaded local path to the ID of the file record it created.

Populated from the {uri, file_id} pairs in progress-report responses, and complete for every upload already reported to PUT v1/files/upload/<id>/progress, which await_uploads() does at each batch boundary. A pair set that does not correspond one-to-one with the reported uploads raises instead of degrading the mapping.

Return type:

dict[pathlib.Path, str]

make_credential_provider()#
Return type:

roboto.storage.object_store.CredentialProvider

register_upload(file, future)#

Register a pending upload future.

Call this immediately after initiating each S3 transfer. The future will be awaited when await_uploads() is called.

Parameters:
Return type:

None

property transaction_id: str#
Return type:

str

property upload_mappings: dict[str, str]#
Return type:

dict[str, str]

class roboto.storage.upload_transaction.UploadableFile#

Bases: TypedDict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:

d = {} for k, v in iterable:

d[k] = v

dict(**kwargs) -> new dictionary initialized with the name=value pairs

in the keyword argument list. For example: dict(one=1, two=2)

destination_path: str#
local_path: pathlib.Path#
upload_uri: str#
roboto.storage.upload_transaction.logger#
On this page