---
sidebar:
  label: roboto.storage.upload_transaction
  order: 8
title: roboto.storage.upload_transaction
---
## Module Contents

### TransactionFile

```python
class roboto.storage.upload_transaction.TransactionFile
```

`from roboto.storage.upload_transaction import TransactionFile`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L40-L43)

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)

**Attributes**

- **TransactionFile.destination_path** (`str`)
- **TransactionFile.file_size** (`int`)
- **TransactionFile.local_path** (`pathlib.Path`)

### UploadTransaction

```python
class roboto.storage.upload_transaction.UploadTransaction(
    items: collections.abc.Sequence[TransactionFile],
    association: roboto.association.Association,
    device_id: Optional[str] = None,
    origination: Optional[str] = None,
    batch_size: Optional[int] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
    caller_org_id: Optional[str] = None,
)
```

`from roboto.storage.upload_transaction import UploadTransaction`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L52-L260)

**Parameters**

- **items** (`collections.abc.Sequence[TransactionFile]`)
- **association** (`roboto.association.Association`)
- **device_id** (`Optional[str]`)
- **origination** (`Optional[str]`)
- **batch_size** (`Optional[int]`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)
- **caller_org_id** (`Optional[str]`)

**Properties**

- **UploadTransaction.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()`](/reference/python-sdk/roboto/storage/upload_transaction#roboto.storage.upload_transaction.UploadTransaction.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.

- **UploadTransaction.transaction_id** (`str`)

- **UploadTransaction.upload_mappings** (`dict[str, str]`)

#### UploadTransaction.await_uploads()

```python
def await_uploads() -> None
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L169-L206)

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`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.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.

**Returns**

- `None`

#### UploadTransaction.make_credential_provider()

```python
def make_credential_provider() -> roboto.storage.object_store.CredentialProvider
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L208-L220)

**Returns**

- `roboto.storage.object_store.CredentialProvider`

#### UploadTransaction.register_upload()

```python
def register_upload(
    file: UploadableFile,
    future: roboto.storage.object_store.FutureLike[None],
) -> None
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L222-L229)

Register a pending upload future.

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

**Parameters**

- **file** (`UploadableFile`)
- **future** (`roboto.storage.object_store.FutureLike[None]`)

**Returns**

- `None`

### UploadableFile

```python
class roboto.storage.upload_transaction.UploadableFile
```

`from roboto.storage.upload_transaction import UploadableFile`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L46-L49)

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)

**Attributes**

- **UploadableFile.destination_path** (`str`)
- **UploadableFile.local_path** (`pathlib.Path`)
- **UploadableFile.upload_uri** (`str`)

### logger

```python
roboto.storage.upload_transaction.logger
```

`from roboto.storage.upload_transaction import logger`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/upload_transaction.py#L37-L37)