Skip to content

AWS S3#

The S3 client and its companion Bucket class provide the AWS S3 implementation of unicloud's storage contract. Credentials are read from the standard AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_DEFAULT_REGION environment variables; additional boto3 configuration (for example a custom botocore.config.Config) can be passed via the configs argument.

unicloud.aws.aws.S3 #

Bases: CloudStorageFactory

AWS S3 client — the :class:CloudStorageFactory implementation for Amazon S3.

Instantiating the class constructs a boto3 S3 client using credentials pulled from the environment. Additional keyword arguments that you would otherwise pass to boto3.client (for example a custom botocore.config.Config, or a different region_name) can be forwarded via the configs parameter.

Examples:

  • Create a client from environment variables only: 
    >>> s3 = S3()  # doctest: +SKIP
  • Create a client with a custom botocore config and region override: 
    >>> from botocore.config import Config
>>> s3 = S3(configs={
...     "config": Config(signature_version="s3v4"),
...     "region_name": "us-west-2",
... })  # doctest: +SKIP
See Also

unicloud.google_cloud.gcs.GCS: The matching Google Cloud Storage implementation.

Source code in src/unicloud/aws/aws.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
class S3(CloudStorageFactory):
    """AWS S3 client — the :class:`CloudStorageFactory` implementation for Amazon S3.

    Instantiating the class constructs a ``boto3`` S3 client using credentials pulled from the
    environment. Additional keyword arguments that you would otherwise pass to ``boto3.client``
    (for example a custom ``botocore.config.Config``, or a different ``region_name``) can be
    forwarded via the ``configs`` parameter.

    Examples:
        - Create a client from environment variables only:
            ```python
            >>> s3 = S3()  # doctest: +SKIP

            ```
        - Create a client with a custom botocore config and region override:
            ```python
            >>> from botocore.config import Config
            >>> s3 = S3(configs={
            ...     "config": Config(signature_version="s3v4"),
            ...     "region_name": "us-west-2",
            ... })  # doctest: +SKIP

            ```

    See Also:
        unicloud.google_cloud.gcs.GCS: The matching Google Cloud Storage implementation.
    """

    def __init__(
        self,
        configs: Optional[Dict] = None,
    ):
        """Initialize the S3 client.

        Credentials are read from the standard AWS environment variables. Any extra keyword
        arguments you want passed to ``boto3.client("s3", ...)`` can be supplied via ``configs``
        — they override the defaults that this class sets for ``service_name``, ``region_name``,
        ``aws_access_key_id``, and ``aws_secret_access_key``.

        Args:
            configs: Optional dictionary of extra keyword arguments to forward to ``boto3.client``.
                Useful for passing a ``botocore.config.Config`` (``{"config": Config(...)}``), a
                non-default region (``{"region_name": "us-west-2"}``), or for swapping the
                service name in tests.

        Raises:
            ValueError: If any of ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, or
                ``AWS_DEFAULT_REGION`` is not set in the environment.
            NoCredentialsError: Propagated from ``boto3`` when it cannot find credentials to sign
                requests with.
            PartialCredentialsError: Propagated from ``boto3`` when only some of the required
                credentials are present.

        Examples:
            - Default construction reading every credential from the environment:
                ```python
                >>> s3 = S3()  # doctest: +SKIP

                ```
            - Pass a custom botocore Config to override the signature version:
                ```python
                >>> from botocore.config import Config
                >>> s3 = S3(configs={
                ...     "config": Config(signature_version="s3v4"),
                ...     "region_name": "us-west-2",
                ... })  # doctest: +SKIP

                ```

        See Also:
            https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-envvars.html: Canonical
                list of the AWS environment variables this class honors.
        """
        self._client = self.create_client(configs)

    @property
    def client(self):
        """Return the cached ``boto3`` S3 client.

        Returns:
            boto3.client: The S3 client instance produced by :meth:`create_client`.
        """
        return self._client

    def create_client(self, configs: Optional[Dict] = None) -> boto3.client:
        """Build a ``boto3`` S3 client from environment credentials plus optional overrides.

        This is the hook used internally by :meth:`__init__` and re-exposed publicly so callers
        can rebuild the client (for example after rotating a signing config) without re-creating
        the :class:`S3` instance. Alternative authentication paths — IAM roles on EC2/ECS/Lambda,
        the shared credentials file written by ``aws configure``, AWS SSO — are all picked up by
        ``boto3`` itself when the corresponding env vars happen to be set to dummy values.

        Args:
            configs: Optional dictionary of extra keyword arguments to merge into the
                ``boto3.client`` call. For example, unsigned requests:
                ``{"config": Config(signature_version=botocore.UNSIGNED)}``. Keys in this dict
                override the defaults the method sets.

        Returns:
            boto3.client: A configured S3 client ready for ``upload_file`` / ``download_file`` /
            etc. calls.

        Raises:
            ValueError: If any of ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, or
                ``AWS_DEFAULT_REGION`` is not set in the environment.
            NoCredentialsError: Raised by ``boto3`` if it cannot resolve credentials.
            PartialCredentialsError: Raised by ``boto3`` if only some credentials are resolvable.

        Examples:
            - Build a client with a custom signature version and inspect the region:
                ```python
                >>> from botocore.config import Config
                >>> s3 = S3(configs={
                ...     "config": Config(signature_version="s3v4"),
                ...     "region_name": "us-west-2",
                ... })  # doctest: +SKIP
                >>> s3.client.meta.region_name  # doctest: +SKIP
                'us-west-2'

                ```
        """
        aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
        if aws_access_key_id is None:
            raise ValueError("AWS_ACCESS_KEY_ID is not set.")

        aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
        if aws_secret_access_key is None:
            raise ValueError("AWS_SECRET_ACCESS_KEY is not set.")

        region = os.getenv("AWS_DEFAULT_REGION")
        if region is None:
            raise ValueError("AWS_DEFAULT_REGION is not set.")

        # Set defaults and allow overrides through client_configs
        client_params = {
            "service_name": "s3",
            "region_name": region,
            "aws_access_key_id": aws_access_key_id,
            "aws_secret_access_key": aws_secret_access_key,
        }
        if configs:
            client_params.update(configs)

        try:
            return boto3.client(**client_params)
        except (NoCredentialsError, PartialCredentialsError) as e:
            logger.error("AWS credentials not found.")
            raise e

    def upload(self, local_path: Union[str, Path], bucket_path: str):
        """Upload a single file to S3 via the factory-level shortcut.

        Prefer :meth:`get_bucket` followed by :meth:`Bucket.upload` for anything beyond a
        one-shot file push — the bucket-level API supports recursive directory uploads and
        ``overwrite`` handling.

        Args:
            local_path: Path to the local file to upload.
            bucket_path: Destination path in the form ``"<bucket_name>/<object_key>"``. Split on
                the first ``/``.

        Raises:
            Exception: Any ``boto3.client.upload_file`` error is re-raised unchanged after being
                logged.

        Examples:
            - Upload a local file to a bucket under a named key:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> s3.upload("local/file.txt", "my-bucket/folder/file.txt")  # doctest: +SKIP

                ```
        """
        bucket_name, object_name = bucket_path.split("/", 1)
        try:
            self.client.upload_file(local_path, bucket_name, object_name)
            logger.info(f"File {local_path} uploaded to {bucket_path}.")
        except Exception as e:
            logger.error("Error uploading file to S3:", exc_info=True)
            raise e

    def download(self, bucket_path: str, local_path: Union[str, Path]):
        """Download a single object from S3 via the factory-level shortcut.

        Args:
            bucket_path: Source path in the form ``"<bucket_name>/<object_key>"``.
            local_path: Local destination path for the downloaded file.

        Raises:
            Exception: Any ``boto3.client.download_file`` error is re-raised unchanged after
                being logged.

        Examples:
            - Download a single object to a local path:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> s3.download("my-bucket/folder/file.txt", "local/file.txt")  # doctest: +SKIP

                ```
        """
        bucket_name, object_name = bucket_path.split("/", 1)
        try:
            self.client.download_file(bucket_name, object_name, local_path)
            logger.info(f"File {bucket_path} downloaded to {local_path}.")
        except Exception as e:
            logger.error("Error downloading file from S3:", exc_info=True)
            raise e

    def get_bucket(self, bucket_name: str) -> "Bucket":
        """Return a :class:`Bucket` handle for per-object operations on ``bucket_name``.

        The returned object wraps a ``boto3.resources.factory.s3.Bucket`` resource (which is a
        richer interface than the flat client) and exposes the unicloud :class:`AbstractBucket`
        surface on top of it.

        Args:
            bucket_name: The AWS S3 bucket name to look up. The method does *not* verify that
                the bucket exists; that error surfaces on the first actual operation.

        Returns:
            Bucket: A :class:`Bucket` wrapper for the named S3 bucket.

        Examples:
            - Get a bucket and list its contents:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.list_files()  # doctest: +SKIP
                ['file1.txt', 'folder/file2.txt']

                ```
        """
        s3 = boto3.resource(
            "s3",
            aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
            aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
            region_name=os.getenv("AWS_DEFAULT_REGION"),
        )
        bucket = s3.Bucket(bucket_name)
        return Bucket(bucket)

client property #

Return the cached boto3 S3 client.

Returns:

Type Description

boto3.client: The S3 client instance produced by :meth:create_client.

__init__(configs=None) #

Initialize the S3 client.

Credentials are read from the standard AWS environment variables. Any extra keyword arguments you want passed to boto3.client("s3", ...) can be supplied via configs — they override the defaults that this class sets for service_name, region_name, aws_access_key_id, and aws_secret_access_key.

Parameters:

Name Type Description Default
configs Optional[Dict]

Optional dictionary of extra keyword arguments to forward to boto3.client. Useful for passing a botocore.config.Config ({"config": Config(...)}), a non-default region ({"region_name": "us-west-2"}), or for swapping the service name in tests.

 None

Raises:

Type Description
ValueError

If any of AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, or AWS_DEFAULT_REGION is not set in the environment.
NoCredentialsError

Propagated from boto3 when it cannot find credentials to sign requests with.
PartialCredentialsError

Propagated from boto3 when only some of the required credentials are present.

Examples:

  • Default construction reading every credential from the environment: 
    >>> s3 = S3()  # doctest: +SKIP
  • Pass a custom botocore Config to override the signature version: 
    >>> from botocore.config import Config
>>> s3 = S3(configs={
...     "config": Config(signature_version="s3v4"),
...     "region_name": "us-west-2",
... })  # doctest: +SKIP
See Also

https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-envvars.html: Canonical list of the AWS environment variables this class honors.

Source code in src/unicloud/aws/aws.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def __init__(
    self,
    configs: Optional[Dict] = None,
):
    """Initialize the S3 client.

    Credentials are read from the standard AWS environment variables. Any extra keyword
    arguments you want passed to ``boto3.client("s3", ...)`` can be supplied via ``configs``
    — they override the defaults that this class sets for ``service_name``, ``region_name``,
    ``aws_access_key_id``, and ``aws_secret_access_key``.

    Args:
        configs: Optional dictionary of extra keyword arguments to forward to ``boto3.client``.
            Useful for passing a ``botocore.config.Config`` (``{"config": Config(...)}``), a
            non-default region (``{"region_name": "us-west-2"}``), or for swapping the
            service name in tests.

    Raises:
        ValueError: If any of ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, or
            ``AWS_DEFAULT_REGION`` is not set in the environment.
        NoCredentialsError: Propagated from ``boto3`` when it cannot find credentials to sign
            requests with.
        PartialCredentialsError: Propagated from ``boto3`` when only some of the required
            credentials are present.

    Examples:
        - Default construction reading every credential from the environment:
            ```python
            >>> s3 = S3()  # doctest: +SKIP

            ```
        - Pass a custom botocore Config to override the signature version:
            ```python
            >>> from botocore.config import Config
            >>> s3 = S3(configs={
            ...     "config": Config(signature_version="s3v4"),
            ...     "region_name": "us-west-2",
            ... })  # doctest: +SKIP

            ```

    See Also:
        https://docs.aws.amazon.com/cli/v1/userguide/cli-configure-envvars.html: Canonical
            list of the AWS environment variables this class honors.
    """
    self._client = self.create_client(configs)

create_client(configs=None) #

Build a boto3 S3 client from environment credentials plus optional overrides.

This is the hook used internally by :meth:__init__ and re-exposed publicly so callers can rebuild the client (for example after rotating a signing config) without re-creating the :class:S3 instance. Alternative authentication paths — IAM roles on EC2/ECS/Lambda, the shared credentials file written by aws configure, AWS SSO — are all picked up by boto3 itself when the corresponding env vars happen to be set to dummy values.

Parameters:

Name Type Description Default
configs Optional[Dict]

Optional dictionary of extra keyword arguments to merge into the boto3.client call. For example, unsigned requests: {"config": Config(signature_version=botocore.UNSIGNED)}. Keys in this dict override the defaults the method sets.

 None

Returns:

Type Description
client

boto3.client: A configured S3 client ready for upload_file / download_file /
client

etc. calls.

Raises:

Type Description
ValueError

If any of AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, or AWS_DEFAULT_REGION is not set in the environment.
NoCredentialsError

Raised by boto3 if it cannot resolve credentials.
PartialCredentialsError

Raised by boto3 if only some credentials are resolvable.

Examples:

  • Build a client with a custom signature version and inspect the region: 
    >>> from botocore.config import Config
>>> s3 = S3(configs={
...     "config": Config(signature_version="s3v4"),
...     "region_name": "us-west-2",
... })  # doctest: +SKIP
>>> s3.client.meta.region_name  # doctest: +SKIP
'us-west-2'
Source code in src/unicloud/aws/aws.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def create_client(self, configs: Optional[Dict] = None) -> boto3.client:
    """Build a ``boto3`` S3 client from environment credentials plus optional overrides.

    This is the hook used internally by :meth:`__init__` and re-exposed publicly so callers
    can rebuild the client (for example after rotating a signing config) without re-creating
    the :class:`S3` instance. Alternative authentication paths — IAM roles on EC2/ECS/Lambda,
    the shared credentials file written by ``aws configure``, AWS SSO — are all picked up by
    ``boto3`` itself when the corresponding env vars happen to be set to dummy values.

    Args:
        configs: Optional dictionary of extra keyword arguments to merge into the
            ``boto3.client`` call. For example, unsigned requests:
            ``{"config": Config(signature_version=botocore.UNSIGNED)}``. Keys in this dict
            override the defaults the method sets.

    Returns:
        boto3.client: A configured S3 client ready for ``upload_file`` / ``download_file`` /
        etc. calls.

    Raises:
        ValueError: If any of ``AWS_ACCESS_KEY_ID``, ``AWS_SECRET_ACCESS_KEY``, or
            ``AWS_DEFAULT_REGION`` is not set in the environment.
        NoCredentialsError: Raised by ``boto3`` if it cannot resolve credentials.
        PartialCredentialsError: Raised by ``boto3`` if only some credentials are resolvable.

    Examples:
        - Build a client with a custom signature version and inspect the region:
            ```python
            >>> from botocore.config import Config
            >>> s3 = S3(configs={
            ...     "config": Config(signature_version="s3v4"),
            ...     "region_name": "us-west-2",
            ... })  # doctest: +SKIP
            >>> s3.client.meta.region_name  # doctest: +SKIP
            'us-west-2'

            ```
    """
    aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
    if aws_access_key_id is None:
        raise ValueError("AWS_ACCESS_KEY_ID is not set.")

    aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
    if aws_secret_access_key is None:
        raise ValueError("AWS_SECRET_ACCESS_KEY is not set.")

    region = os.getenv("AWS_DEFAULT_REGION")
    if region is None:
        raise ValueError("AWS_DEFAULT_REGION is not set.")

    # Set defaults and allow overrides through client_configs
    client_params = {
        "service_name": "s3",
        "region_name": region,
        "aws_access_key_id": aws_access_key_id,
        "aws_secret_access_key": aws_secret_access_key,
    }
    if configs:
        client_params.update(configs)

    try:
        return boto3.client(**client_params)
    except (NoCredentialsError, PartialCredentialsError) as e:
        logger.error("AWS credentials not found.")
        raise e

upload(local_path, bucket_path) #

Upload a single file to S3 via the factory-level shortcut.

Prefer :meth:get_bucket followed by :meth:Bucket.upload for anything beyond a one-shot file push — the bucket-level API supports recursive directory uploads and overwrite handling.

Parameters:

Name Type Description Default
local_path Union[str, Path]

Path to the local file to upload.

 required
bucket_path str

Destination path in the form "<bucket_name>/<object_key>". Split on the first /.

 required

Raises:

Type Description
Exception

Any boto3.client.upload_file error is re-raised unchanged after being logged.

Examples:

  • Upload a local file to a bucket under a named key: 
    >>> s3 = S3()  # doctest: +SKIP
>>> s3.upload("local/file.txt", "my-bucket/folder/file.txt")  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def upload(self, local_path: Union[str, Path], bucket_path: str):
    """Upload a single file to S3 via the factory-level shortcut.

    Prefer :meth:`get_bucket` followed by :meth:`Bucket.upload` for anything beyond a
    one-shot file push — the bucket-level API supports recursive directory uploads and
    ``overwrite`` handling.

    Args:
        local_path: Path to the local file to upload.
        bucket_path: Destination path in the form ``"<bucket_name>/<object_key>"``. Split on
            the first ``/``.

    Raises:
        Exception: Any ``boto3.client.upload_file`` error is re-raised unchanged after being
            logged.

    Examples:
        - Upload a local file to a bucket under a named key:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> s3.upload("local/file.txt", "my-bucket/folder/file.txt")  # doctest: +SKIP

            ```
    """
    bucket_name, object_name = bucket_path.split("/", 1)
    try:
        self.client.upload_file(local_path, bucket_name, object_name)
        logger.info(f"File {local_path} uploaded to {bucket_path}.")
    except Exception as e:
        logger.error("Error uploading file to S3:", exc_info=True)
        raise e

download(bucket_path, local_path) #

Download a single object from S3 via the factory-level shortcut.

Parameters:

Name Type Description Default
bucket_path str

Source path in the form "<bucket_name>/<object_key>".

 required
local_path Union[str, Path]

Local destination path for the downloaded file.

 required

Raises:

Type Description
Exception

Any boto3.client.download_file error is re-raised unchanged after being logged.

Examples:

  • Download a single object to a local path: 
    >>> s3 = S3()  # doctest: +SKIP
>>> s3.download("my-bucket/folder/file.txt", "local/file.txt")  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def download(self, bucket_path: str, local_path: Union[str, Path]):
    """Download a single object from S3 via the factory-level shortcut.

    Args:
        bucket_path: Source path in the form ``"<bucket_name>/<object_key>"``.
        local_path: Local destination path for the downloaded file.

    Raises:
        Exception: Any ``boto3.client.download_file`` error is re-raised unchanged after
            being logged.

    Examples:
        - Download a single object to a local path:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> s3.download("my-bucket/folder/file.txt", "local/file.txt")  # doctest: +SKIP

            ```
    """
    bucket_name, object_name = bucket_path.split("/", 1)
    try:
        self.client.download_file(bucket_name, object_name, local_path)
        logger.info(f"File {bucket_path} downloaded to {local_path}.")
    except Exception as e:
        logger.error("Error downloading file from S3:", exc_info=True)
        raise e

get_bucket(bucket_name) #

Return a :class:Bucket handle for per-object operations on bucket_name.

The returned object wraps a boto3.resources.factory.s3.Bucket resource (which is a richer interface than the flat client) and exposes the unicloud :class:AbstractBucket surface on top of it.

Parameters:

Name Type Description Default
bucket_name str

The AWS S3 bucket name to look up. The method does not verify that the bucket exists; that error surfaces on the first actual operation.

 required

Returns:

Name Type Description
Bucket Bucket

A :class:Bucket wrapper for the named S3 bucket.

Examples:

  • Get a bucket and list its contents: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.list_files()  # doctest: +SKIP
['file1.txt', 'folder/file2.txt']
Source code in src/unicloud/aws/aws.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def get_bucket(self, bucket_name: str) -> "Bucket":
    """Return a :class:`Bucket` handle for per-object operations on ``bucket_name``.

    The returned object wraps a ``boto3.resources.factory.s3.Bucket`` resource (which is a
    richer interface than the flat client) and exposes the unicloud :class:`AbstractBucket`
    surface on top of it.

    Args:
        bucket_name: The AWS S3 bucket name to look up. The method does *not* verify that
            the bucket exists; that error surfaces on the first actual operation.

    Returns:
        Bucket: A :class:`Bucket` wrapper for the named S3 bucket.

    Examples:
        - Get a bucket and list its contents:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.list_files()  # doctest: +SKIP
            ['file1.txt', 'folder/file2.txt']

            ```
    """
    s3 = boto3.resource(
        "s3",
        aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
        region_name=os.getenv("AWS_DEFAULT_REGION"),
    )
    bucket = s3.Bucket(bucket_name)
    return Bucket(bucket)

unicloud.aws.aws.Bucket #

Bases: AbstractBucket

AWS S3 bucket handle — the :class:AbstractBucket implementation for S3.

Instances wrap a boto3.resources.factory.s3.Bucket and expose the unicloud contract on top of it: upload/download (files and directories), delete, list, existence-check, and rename.

Examples:

  • Prefer :meth:S3.get_bucket over constructing directly, then probe the bucket: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.file_exists("folder/file.txt")  # doctest: +SKIP
True
Source code in src/unicloud/aws/aws.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
class Bucket(AbstractBucket):
    """AWS S3 bucket handle — the :class:`AbstractBucket` implementation for S3.

    Instances wrap a ``boto3.resources.factory.s3.Bucket`` and expose the unicloud contract on top
    of it: upload/download (files and directories), delete, list, existence-check, and rename.

    Examples:
        - Prefer :meth:`S3.get_bucket` over constructing directly, then probe the bucket:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.file_exists("folder/file.txt")  # doctest: +SKIP
            True

            ```
    """

    def __init__(self, bucket):  # :boto3.resources("s3").Bucket
        """Wrap a ``boto3`` S3 Bucket resource.

        Args:
            bucket: A ``boto3.resources.factory.s3.Bucket`` instance — typically produced by
                ``boto3.resource("s3").Bucket(name)``.

        Examples:
            - Instantiate directly from a boto3 resource:
                ```python
                >>> import boto3
                >>> resource = boto3.resource("s3")  # doctest: +SKIP
                >>> bucket = Bucket(resource.Bucket("my-bucket"))  # doctest: +SKIP

                ```
            - Or let S3.get_bucket build it for you (preferred):
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP

                ```
        """
        self._bucket = bucket

    def __str__(self):
        """Return ``"Bucket: <name>"``.

        Returns:
            str: Human-readable representation including the bucket name.
        """
        return f"Bucket: {self.name}"

    def __repr__(self):
        """Return ``"Bucket: <name>"`` — same as :meth:`__str__`.

        Returns:
            str: Developer-facing representation.
        """
        return f"Bucket: {self.name}"

    @property
    def bucket(self):
        """Return the underlying ``boto3`` S3 Bucket resource.

        Exposed as an escape hatch for callers that need to drop down to the native SDK
        (for example to set lifecycle policies, which unicloud does not wrap).

        Returns:
            boto3.resources.factory.s3.Bucket: The wrapped boto3 resource.
        """
        return self._bucket

    @property
    def name(self):
        """Return the bucket name.

        Returns:
            str: The name of the wrapped S3 bucket.
        """
        return self.bucket.name

    def list_files(self, prefix: Optional[str] = None) -> List[str]:
        """List object keys in the bucket, optionally filtered by a key prefix.

        Args:
            prefix: Optional key prefix to filter the listing. Passing ``"folder/"`` lists every
                object whose key starts with ``"folder/"``. When ``None`` (the default), every
                object in the bucket is returned.

        Returns:
            List[str]: Object keys matching the prefix, in the order the SDK returns them.

        Examples:
            - List every object in the bucket:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.list_files()  # doctest: +SKIP
                ['file1.txt', 'folder/file2.txt']

                ```
            - List only objects under a folder:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.list_files(prefix="folder/")  # doctest: +SKIP
                ['folder/file2.txt']

                ```
        """
        if prefix is None:
            prefix = ""

        return [obj.key for obj in self.bucket.objects.filter(Prefix=prefix)]

    def upload(
        self, local_path: Union[str, Path], bucket_path: str, overwrite: bool = False
    ):
        """Upload a file or directory to the bucket.

        When ``local_path`` is a directory, every file beneath it is uploaded recursively and the
        relative tree under ``bucket_path`` is preserved. Empty directories raise a
        ``ValueError`` because S3 has no concept of an empty directory.

        Args:
            local_path: Path to the local file or directory to upload.
            bucket_path: Destination key (for a single file) or destination prefix (for a
                directory). Trailing ``/`` is tolerated.
            overwrite: If ``False`` (the default), uploading to a key that already exists raises
                ``ValueError``. If ``True``, the existing object is replaced silently.

        Raises:
            FileNotFoundError: If ``local_path`` does not exist.
            ValueError: If ``local_path`` is an empty directory, neither a file nor a directory,
                or already exists in the bucket while ``overwrite=False``.

        Examples:
            - Upload a single file:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.upload("local/file.txt", "folder/file.txt", overwrite=False)  # doctest: +SKIP

                ```
            - Upload a directory recursively, overwriting any conflicts:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.upload("local/dir", "remote/dir", overwrite=True)  # doctest: +SKIP

                ```
        """
        local_path = Path(local_path)
        if not local_path.exists():
            raise FileNotFoundError(f"Path {local_path} does not exist.")

        if local_path.is_file():
            self._upload_file(local_path, bucket_path, overwrite)
        elif local_path.is_dir():
            self._upload_directory(local_path, bucket_path, overwrite)
        else:
            raise ValueError(
                f"Invalid path type: {local_path} is neither a file nor a directory."
            )

    def _upload_file(self, local_path: Path, bucket_path: str, overwrite: bool):
        """Upload a single file, honoring the ``overwrite`` flag.

        Args:
            local_path: Local file to upload.
            bucket_path: Destination object key in the bucket.
            overwrite: When ``False``, raises if ``bucket_path`` already exists.

        Raises:
            ValueError: If ``bucket_path`` already exists and ``overwrite=False``.
        """
        if not overwrite and self.file_exists(bucket_path):
            raise ValueError(f"File {bucket_path} already exists in the bucket.")
        self.bucket.upload_file(Filename=str(local_path), Key=bucket_path)
        logger.info(f"File {local_path} uploaded to {bucket_path}.")

    def _upload_directory(self, local_path: Path, bucket_path: str, overwrite: bool):
        """Upload every file under ``local_path`` recursively.

        Args:
            local_path: Local directory to walk.
            bucket_path: Destination prefix in the bucket.
            overwrite: Forwarded to :meth:`_upload_file` for each uploaded file.

        Raises:
            ValueError: If ``local_path`` is empty.
        """
        if local_path.is_dir() and not any(local_path.iterdir()):
            raise ValueError(f"Directory {local_path} is empty.")

        for root, _, files in os.walk(local_path):
            for file in files:
                file_path = Path(root) / file
                relative_path = file_path.relative_to(local_path)
                s3_path = f"{bucket_path.rstrip('/')}/{relative_path.as_posix()}"
                self._upload_file(file_path, s3_path, overwrite)

    def download(
        self, bucket_path: str, local_path: Union[str, Path], overwrite: bool = False
    ):
        """Download a file or a directory from the bucket.

        A trailing ``/`` on ``bucket_path`` triggers the recursive-directory code path; anything
        else is treated as a single-object download.

        Args:
            bucket_path: Path inside the bucket to download. Trailing ``/`` means "directory".
            local_path: Local destination. For a single file this is the full filename; for a
                directory it is the directory root (created if missing).
            overwrite: If ``False`` (the default), existing local files raise ``ValueError``. If
                ``True``, they are overwritten.

        Raises:
            ValueError: If the local destination already exists with ``overwrite=False``, or if
                the bucket directory is empty.

        Examples:
            - Download a single object:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.download("folder/file.txt", "local/file.txt", overwrite=False)  # doctest: +SKIP

                ```
            - Download a directory recursively with overwrites:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.download("folder/", "local/folder/", overwrite=True)  # doctest: +SKIP

                ```
        """
        local_path = Path(local_path)
        if bucket_path.endswith("/"):
            self._download_directory(bucket_path, local_path, overwrite)
        else:
            self._download_file(bucket_path, local_path, overwrite)

    def _download_file(self, bucket_path: str, local_path: Path, overwrite: bool):
        """Download a single object, honoring the ``overwrite`` flag.

        Args:
            bucket_path: Source object key in the bucket.
            local_path: Local destination filename. Parent directories are created if missing.
            overwrite: When ``False``, raises if ``local_path`` already exists.

        Raises:
            ValueError: If ``local_path`` already exists and ``overwrite=False``.
        """
        if local_path.exists() and not overwrite:
            raise ValueError(f"File {local_path} already exists locally.")

        local_path.parent.mkdir(parents=True, exist_ok=True)

        self.bucket.download_file(Key=bucket_path, Filename=str(local_path))
        logger.info(f"File {bucket_path} downloaded to {local_path}.")

    def _download_directory(self, bucket_path: str, local_path: Path, overwrite: bool):
        """Download every object under the prefix ``bucket_path`` recursively.

        Args:
            bucket_path: Source prefix in the bucket (should end with ``/``).
            local_path: Local root directory to write into; created if missing.
            overwrite: Forwarded to :meth:`_download_file` for each file.

        Raises:
            ValueError: If the prefix yields no objects.
        """
        if not any(self.list_files(bucket_path)):
            raise ValueError(f"Directory {bucket_path} is empty.")

        local_path.mkdir(parents=True, exist_ok=True)
        for obj in self.bucket.objects.filter(Prefix=bucket_path):
            if obj.key.endswith("/"):
                continue
            relative_path = Path(obj.key).relative_to(bucket_path)
            self._download_file(obj.key, local_path / relative_path, overwrite)

    def delete(self, bucket_path: str):
        """Delete a single object or a directory (recursively) from the bucket.

        A trailing ``/`` on ``bucket_path`` triggers the recursive delete; anything else is
        treated as a single-object delete.

        Args:
            bucket_path: Object key or directory prefix to delete. Trailing ``/`` means
                "directory".

        Raises:
            ValueError: If the key does not exist (for single files) or the prefix matches
                nothing (for directories).

        Examples:
            - Delete a single file:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.delete("folder/file.txt")  # doctest: +SKIP

                ```
            - Delete every object under a prefix:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.delete("folder/")  # doctest: +SKIP

                ```
        """
        if bucket_path.endswith("/"):
            self._delete_directory(bucket_path)
        else:
            self._delete_file(bucket_path)

    def _delete_file(self, bucket_path: str):
        """Delete a single object, raising if it does not exist.

        Args:
            bucket_path: Exact object key to delete.

        Raises:
            ValueError: If no object with that exact key exists.
        """
        objects = list(self.bucket.objects.filter(Prefix=bucket_path))
        if not objects or objects[0].key != bucket_path:
            raise ValueError(f"File {bucket_path} not found in the bucket.")
        self.bucket.Object(bucket_path).delete()
        logger.info(f"Deleted: {bucket_path}")

    def _delete_directory(self, bucket_path: str):
        """Delete every object matching the prefix ``bucket_path``.

        Args:
            bucket_path: Prefix of the directory to delete (should end with ``/``).

        Raises:
            ValueError: If the prefix yields no objects.
        """
        objects = list(self.bucket.objects.filter(Prefix=bucket_path))
        if not objects:
            raise ValueError(f"No files found in the directory: {bucket_path}")

        for obj in objects:
            obj.delete()
            print(f"Deleted {obj.key}.")

    def file_exists(self, file_name: str) -> bool:
        """Return whether an exact object key exists in the bucket.

        Implemented as a ``list_objects`` prefix filter followed by an exact-match check, so it
        is safe against the common "prefix match" pitfall where ``"file"`` would also match
        ``"file-backup"``.

        Args:
            file_name: Exact object key to check.

        Returns:
            bool: ``True`` if an object with that exact key exists, ``False`` otherwise.

        Examples:
            - Check for an existing object:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.file_exists("folder/file.txt")  # doctest: +SKIP
                True

                ```
            - Check for a missing object:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.file_exists("folder/missing.txt")  # doctest: +SKIP
                False

                ```
        """
        objs = list(self.bucket.objects.filter(Prefix=file_name))
        return len(objs) > 0 and objs[0].key == file_name

    def rename(self, old_path: str, new_path: str):
        """Rename an object or directory by copy-then-delete.

        S3 has no native rename, so this method copies each matching object to the new key and
        then deletes the original. For a single object the operation is effectively atomic; for
        a directory it is *not* — a crash mid-rename leaves partial state.

        Args:
            old_path: Current object key or directory prefix. Trailing ``/`` signals a directory
                rename.
            new_path: New object key or directory prefix to rename to.

        Raises:
            ValueError: If ``old_path`` does not exist, or if ``new_path`` already exists.

        Examples:
            - Rename a single object:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.rename("old_file.txt", "new_file.txt")  # doctest: +SKIP

                ```
            - Rename a directory recursively:
                ```python
                >>> s3 = S3()  # doctest: +SKIP
                >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
                >>> bucket.rename("old_dir/", "new_dir/")  # doctest: +SKIP

                ```
        """
        # Check if the old path exists
        objects = list(self.bucket.objects.filter(Prefix=old_path))
        if not objects:
            raise ValueError(f"The path '{old_path}' does not exist in the bucket.")

        # Check if the new path already exists
        if any(self.bucket.objects.filter(Prefix=new_path)):
            raise ValueError(f"The destination path '{new_path}' already exists.")

        # Perform the rename
        for obj in objects:
            old_object_name = obj.key
            if old_path.endswith("/") and not old_object_name.startswith(old_path):
                continue  # Skip unrelated files
            new_object_name = old_object_name.replace(old_path, new_path, 1)
            # create a copy of the object to the new path
            self.bucket.Object(new_object_name).copy_from(
                CopySource={"Bucket": self.bucket.name, "Key": old_object_name}
            )
            # delete the original object
            obj.delete()

        logger.info(f"Renamed '{old_path}' to '{new_path}'.")

bucket property #

Return the underlying boto3 S3 Bucket resource.

Exposed as an escape hatch for callers that need to drop down to the native SDK (for example to set lifecycle policies, which unicloud does not wrap).

Returns:

Type Description

boto3.resources.factory.s3.Bucket: The wrapped boto3 resource.

name property #

Return the bucket name.

Returns:

Name Type Description
str

The name of the wrapped S3 bucket.

__init__(bucket) #

Wrap a boto3 S3 Bucket resource.

Parameters:

Name Type Description Default
bucket

A boto3.resources.factory.s3.Bucket instance — typically produced by boto3.resource("s3").Bucket(name).

 required

Examples:

  • Instantiate directly from a boto3 resource: 
    >>> import boto3
>>> resource = boto3.resource("s3")  # doctest: +SKIP
>>> bucket = Bucket(resource.Bucket("my-bucket"))  # doctest: +SKIP
  • Or let S3.get_bucket build it for you (preferred): 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
def __init__(self, bucket):  # :boto3.resources("s3").Bucket
    """Wrap a ``boto3`` S3 Bucket resource.

    Args:
        bucket: A ``boto3.resources.factory.s3.Bucket`` instance — typically produced by
            ``boto3.resource("s3").Bucket(name)``.

    Examples:
        - Instantiate directly from a boto3 resource:
            ```python
            >>> import boto3
            >>> resource = boto3.resource("s3")  # doctest: +SKIP
            >>> bucket = Bucket(resource.Bucket("my-bucket"))  # doctest: +SKIP

            ```
        - Or let S3.get_bucket build it for you (preferred):
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP

            ```
    """
    self._bucket = bucket

__str__() #

Return "Bucket: <name>".

Returns:

Name Type Description
str

Human-readable representation including the bucket name.
Source code in src/unicloud/aws/aws.py 
310
311
312
313
314
315
316
def __str__(self):
    """Return ``"Bucket: <name>"``.

    Returns:
        str: Human-readable representation including the bucket name.
    """
    return f"Bucket: {self.name}"

__repr__() #

Return "Bucket: <name>" — same as :meth:__str__.

Returns:

Name Type Description
str

Developer-facing representation.
Source code in src/unicloud/aws/aws.py 
318
319
320
321
322
323
324
def __repr__(self):
    """Return ``"Bucket: <name>"`` — same as :meth:`__str__`.

    Returns:
        str: Developer-facing representation.
    """
    return f"Bucket: {self.name}"

list_files(prefix=None) #

List object keys in the bucket, optionally filtered by a key prefix.

Parameters:

Name Type Description Default
prefix Optional[str]

Optional key prefix to filter the listing. Passing "folder/" lists every object whose key starts with "folder/". When None (the default), every object in the bucket is returned.

 None

Returns:

Type Description
List[str]

List[str]: Object keys matching the prefix, in the order the SDK returns them.

Examples:

  • List every object in the bucket: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.list_files()  # doctest: +SKIP
['file1.txt', 'folder/file2.txt']
  • List only objects under a folder: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.list_files(prefix="folder/")  # doctest: +SKIP
['folder/file2.txt']
Source code in src/unicloud/aws/aws.py 
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
def list_files(self, prefix: Optional[str] = None) -> List[str]:
    """List object keys in the bucket, optionally filtered by a key prefix.

    Args:
        prefix: Optional key prefix to filter the listing. Passing ``"folder/"`` lists every
            object whose key starts with ``"folder/"``. When ``None`` (the default), every
            object in the bucket is returned.

    Returns:
        List[str]: Object keys matching the prefix, in the order the SDK returns them.

    Examples:
        - List every object in the bucket:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.list_files()  # doctest: +SKIP
            ['file1.txt', 'folder/file2.txt']

            ```
        - List only objects under a folder:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.list_files(prefix="folder/")  # doctest: +SKIP
            ['folder/file2.txt']

            ```
    """
    if prefix is None:
        prefix = ""

    return [obj.key for obj in self.bucket.objects.filter(Prefix=prefix)]

upload(local_path, bucket_path, overwrite=False) #

Upload a file or directory to the bucket.

When local_path is a directory, every file beneath it is uploaded recursively and the relative tree under bucket_path is preserved. Empty directories raise a ValueError because S3 has no concept of an empty directory.

Parameters:

Name Type Description Default
local_path Union[str, Path]

Path to the local file or directory to upload.

 required
bucket_path str

Destination key (for a single file) or destination prefix (for a directory). Trailing / is tolerated.

 required
overwrite bool

If False (the default), uploading to a key that already exists raises ValueError. If True, the existing object is replaced silently.

 False

Raises:

Type Description
FileNotFoundError

If local_path does not exist.
ValueError

If local_path is an empty directory, neither a file nor a directory, or already exists in the bucket while overwrite=False.

Examples:

  • Upload a single file: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.upload("local/file.txt", "folder/file.txt", overwrite=False)  # doctest: +SKIP
  • Upload a directory recursively, overwriting any conflicts: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.upload("local/dir", "remote/dir", overwrite=True)  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def upload(
    self, local_path: Union[str, Path], bucket_path: str, overwrite: bool = False
):
    """Upload a file or directory to the bucket.

    When ``local_path`` is a directory, every file beneath it is uploaded recursively and the
    relative tree under ``bucket_path`` is preserved. Empty directories raise a
    ``ValueError`` because S3 has no concept of an empty directory.

    Args:
        local_path: Path to the local file or directory to upload.
        bucket_path: Destination key (for a single file) or destination prefix (for a
            directory). Trailing ``/`` is tolerated.
        overwrite: If ``False`` (the default), uploading to a key that already exists raises
            ``ValueError``. If ``True``, the existing object is replaced silently.

    Raises:
        FileNotFoundError: If ``local_path`` does not exist.
        ValueError: If ``local_path`` is an empty directory, neither a file nor a directory,
            or already exists in the bucket while ``overwrite=False``.

    Examples:
        - Upload a single file:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.upload("local/file.txt", "folder/file.txt", overwrite=False)  # doctest: +SKIP

            ```
        - Upload a directory recursively, overwriting any conflicts:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.upload("local/dir", "remote/dir", overwrite=True)  # doctest: +SKIP

            ```
    """
    local_path = Path(local_path)
    if not local_path.exists():
        raise FileNotFoundError(f"Path {local_path} does not exist.")

    if local_path.is_file():
        self._upload_file(local_path, bucket_path, overwrite)
    elif local_path.is_dir():
        self._upload_directory(local_path, bucket_path, overwrite)
    else:
        raise ValueError(
            f"Invalid path type: {local_path} is neither a file nor a directory."
        )

download(bucket_path, local_path, overwrite=False) #

Download a file or a directory from the bucket.

A trailing / on bucket_path triggers the recursive-directory code path; anything else is treated as a single-object download.

Parameters:

Name Type Description Default
bucket_path str

Path inside the bucket to download. Trailing / means "directory".

 required
local_path Union[str, Path]

Local destination. For a single file this is the full filename; for a directory it is the directory root (created if missing).

 required
overwrite bool

If False (the default), existing local files raise ValueError. If True, they are overwritten.

 False

Raises:

Type Description
ValueError

If the local destination already exists with overwrite=False, or if the bucket directory is empty.

Examples:

  • Download a single object: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.download("folder/file.txt", "local/file.txt", overwrite=False)  # doctest: +SKIP
  • Download a directory recursively with overwrites: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.download("folder/", "local/folder/", overwrite=True)  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
def download(
    self, bucket_path: str, local_path: Union[str, Path], overwrite: bool = False
):
    """Download a file or a directory from the bucket.

    A trailing ``/`` on ``bucket_path`` triggers the recursive-directory code path; anything
    else is treated as a single-object download.

    Args:
        bucket_path: Path inside the bucket to download. Trailing ``/`` means "directory".
        local_path: Local destination. For a single file this is the full filename; for a
            directory it is the directory root (created if missing).
        overwrite: If ``False`` (the default), existing local files raise ``ValueError``. If
            ``True``, they are overwritten.

    Raises:
        ValueError: If the local destination already exists with ``overwrite=False``, or if
            the bucket directory is empty.

    Examples:
        - Download a single object:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.download("folder/file.txt", "local/file.txt", overwrite=False)  # doctest: +SKIP

            ```
        - Download a directory recursively with overwrites:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.download("folder/", "local/folder/", overwrite=True)  # doctest: +SKIP

            ```
    """
    local_path = Path(local_path)
    if bucket_path.endswith("/"):
        self._download_directory(bucket_path, local_path, overwrite)
    else:
        self._download_file(bucket_path, local_path, overwrite)

delete(bucket_path) #

Delete a single object or a directory (recursively) from the bucket.

A trailing / on bucket_path triggers the recursive delete; anything else is treated as a single-object delete.

Parameters:

Name Type Description Default
bucket_path str

Object key or directory prefix to delete. Trailing / means "directory".

 required

Raises:

Type Description
ValueError

If the key does not exist (for single files) or the prefix matches nothing (for directories).

Examples:

  • Delete a single file: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.delete("folder/file.txt")  # doctest: +SKIP
  • Delete every object under a prefix: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.delete("folder/")  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
def delete(self, bucket_path: str):
    """Delete a single object or a directory (recursively) from the bucket.

    A trailing ``/`` on ``bucket_path`` triggers the recursive delete; anything else is
    treated as a single-object delete.

    Args:
        bucket_path: Object key or directory prefix to delete. Trailing ``/`` means
            "directory".

    Raises:
        ValueError: If the key does not exist (for single files) or the prefix matches
            nothing (for directories).

    Examples:
        - Delete a single file:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.delete("folder/file.txt")  # doctest: +SKIP

            ```
        - Delete every object under a prefix:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.delete("folder/")  # doctest: +SKIP

            ```
    """
    if bucket_path.endswith("/"):
        self._delete_directory(bucket_path)
    else:
        self._delete_file(bucket_path)

file_exists(file_name) #

Return whether an exact object key exists in the bucket.

Implemented as a list_objects prefix filter followed by an exact-match check, so it is safe against the common "prefix match" pitfall where "file" would also match "file-backup".

Parameters:

Name Type Description Default
file_name str

Exact object key to check.

 required

Returns:

Name Type Description
bool bool

True if an object with that exact key exists, False otherwise.

Examples:

  • Check for an existing object: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.file_exists("folder/file.txt")  # doctest: +SKIP
True
  • Check for a missing object: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.file_exists("folder/missing.txt")  # doctest: +SKIP
False
Source code in src/unicloud/aws/aws.py 
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
def file_exists(self, file_name: str) -> bool:
    """Return whether an exact object key exists in the bucket.

    Implemented as a ``list_objects`` prefix filter followed by an exact-match check, so it
    is safe against the common "prefix match" pitfall where ``"file"`` would also match
    ``"file-backup"``.

    Args:
        file_name: Exact object key to check.

    Returns:
        bool: ``True`` if an object with that exact key exists, ``False`` otherwise.

    Examples:
        - Check for an existing object:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.file_exists("folder/file.txt")  # doctest: +SKIP
            True

            ```
        - Check for a missing object:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.file_exists("folder/missing.txt")  # doctest: +SKIP
            False

            ```
    """
    objs = list(self.bucket.objects.filter(Prefix=file_name))
    return len(objs) > 0 and objs[0].key == file_name

rename(old_path, new_path) #

Rename an object or directory by copy-then-delete.

S3 has no native rename, so this method copies each matching object to the new key and then deletes the original. For a single object the operation is effectively atomic; for a directory it is not — a crash mid-rename leaves partial state.

Parameters:

Name Type Description Default
old_path str

Current object key or directory prefix. Trailing / signals a directory rename.

 required
new_path str

New object key or directory prefix to rename to.

 required

Raises:

Type Description
ValueError

If old_path does not exist, or if new_path already exists.

Examples:

  • Rename a single object: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.rename("old_file.txt", "new_file.txt")  # doctest: +SKIP
  • Rename a directory recursively: 
    >>> s3 = S3()  # doctest: +SKIP
>>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
>>> bucket.rename("old_dir/", "new_dir/")  # doctest: +SKIP
Source code in src/unicloud/aws/aws.py 
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
def rename(self, old_path: str, new_path: str):
    """Rename an object or directory by copy-then-delete.

    S3 has no native rename, so this method copies each matching object to the new key and
    then deletes the original. For a single object the operation is effectively atomic; for
    a directory it is *not* — a crash mid-rename leaves partial state.

    Args:
        old_path: Current object key or directory prefix. Trailing ``/`` signals a directory
            rename.
        new_path: New object key or directory prefix to rename to.

    Raises:
        ValueError: If ``old_path`` does not exist, or if ``new_path`` already exists.

    Examples:
        - Rename a single object:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.rename("old_file.txt", "new_file.txt")  # doctest: +SKIP

            ```
        - Rename a directory recursively:
            ```python
            >>> s3 = S3()  # doctest: +SKIP
            >>> bucket = s3.get_bucket("my-bucket")  # doctest: +SKIP
            >>> bucket.rename("old_dir/", "new_dir/")  # doctest: +SKIP

            ```
    """
    # Check if the old path exists
    objects = list(self.bucket.objects.filter(Prefix=old_path))
    if not objects:
        raise ValueError(f"The path '{old_path}' does not exist in the bucket.")

    # Check if the new path already exists
    if any(self.bucket.objects.filter(Prefix=new_path)):
        raise ValueError(f"The destination path '{new_path}' already exists.")

    # Perform the rename
    for obj in objects:
        old_object_name = obj.key
        if old_path.endswith("/") and not old_object_name.startswith(old_path):
            continue  # Skip unrelated files
        new_object_name = old_object_name.replace(old_path, new_path, 1)
        # create a copy of the object to the new path
        self.bucket.Object(new_object_name).copy_from(
            CopySource={"Bucket": self.bucket.name, "Key": old_object_name}
        )
        # delete the original object
        obj.delete()

    logger.info(f"Renamed '{old_path}' to '{new_path}'.")