---
sidebar:
  hidden: true
title: roboto.storage.object_store.s3
---
## Module Contents

### ProgressCallbackInvoker

```python
class roboto.storage.object_store.s3.ProgressCallbackInvoker(callback)
```

`from roboto.storage.object_store.s3 import ProgressCallbackInvoker`

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

Bases: `boto3.s3.transfer.BaseSubscriber`

Invoke a provided callback via a subscriber.

Taken from [https://github.com/boto/boto3/blob/develop/boto3/s3/transfer.py#L505C1-L516C42](https://github.com/boto/boto3/blob/develop/boto3/s3/transfer.py#L505C1-L516C42)

#### ProgressCallbackInvoker.on_progress()

```python
def on_progress(bytes_transferred, **kwargs)
```

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

Callback to be invoked when progress is made on transfer

This callback can be useful for:

> - Recording and displaying progress

**Parameters**

- **future** (`s3transfer.futures.TransferFuture`): The TransferFuture representing the requested transfer.
- **bytes_transferred** (`int`): The number of bytes transferred for that invocation of the callback. Note that a negative amount can be provided, which usually indicates that an in-progress request needed to be retried and thus progress was rewound.

### S3Store

```python
class roboto.storage.object_store.s3.S3Store(
    s3_client: botocore.client.BaseClient,
    transfer_config: Optional[boto3.s3.transfer.TransferConfig] = None,
)
```

`from roboto.storage.object_store import S3Store`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/object_store/s3.py#L44-L136)

Bases: [`roboto.storage.object_store.object_store.ObjectStore`](/reference/python-sdk/roboto/storage/object_store/object_store#roboto.storage.object_store.object_store.ObjectStore)

Base class for protocol classes.

Protocol classes are defined as:

```
class Proto(Protocol):
    def meth(self) -> int:
        ...
```

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

```
class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check
```

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

```
class GenProto[T](Protocol):
    def meth(self) -> T:
        ...
```

**Parameters**

- **s3_client** (`botocore.client.BaseClient`)
- **transfer_config** (`Optional[boto3.s3.transfer.TransferConfig]`)

#### S3Store.create()

```python
@classmethod
def create(
    credential_provider: roboto.storage.object_store.object_store.CredentialProvider,
    **kwargs,
) -> S3Store
```

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

Factory function to assemble an S3Store with refreshable credentials.

**Parameters**

- **credential_provider** (`roboto.storage.object_store.object_store.CredentialProvider`)

**Returns**

- `S3Store`

#### S3Store.get()

```python
def get(
    source_uri: str,
    destination: pathlib.Path,
    on_progress: Optional[roboto.storage.object_store.object_store.OnProgress] = None,
) -> roboto.storage.object_store.object_store.FutureLike[None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/object_store/s3.py#L118-L136)

Downloads a file from a cloud URI to a local path.

Parent directories of the destination path are created automatically if they don't exist.

**Parameters**

- **source_uri** (`str`): Full URI (e.g., 's3://my-bucket/folder/data.csv')
- **destination** (`pathlib.Path`): Local path where the file should be saved.
- **on_progress** (`Optional[roboto.storage.object_store.object_store.OnProgress]`): Optional callback to be periodically called with the number of bytes downloaded.

**Returns**

- `roboto.storage.object_store.object_store.FutureLike[None]`

#### S3Store.put()

```python
def put(
    source: pathlib.Path,
    destination_uri: str,
    on_progress: Optional[roboto.storage.object_store.object_store.OnProgress] = None,
) -> roboto.storage.object_store.object_store.FutureLike[None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/storage/object_store/s3.py#L100-L116)

Uploads a local file to a specific cloud URI.

**Parameters**

- **source** (`pathlib.Path`): Local path to the file.
- **destination_uri** (`str`): Full URI (e.g., 's3://my-bucket/folder/data.csv')
- **on_progress** (`Optional[roboto.storage.object_store.object_store.OnProgress]`): Optional callback to be periodically called with the number of bytes uploaded.

**Returns**

- `roboto.storage.object_store.object_store.FutureLike[None]`