miniopy_async package

miniopy-async - Asynchronous MinIO Client SDK for Python

>>> from miniopy_async import Minio
>>> import asyncio
>>> client = Minio(
...     "play.min.io",
...     access_key="Q3AM3UQ867SPQQA43P2F",
...     secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
...     secure=True  # http for False, https for True
... )
>>> buckets = asyncio.run(
...     client.list_buckets()
... )
>>> for bucket in buckets:
...     print(bucket.name, bucket.creation_date)

Subpackages

miniopy_async.credentials package

Submodules

miniopy_async.api module

Simple Storage Service (aka S3) client to perform bucket and object operations.

Bases: object

Simple Storage Service (aka S3) client to perform bucket and object operations.

Parameters:

endpoint (str) – Hostname of a S3 service.
access_key (str | None) – Access key (aka user ID) of your account in S3 service.
secret_key (str | None) – Secret Key (aka password) of your account in S3 service.
session_token (str | None) – Session token of your account in S3 service.
secure (bool) – Flag to indicate to use secure (TLS) connection to S3service or not.
region (str | None) – Region name of buckets in S3 service.
credentials (Provider | None) – Credentials provider of your account in S3 service.
client_session (Callable[..., aiohttp.ClientSession | aiohttp_retry.RetryClient] | None) – Custom HTTP client session caller.
cert_check (bool) – Flag to indicate to verify SSL certificate or not.
server_url (str | None) – Server url of minio service, used for presigned url.

Returns:

Minio object

Return type:

Minio

# Create client with anonymous access.
client = Minio("play.min.io")

# Create client with access and secret key.
client = Minio("s3.amazonaws.com", "ACCESS-KEY", "SECRET-KEY")

# Create client with access key and secret key with specific region.
client = Minio(
    "play.minio.io:9000",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    region="my-region",
)

async bucket_exists(bucket_name: str) → bool[source]

Check if a bucket exists.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: True if the bucket exists.
Return type:: bool
Raises:: ValueError – If bucket not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    result = await client.bucket_exists("my-bucket")
    if result:
        print("my-bucket exists")
    else:
        print("my-bucket does not exist")

asyncio.run(main())

async compose_object(bucket_name: str, object_name: str, sources: list[ComposeSource], sse: Sse | None = None, metadata: Dict[str, str | List[str] | Tuple[str]] | None = None, tags: Tags | None = None, retention: Retention | None = None, legal_hold: bool = False) → ObjectWriteResult[source]

Create an object by combining data from different source objects using server-side copy.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
sources (list[ComposeSource]) – List of ComposeSource object.
sse (Sse | None) – Server-side encryption of destination object.
metadata (DictType | None) – Any user-defined metadata to be copied along with destination object.
tags (Tags | None) – Tags for destination object.
retention (Retention | None) – Retention configuration object.
legal_hold (bool) – Flag to set legal hold for destination object.

Returns:

ObjectWriteResult object.

Return type:

ObjectWriteResult

Raises:

ValueError – If sources is not of non-empty type list or tuple.
ValueError – If sources is not of type ComposeSource.
ValueError – If tags is not of type Tags.
ValueError – If retention is not of type Retention.

from miniopy_async import Minio
from miniopy_async.commonconfig import ComposeSource
from miniopy_async.sse import SseS3
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

# Each part must larger than 5MB
sources = [
    ComposeSource("my-job-bucket", "my-object-part-one"),
    ComposeSource("my-job-bucket", "my-object-part-two"),
    ComposeSource("my-job-bucket", "my-object-part-three"),
]

async def main():
    # Create my-bucket/my-object by combining source object
    # list.
    print('example one')
    result = await client.compose_object("my-bucket",
    "my-object", sources)
    print(result.object_name, result.version_id)

    # Create my-bucket/my-object with user metadata by combining
    # source object list.
    print('example two')
    result = await client.compose_object(
        "my-bucket",
        "my-object",
        sources,
        metadata={"Content-Type": "application/octet-stream"},
    )
    print(result.object_name, result.version_id)

    # Create my-bucket/my-object with user metadata and
    # server-side encryption by combining source object list.
    print('example three')
    result = await client.compose_object(
        "my-bucket",
        "my-object",
        sources,
        sse=SseS3()
    )
    print(result.object_name, result.version_id)

asyncio.run(main())

Create an object by server-side copying data from another object. In this API maximum supported source object size is 5GiB.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
source (CopySource) – CopySource object.
sse (Sse | None) – Server-side encryption of destination object.
metadata (DictType | None) – Any user-defined metadata to be copied along with destination object.
tags (Tags | None) – Tags for destination object.
retention (Retention | None) – Retention configuration object.
legal_hold (bool) – Flag to set legal hold for destination object.
metadata_directive (str | None) – Directive used to handle user metadata for destination object.
tagging_directive (str | None) – Directive used to handle tags for destination object.

Returns:

ObjectWriteResult object.

Return type:

ObjectWriteResult

Raises:

ValueError – If source is not of type CopySource.
ValueError – If tags is not of type Tags.
ValueError – If retention is not of type Retention.
ValueError – If metadata_directive is not of type COPY or REPLACE.
ValueError – If tagging_directive is not of type COPY or REPLACE.
ValueError – If source object size is greater than 5GiB.

from datetime import datetime, timezone
from miniopy_async import Minio
from miniopy_async.commonconfig import REPLACE, CopySource
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # copy an object from a bucket to another.
    print("example one")
    result = await client.copy_object(
        "my-job-bucket",
        "my-copied-object1",
        CopySource("my-bucket", "my-object"),
    )
    print(result.object_name, result.version_id)

    # copy an object with condition.
    print("example two")
    result = await client.copy_object(
        "my-job-bucket",
        "my-copied-object2",
        CopySource(
            "my-bucket",
            "my-object",
            modified_since=datetime(2014, 4, 1,
            tzinfo=timezone.utc),
        ),
    )
    print(result.object_name, result.version_id)

    # copy an object from a bucket with replacing metadata.
    print("example three")
    metadata = {"Content-Type": "application/octet-stream"}
    result = await client.copy_object(
        "my-job-bucket",
        "my-copied-object3",
        CopySource("my-bucket", "my-object"),
        metadata=metadata,
        metadata_directive=REPLACE,
    )
    print(result.object_name, result.version_id)

asyncio.run(main())

async delete_bucket_encryption(bucket_name: str)[source]

Delete encryption configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Raises:: S3Error – If server side encryption configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_encryption("my-bucket")

asyncio.run(main())

async delete_bucket_lifecycle(bucket_name: str)[source]

Delete notification configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_lifecycle("my-bucket")

asyncio.run(main())

async delete_bucket_notification(bucket_name: str)[source]

Delete notification configuration of a bucket. On success, S3 service stops notification of events previously set of the bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_notification("my-bucket")

asyncio.run(main())

async delete_bucket_policy(bucket_name: str)[source]

Delete bucket policy configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_policy("my-bucket")

asyncio.run(main())

async delete_bucket_replication(bucket_name: str)[source]

Delete replication configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_replication("my-bucket")

asyncio.run(main())

async delete_bucket_tags(bucket_name: str)[source]

Delete tags configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_bucket_tags("my-bucket")

asyncio.run(main())

async delete_object_lock_config(bucket_name: str)[source]

Delete object-lock configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_object_lock_config("my-bucket")

asyncio.run(main())

async delete_object_tags(bucket_name: str, object_name: str, version_id: str | None = None)[source]

Delete tags configuration of an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the Object.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.delete_object_tags("my-bucket", "my-object")

asyncio.run(main())

disable_accelerate_endpoint()[source]: Disables accelerate endpoint for Amazon S3 endpoint.

disable_dualstack_endpoint()[source]: Disables dualstack endpoint for Amazon S3 endpoint.

async disable_object_legal_hold(bucket_name: str, object_name: str, version_id: str | None = None)[source]

Disable legal hold on an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the object.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.disable_object_legal_hold("my-bucket", "my-object")

asyncio.run(main())

disable_virtual_style_endpoint()[source]: Disables virtual style endpoint.

enable_accelerate_endpoint()[source]: Enables accelerate endpoint for Amazon S3 endpoint.

enable_dualstack_endpoint()[source]: Enables dualstack endpoint for Amazon S3 endpoint.

async enable_object_legal_hold(bucket_name: str, object_name: str, version_id: str | None = None)[source]

Enable legal hold on an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the object.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.enable_object_legal_hold("my-bucket", "my-object")

asyncio.run(main())

enable_virtual_style_endpoint()[source]: Enables virtual style endpoint.

Downloads data of an object to file.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
file_path (str) – Name of file to download.
request_headers (DictType | None) – Any additional headers to be added with GET request.
ssec (SseCustomerKey | None) – Server-side encryption customer key.
version_id (str | None) – Version-ID of the object.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.
tmp_file_path (str | None) – Path to a temporary file.
progress (ProgressType | None) – A progress object.
session (aiohttp.ClientSession | aiohttp_retry.RetryClient | None) – aiohttp.ClientSession() object.

Returns:

Object information.

Return type:

Object

Raises:

ValueError – If file_path is a directory.

from miniopy_async import Minio
from miniopy_async.sse import SseCustomerKey
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Download data of an object.
    print("example one")
    await client.fget_object("my-bucket", "my-object",
    "my-filename")

    # Download data of an object of version-ID.
    print("example two")
    await client.fget_object(
        "my-bucket", "my-object", "my-filename",
        version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
    )

    # Download data of an SSE-C encrypted object.
    print("example three")
    await client.fget_object(
        "my-bucket", "my-object", "my-filename",
        ssec=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
    )

asyncio.run(main())

async fput_object(bucket_name: str, object_name: str, file_path: str, content_type: str = 'application/octet-stream', metadata: Dict[str, str | List[str] | Tuple[str]] | None = None, sse: Sse | None = None, progress: ProgressType | None = None, part_size: int = 0, num_parallel_uploads: int = 3, tags: Tags | None = None, retention: Retention | None = None, legal_hold: bool = False) → ObjectWriteResult[source]

Uploads data from a file to an object in a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
file_path (str) – Name of file to upload.
content_type (str) – Content type of the object.
metadata (DictType | None) – Any additional metadata to be uploaded along with your PUT request.
sse (Sse | None) – Server-side encryption.
progress (ProgressType | None) – A progress object.
part_size (int) – Multipart part size
num_parallel_uploads (int) – Number of parallel uploads.
tags (Tags | None) – Tags for the object.
retention (Retention | None) – Retention configuration object.
legal_hold (bool) – Flag to set legal hold for the object.

Returns:

ObjectWriteResult object.

Return type:

ObjectWriteResult

from datetime import datetime, timedelta
from miniopy_async import Minio
from miniopy_async.commonconfig import GOVERNANCE, Tags
from miniopy_async.retention import Retention
from miniopy_async.sse import SseCustomerKey, SseKMS, SseS3
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Upload data.
    print("example one")
    await client.fput_object(
        "my-bucket", "my-object1", "my-filename",
    )

    # Upload data with content-type.
    print("example two")
    await client.fput_object(
        "my-bucket", "my-object2", "my-filename",
        content_type="application/octet-stream",
    )

    # Upload data with metadata.
    print("example three")
    await client.fput_object(
        "my-bucket", "my-object3", "my-filename",
        metadata={"Content-Type": "application/octet-stream"},
    )

    # Upload data with customer key type of server-side encryption.
    print("example four")
    await client.fput_object(
        "my-bucket", "my-object4", "my-filename",
        sse=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
    )

    # Upload data with KMS type of server-side encryption.
    print("example five")
    await client.fput_object(
        "my-bucket", "my-object5", "my-filename",
        sse=SseKMS("KMS-KEY-ID", {"Key1": "Value1", "Key2":
        "Value2"}),
    )

    # Upload data with S3 type of server-side encryption.
    print("example six")
    await client.fput_object(
        "my-bucket", "my-object6", "my-filename",
        sse=SseS3(),
    )

    # Upload data with tags, retention and legal-hold.
    print("example seven")
    date = datetime.utcnow().replace(
        hour=0, minute=0, second=0, microsecond=0,
    ) + timedelta(days=30)
    tags = Tags(for_object=True)
    tags["User"] = "jsmith"
    await client.fput_object(
        "my-bucket", "my-object7", "my-filename",
        tags=tags,
        retention=Retention(GOVERNANCE, date),
        legal_hold=True,
    )

asyncio.run(main())

async get_bucket_encryption(bucket_name: str) → SSEConfig | None[source]

Get encryption configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: SSEConfig object.
Return type:: SSEConfig | None
Raises:: S3Error – If server side encryption configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_bucket_encryption("my-bucket")
    print(config)

asyncio.run(main())

async get_bucket_lifecycle(bucket_name: str) → LifecycleConfig | None[source]

Get bucket lifecycle configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: LifecycleConfig object.
Return type:: LifecycleConfig | None
Raises:: S3Error – If the lifecycle configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_bucket_lifecycle("my-bucket")
    print(config)

asyncio.run(main())

async get_bucket_notification(bucket_name: str) → NotificationConfig[source]

Get notification configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: NotificationConfig object.
Return type:: NotificationConfig

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_bucket_notification("my-bucket")
    print(config)

asyncio.run(main())

async get_bucket_policy(bucket_name: str) → str[source]

Get bucket policy configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: Bucket policy configuration as JSON string.
Return type:: str

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    policy = await client.get_bucket_policy("my-bucket")
    print(policy)

asyncio.run(main())

async get_bucket_replication(bucket_name: str) → ReplicationConfig | None[source]

Get bucket replication configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: ReplicationConfig object.
Return type:: ReplicationConfig | None
Raises:: S3Error – If the replication configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_bucket_replication("my-bucket")
    print(config)

asyncio.run(main())

async get_bucket_tags(bucket_name: str) → Tags | None[source]

Get tags configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: Tags object.
Return type:: Tags | None
Raises:: S3Error – If the tags configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    tags = await client.get_bucket_tags("my-bucket")
    print(tags)

asyncio.run(main())

async get_bucket_versioning(bucket_name: str) → VersioningConfig[source]

Get versioning configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: VersioningConfig.
Return type:: VersioningConfig

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_bucket_versioning("my-bucket")
    print(config.status)

asyncio.run(main())

Get data of an object. Returned response should be closed after use to release network resources. To reuse the connection, it’s required to call response.release() explicitly.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
session (aiohttp.ClientSession | aiohttp_retry.RetryClient) – aiohttp.ClientSession() object.
offset (int) – Start byte position of object data.
length (int) – Number of bytes of object data from offset.
request_headers (DictType | None) – Any additional headers to be added with GET request.
ssec (SseCustomerKey | None) – Server-side encryption customer key.
version_id (str | None) – Version-ID of the object.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.

Returns:

aiohttp.client_reqrep.ClientResponse object.

Return type:

aiohttp.ClientResponse

from miniopy_async import Minio
from miniopy_async.sse import SseCustomerKey
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Get data of an object.
    print('example one')
    async with aiohttp.ClientSession() as session:
        response = await client.get_object("my-bucket", "my-object", session)
        # Read data from response.

    # Get data of an object from offset and length.
    print('example two')
    async with aiohttp.ClientSession() as session:
        response = await client.get_object(
            "my-bucket", "my-object", session, offset=512, length=1024,
        )
        # Read data from response.

    # Get data of an object of version-ID.
    print('example three')
    async with aiohttp.ClientSession() as session:
        response = await client.get_object(
            "my-bucket", "my-object", sessioin
            version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
        )
        # Read data from response.

    # Get data of an SSE-C encrypted object.
    print('example four')
    async with aiohttp.ClientSession() as session:
        response = await client.get_object(
            "my-bucket", "my-object", session
            ssec=SseCustomerKey(
            b"32byteslongsecretkeymustprovided"),
        )
        # Read data from response.

asyncio.run(main())

async get_object_lock_config(bucket_name: str) → ObjectLockConfig[source]

Get object-lock configuration of a bucket.

Parameters:: bucket_name (str) – Name of the bucket.
Returns:: ObjectLockConfig object.
Return type:: ObjectLockConfig

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_object_lock_config("my-bucket")
    print(config)

asyncio.run(main())

async get_object_retention(bucket_name: str, object_name: str, version_id: str | None = None) → Retention | None[source]

Get retention configuration of an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the object.

Returns:

Retention object.

Return type:

Retention | None

Raises:

S3Error – If the object lock configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    config = await client.get_object_retention("my-bucket",
    "my-object")
    print(config)

asyncio.run(main())

async get_object_tags(bucket_name: str, object_name: str, version_id: str | None = None) → Tags | None[source]

Get tags configuration of a object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the Object.

Returns:

Tags object.

Return type:

Tags | None

Raises:

S3Error – If the tags configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    tags = await client.get_object_tags("my-bucket", "my-object")
    print(tags)

asyncio.run(main())

Get presigned URL of an object for HTTP method, expiry time and custom request parameters.

Parameters:

method (str) – HTTP method.
bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
expires (datetime.timedelta) – Expiry in seconds; defaults to 7 days.
response_headers (DictType | None) – Optional response_headers argument to specify response fields like date, size, type of file, data about server, etc.
request_date (datetime.timedelta) – Optional request_date argument to specify a different request date. Default is current date.
version_id (str | None) – Version ID of the object.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.

Returns:

URL string.

Return type:

str

Raises:

ValueError – If expires is not between 1 second to 7 days.

from datetime import timedelta
from miniopy_async import Minio
import asyncio

async def main():
    client = Minio(
        "play.min.io",
        access_key="Q3AM3UQ867SPQQA43P2F",
        secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
        secure=True  # http for False, https for True
    )
    # Get presigned URL string to delete 'my-object' in
    # 'my-bucket' with one day expiry.
    print('example one')
    url = await client.get_presigned_url(
        "DELETE",
        "my-bucket",
        "my-object",
        expires=timedelta(days=1),
    )
    print('url:', url)

    # Get presigned URL string to upload 'my-object' in
    # 'my-bucket' with response-content-type as application/json
    # and one day expiry.
    print('example two')
    url = await client.get_presigned_url(
        "PUT",
        "my-bucket",
        "my-object",
        expires=timedelta(days=1),
        response_headers={"response-content-type":
        "application/json"},
    )
    print('url:', url)

    # Get presigned URL string to download 'my-object' in
    # 'my-bucket' with two hours expiry.
    print('example three')
    url = await client.get_presigned_url(
        "GET",
        "my-bucket",
        "my-object",
        expires=timedelta(hours=2),
    )
    print('url:', url)


asyncio.run(main())

async is_object_legal_hold_enabled(bucket_name: str, object_name: str, version_id: str | None = None) → bool[source]

Returns true if legal hold is enabled on an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the object.

Returns:

Whether the legal hold is enabled or not.

Return type:

bool

Raises:

S3Error – If the object lock configuration is not found.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    if await client.is_object_legal_hold_enabled("my-bucket",
    "my-object"):
        print("legal hold is enabled on my-object")
    else:
        print("legal hold is not enabled on my-object")

asyncio.run(main())

async list_buckets() → list[Bucket][source]

List information of all accessible buckets.

Returns:: List of Bucket object.
Return type:: list[Bucket]

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    buckets = await client.list_buckets()
    for bucket in buckets:
        print(bucket.name, bucket.creation_date)

asyncio.run(main())

list_objects(bucket_name: str, prefix: str | None = None, recursive: bool = False, start_after: str | None = None, include_user_meta: bool = False, include_version: bool = False, use_api_v1: bool = False, use_url_encoding_type: bool = True, fetch_owner: bool = False, extra_headers: Dict[str, str | List[str] | Tuple[str]] | None = None, extra_query_params: Dict[str, str | List[str] | Tuple[str]] | None = None) → ListObjects[source]

Lists object information of a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
prefix (str | None) – Object name starts with prefix.
recursive (bool) – List recursively than directory structure emulation.
start_after (str | None) – List objects after this key name.
include_user_meta (bool) – MinIO specific flag to control to include user metadata.
include_version (bool) – Flag to control whether include object versions.
use_api_v1 (bool) – Flag to control to use ListObjectV1 S3 API or not.
use_url_encoding_type (bool) – Flag to control whether URL encoding type to be used or not.
fetch_owner (bool) – Flag to control whether to fetch owner information.
extra_headers (DictType | None) – Any additional headers to be added with GET request.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.

Returns:

Iterator of Object.

Return type:

ListObjects

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # List objects information.
    print('example one')
    objects = await client.list_objects("my-bucket")
    for obj in objects:
        print('obj:',obj)

    async for obj in client.list_objects("my-bucket"):
        print('obj:',obj)

    # List objects information whose names starts with "my/prefix/".
    print('example two')
    objects = await client.list_objects("my-bucket",
    prefix="my/prefix/")
    for obj in objects:
        print('obj:',obj)

    # List objects information recursively.
    print('example three')
    objects = await client.list_objects("my-bucket", recursive=True)
    for obj in objects:
        print('obj:',obj)

    # List objects information recursively whose names starts with
    # "my/prefix/".
    print('example four')
    objects = await client.list_objects(
        "my-bucket", prefix="my/prefix/", recursive=True,
    )
    for obj in objects:
        print('obj:',obj)

    # List objects information recursively after object name
    # "my/prefix/world/1".
    print('example five')
    objects = await client.list_objects(
        "my-bucket", recursive=True,
        start_after="my/prefix/world/1",
    )
    for obj in objects:
        print('obj:',obj)

asyncio.run(main())

async listen_bucket_notification(bucket_name: str, prefix: str = '', suffix: str = '', events: tuple[str, ...] = ('s3:ObjectCreated:*', 's3:ObjectRemoved:*', 's3:ObjectAccessed:*')) → AsyncEventIterable[source]

Listen events of object prefix and suffix of a bucket. Caller should iterate returned iterator to read new events.

Parameters:

bucket_name (str) – Name of the bucket.
prefix (str) – Listen events of object starts with prefix.
suffix (str) – Listen events of object ends with suffix.
events (tuple[str, ...]) – Events to listen.

Returns:

Iterator of event records as dict.

Return type:

AsyncEventIterable[aiohttp.ClientResponse]

Raises:

ValueError – If ListenBucketNotification API is not supported in Amazon S3.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    events = await client.listen_bucket_notification(
        "my-bucket",
        prefix="my-prefix/",
        events=("s3:ObjectCreated:*", "s3:ObjectRemoved:*"),
    )
    async for event in events:
        print('event:',event)

asyncio.run(main())

async make_bucket(bucket_name: str, location: str | None = None, object_lock: bool = False)[source]

Create a bucket with region and object lock.

Parameters:

bucket_name (str) – Name of the bucket.
location (str | None) – Region in which the bucket will be created.
object_lock (bool) – Flag to set object-lock feature.

Raises:

ValueError – If location is not same as region passed via base url

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True
)

async def main():
    # Create bucket.
    print('example one')
    await client.make_bucket("my-bucket1")

    # Create bucket on specific region.
    print('example two')
    await client.make_bucket("my-bucket2", "us-east-1")

    # Create bucket with object-lock feature on specific region.
    print('example three')
    await client.make_bucket("my-bucket3", "us-east-1",
    object_lock=True)

asyncio.run(main())

Get presigned URL of an object to download its data with expiry time and custom request parameters.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
expires (datetime.timedelta) – Expiry in seconds; defaults to 7 days.
response_headers (DictType | None) – Optional response_headers argument to specify response fields like date, size, type of file, data about server, etc.
request_date (datetime.timedelta) – Optional request_date argument to specify a different request date. Default is current date.
version_id (str | None) – Version ID of the object.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.

Returns:

URL string.

Return type:

str

from datetime import timedelta
from miniopy_async import Minio
import asyncio

async def main():
    client = Minio(
        "play.min.io",
        access_key="Q3AM3UQ867SPQQA43P2F",
        secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
        secure=True  # http for False, https for True
    )

    # Get presigned URL string to download 'my-object' in
    # 'my-bucket' with default expiry (i.e. 7 days).
    print('example one')
    url = await client.presigned_get_object("my-bucket",
    "my-object")
    print('url:', url)

    # Get presigned URL string to download 'my-object' in
    # 'my-bucket' with two hours expiry.
    print('example two')
    url = await client.presigned_get_object(
        "my-bucket", "my-object", expires=timedelta(hours=2),
    )
    print('url:', url)


asyncio.run(main())

async presigned_post_policy(policy: PostPolicy) → dict[str, str][source]

Get form-data of PostPolicy of an object to upload its data using POST method.

Parameters:

policy (PostPolicy) – PostPolicy.

Returns:

dict contains form-data.

Return type:

dict[str, str]

Raises:

ValueError – If policy is not of type PostPolicy.
ValueError – If the access is anonymous.

from datetime import datetime, timedelta
from miniopy_async import Minio
from miniopy_async.datatypes import PostPolicy
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

policy = PostPolicy(
    "my-bucket", datetime.utcnow() + timedelta(days=10),
)
policy.add_starts_with_condition("key", "my/object/prefix/")
policy.add_content_length_range_condition(1*1024*1024, 10*1024*1024)

async def main():
    form_data = await client.presigned_post_policy(policy)
    curl_cmd = (
        "curl -X POST "
        "https://play.min.io/my-bucket "
        "{0} -F file=@<FILE>"
    ).format(
        " ".join(["-F {0}={1}".format(k, v) for k,
        v in form_data.items()]),
    )
    print('curl_cmd:',curl_cmd)

asyncio.run(main())

async presigned_put_object(bucket_name: str, object_name: str, expires: timedelta = datetime.timedelta(days=7)) → str[source]

Get presigned URL of an object to upload data with expiry time and custom request parameters.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
expires (datetime.timedelta) – Expiry in seconds; defaults to 7 days.

Returns:

URL string.

Return type:

str

from datetime import timedelta
from miniopy_async import Minio
import asyncio

async def main():
    client = Minio(
        "play.min.io",
        access_key="Q3AM3UQ867SPQQA43P2F",
        secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
        secure=True  # http for False, https for True
    )

    # Get presigned URL string to upload data to 'my-object' in
    # 'my-bucket' with default expiry (i.e. 7 days).
    url = await client.presigned_put_object("my-bucket",
    "my-object")
    print('url:', url)

    # Get presigned URL string to upload data to 'my-object' in
    # 'my-bucket' with two hours expiry.
    url = await client.presigned_put_object(
        "my-bucket", "my-object", expires=timedelta(hours=2),
    )
    print('url:', url)

asyncio.run(main())

Prompt an object using natural language.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
prompt (str) – Prompt the Object to interact with the AI model. request.
lambda_arn (str | None) – Lambda ARN to use for prompt.
request_headers (DictType | None) – Any additional headers to be added with POST
ssec (SseCustomerKey | None) – Server-side encryption customer key.
version_id (str | None) – Version-ID of the object.
kwargs (Any | None) – Extra parameters for advanced usage.

Returns:

aiohttp.ClientResponse object.

Return type:

aiohttp.ClientResponse

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    response = await self.client.prompt_object(
        "ai-data",
        "receipt_from_la.png",
        "What is the address of the restaurant?"
        stream=False
    )

    print(await response.text())

asyncio.run(main())

async put_object(bucket_name: str, object_name: str, data: BinaryIO | FileIOWrapperBase, length: int, content_type: str = 'application/octet-stream', metadata: Dict[str, str | List[str] | Tuple[str]] | None = None, sse: Sse | None = None, progress: ProgressType | None = None, part_size: int = 0, num_parallel_uploads: int = 3, tags: Tags | None = None, retention: Retention | None = None, legal_hold: bool = False) → ObjectWriteResult[source]

Uploads data from a stream to an object in a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
data (BinaryIO | FileIOWrapperBase) – An object having callable read() returning bytes object.
length (int) – Data size; -1 for unknown size and set valid part_size.
content_type (str) – Content type of the object.
metadata (DictType | None) – Any additional metadata to be uploaded along with your PUT request.
sse (Sse | None) – Server-side encryption.
progress (ProgressType | None) – A progress object.
part_size (int) – Multipart part size.
num_parallel_uploads (int) – Number of parallel uploads.
tags (Tags | None) – Tags for the object.
retention (Retention | None) – Retention configuration object.
legal_hold (bool) – Flag to set legal hold for the object.

Returns:

ObjectWriteResult object.

Return type:

ObjectWriteResult

Raises:

ValueError – If tags is not of type Tags.
ValueError – If retention is not of type Retention.
ValueError – If input data doesn’t have callable read().
IOError – If input data doesn’t have enough data to read.

import io
from datetime import datetime, timedelta
from urllib.request import urlopen
from miniopy_async import Minio
from miniopy_async.commonconfig import GOVERNANCE, Tags
from miniopy_async.retention import Retention
from miniopy_async.sse import SseCustomerKey, SseKMS, SseS3
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Upload data.
    print('example one')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
    )

    # Upload unknown sized data.
    print('example two')
    data = urlopen(
        "https://raw.githubusercontent.com/hlf20010508/miniopy
        -async/master/README.md",
    )
    await client.put_object(
        "my-bucket", "my-object", data, length=-1,
        part_size=10*1024*1024,
    )

    # Upload data with content-type.
    print('example three')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        content_type="application/csv",
    )

    # Upload data with metadata.
    print('example four')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        metadata={"Content-Type": "application/octet-stream"},
    )

    # Upload data with customer key type of server-side encryption.
    print('example five')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        sse=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
    )

    # Upload data with KMS type of server-side encryption.
    print('example six')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        sse=SseKMS("KMS-KEY-ID", {"Key1": "Value1", "Key2":
        "Value2"}),
    )

    # Upload data with S3 type of server-side encryption.
    print('example seven')
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        sse=SseS3(),
    )

    # Upload data with tags, retention and legal-hold.
    print('example eight')
    date = datetime.utcnow().replace(
        hour=0, minute=0, second=0, microsecond=0,
    ) + timedelta(days=30)
    tags = Tags(for_object=True)
    tags["User"] = "jsmith"
    await client.put_object(
        "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
        tags=tags,
        retention=Retention(GOVERNANCE, date),
        legal_hold=True,
    )

asyncio.run(main())

async remove_bucket(bucket_name: str)[source]

Remove an empty bucket.

Parameters:: bucket_name (str) – Name of the bucket.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.remove_bucket("my-bucket")

asyncio.run(main())

async remove_object(bucket_name: str, object_name: str, version_id: str | None = None)[source]

Remove an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (str | None) – Version ID of the object.

from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Remove object.
    print('example one')
    await client.remove_object("my-bucket", "my-object")

    # Remove version of an object.
    print('example two')
    await client.remove_object(
        "my-bucket", "my-object",
        version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
    )

asyncio.run(main())

remove_objects(bucket_name: str, delete_object_list: Iterable[DeleteObject], bypass_governance_mode: bool = False) → DeleteErrors[source]

Remove multiple objects.

Parameters:

bucket_name (str) – Name of the bucket.
delete_object_list (Iterable[DeleteObject]) – An iterable containing DeleteObject object.
bypass_governance_mode (bool) – Bypass Governance retention mode.

Returns:

A DeleteErrors object to be an async generator and list.

Return type:

DeleteErrors

from miniopy_async import Minio
from miniopy_async.deleteobjects import DeleteObject
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Remove list of objects.
    print('example one')
    errors = await client.remove_objects(
        "my-bucket",
        [
            DeleteObject("my-object1"),
            DeleteObject("my-object2"),
            DeleteObject("my-object3",
            "13f88b18-8dcd-4c83-88f2-8631fdb6250c"),
        ],
    )
    for error in errors:
        print("error occured when deleting object", error)

    # As async generator
    print('example two')
    async for error in client.remove_objects(
        "my-bucket",
        [
            DeleteObject("my-object1"),
            DeleteObject("my-object2"),
            DeleteObject("my-object3",
            "13f88b18-8dcd-4c83-88f2-8631fdb6250c"),
        ],
    ):
        print("error occured when deleting object", error)

    # Remove a prefix recursively.
    print('example three')
    delete_object_list = [DeleteObject(obj.object_name)
        for obj in await client.list_objects(
            "my-bucket",
            "my/prefix/",
            recursive=True
        )
    ]
    errors = await client.remove_objects("my-bucket", delete_object_list)
    for error in errors:
        print("error occured when deleting object", error)

asyncio.run(main())

async select_object_content(bucket_name: str, object_name: str, request: SelectRequest) → SelectObjectReader[source]

Select content of an object by SQL expression.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
request (SelectRequest) – SelectRequest object.

Returns:

A reader contains requested records and progress information.

Return type:

SelectObjectReader

Raises:

ValueError – If request is not of type SelectRequest.

from miniopy_async import Minio
from miniopy_async.select import (CSVInputSerialization,
CSVOutputSerialization, SelectRequest)
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    result = await client.select_object_content(
        "my-bucket",
        "my-object.csv",
        SelectRequest(
            "select * from s3object",
            CSVInputSerialization(),
            CSVOutputSerialization(),
            request_progress=True,
        ),
    )
    print('data:')
    async for data in result.stream():
        print(data.decode())
    print('status:',result.stats())

asyncio.run(main())

set_app_info(app_name: str, app_version: str)[source]

Set your application name and version to user agent header.

Parameters:

app_name (str) – Application name.
app_version (str) – Application version.

Raises:

ValueError – If application name or version is empty.

client.set_app_info('my_app', '1.0.2')

async set_bucket_encryption(bucket_name: str, config: SSEConfig)[source]

Set encryption configuration of a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (SSEConfig) – SSEConfig object.

Raises:

ValueError – If config is not of type SSEConfig.

from miniopy_async import Minio
from miniopy_async.sseconfig import Rule, SSEConfig
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.set_bucket_encryption(
        "my-bucket", SSEConfig(Rule.new_sse_s3_rule()),
    )

asyncio.run(main())

async set_bucket_lifecycle(bucket_name: str, config: LifecycleConfig)[source]

Set bucket lifecycle configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (LifecycleConfig) – LifecycleConfig object.

Raises:

ValueError – If config is not of type LifecycleConfig.

from miniopy_async import Minio
from miniopy_async.commonconfig import ENABLED, Filter
from miniopy_async.lifecycleconfig import Expiration,
LifecycleConfig, Rule, Transition
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

config = LifecycleConfig(
    [
        Rule(
            ENABLED,
            rule_filter=Filter(prefix="documents/"),
            rule_id="rule1",
            transition=Transition(days=30, storage_class="GLACIER"),
        ),
        Rule(
            ENABLED,
            rule_filter=Filter(prefix="logs/"),
            rule_id="rule2",
            expiration=Expiration(days=365),
        ),
    ],
)

async def main():
    await client.set_bucket_lifecycle("my-bucket", config)

asyncio.run(main())

async set_bucket_notification(bucket_name: str, config: NotificationConfig)[source]

Set notification configuration of a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (NotificationConfig) – class:NotificationConfig object.

Raises:

ValueError – If config is not of type NotificationConfig.

from miniopy_async import Minio
from miniopy_async.notificationconfig import (NotificationConfig,
PrefixFilterRule, QueueConfig)
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

config = NotificationConfig(
    queue_config_list=[
        QueueConfig(
            "QUEUE-ARN-OF-THIS-BUCKET",
            ["s3:ObjectCreated:*"],
            config_id="1",
            prefix_filter_rule=PrefixFilterRule("abc"),
        ),
    ],
)

async def main():
    await client.set_bucket_notification("my-bucket", config)

asyncio.run(main())

async set_bucket_policy(bucket_name: str, policy: str | bytes)[source]

Set bucket policy configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
policy (str | bytes) – Bucket policy configuration as JSON string.

import json
from miniopy_async import Minio
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Example anonymous read-only bucket policy.
    print('example one')
    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": ["s3:GetBucketLocation", "s3:ListBucket"],
                "Resource": "arn:aws:s3:::my-bucket",
            },
            {
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": "s3:GetObject",
                "Resource": "arn:aws:s3:::my-bucket/*",
            },
        ],
    }
    await client.set_bucket_policy("my-bucket", json.dumps(policy))

    # Example anonymous read-write bucket policy.
    print('example two')
    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": [
                    "s3:GetBucketLocation",
                    "s3:ListBucket",
                    "s3:ListBucketMultipartUploads",
                ],
                "Resource": "arn:aws:s3:::my-bucket",
            },
            {
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": [
                    "s3:GetObject",
                    "s3:PutObject",
                    "s3:DeleteObject",
                    "s3:ListMultipartUploadParts",
                    "s3:AbortMultipartUpload",
                ],
                "Resource": "arn:aws:s3:::my-bucket/images/*",
            },
        ],
    }
    await client.set_bucket_policy("my-bucket", json.dumps(policy))

asyncio.run(main())

async set_bucket_replication(bucket_name: str, config: ReplicationConfig)[source]

Set bucket replication configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (ReplicationConfig) – ReplicationConfig object.

Raises:

ValueError – If config is not of type ReplicationConfig.

from miniopy_async import Minio
from miniopy_async.commonconfig import DISABLED, ENABLED, AndOperator, Filter, Tags
from miniopy_async.replicationconfig import (DeleteMarkerReplication, Destination, ReplicationConfig, Rule)
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

bucket_tags = Tags.new_bucket_tags()
bucket_tags["Project"] = "Project One"
bucket_tags["User"] = "jsmith"

config = ReplicationConfig(
    "REPLACE-WITH-ACTUAL-ROLE",
    [
        Rule(
            Destination(
                "REPLACE-WITH-ACTUAL-DESTINATION-BUCKET-ARN",
            ),
            ENABLED,
            delete_marker_replication=DeleteMarkerReplication(
                DISABLED,
            ),
            rule_filter=Filter(
                AndOperator(
                    "TaxDocs",
                    bucket_tags,
                ),
            ),
            rule_id="rule1",
            priority=1,
        ),
    ],
)

async def main():
    await client.set_bucket_replication("my-bucket", config)

asyncio.run(main())

async set_bucket_tags(bucket_name: str, tags: Tags)[source]

Set tags configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
tags (Tags) – Tags object.

Raises:

ValueError – If tags is not of type Tags.

from miniopy_async import Minio
from miniopy_async.commonconfig import Tags
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

tags = Tags.new_bucket_tags()
tags["Project"] = "Project One"
tags["User"] = "jsmith"

async def main():
    await client.set_bucket_tags("my-bucket", tags)

asyncio.run(main())

async set_bucket_versioning(bucket_name: str, config: VersioningConfig)[source]

Set versioning configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (VersioningConfig) – VersioningConfig.

Raises:

ValueError – If config is not of type VersioningConfig.

from miniopy_async import Minio
from miniopy_async.commonconfig import ENABLED
from miniopy_async.versioningconfig import VersioningConfig
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    await client.set_bucket_versioning("my-bucket",
    VersioningConfig(ENABLED))

asyncio.run(main())

async set_object_lock_config(bucket_name: str, config: ObjectLockConfig)[source]

Set object-lock configuration to a bucket.

Parameters:

bucket_name (str) – Name of the bucket.
config (ObjectLockConfig) – ObjectLockConfig object.

Raises:

ValueError – If config is not of type ObjectLockConfig.

from miniopy_async import Minio
from miniopy_async.commonconfig import GOVERNANCE
from miniopy_async.objectlockconfig import DAYS, ObjectLockConfig
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

config = ObjectLockConfig(GOVERNANCE, 15, DAYS)

async def main():
    await client.set_object_lock_config("my-bucket", config)

asyncio.run(main())

async set_object_retention(bucket_name: str, object_name: str, config: Retention, version_id: str | None = None)[source]

Set retention configuration on an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
config (Retention) – Retention object.
version_id (str | None) – Version ID of the object.

Raises:

ValueError – If config is not of type Retention.

from datetime import datetime, timedelta
from miniopy_async import Minio
from miniopy_async.commonconfig import GOVERNANCE
from miniopy_async.retention import Retention
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

config = Retention(GOVERNANCE, datetime.utcnow() + timedelta(
days=10))

async def main():
    await client.set_object_retention(
        "my-bucket",
        "my-object",
        config
    )

asyncio.run(main())

async set_object_tags(bucket_name: str, object_name: str, tags: Tags, version_id: str | None = None)[source]

Set tags configuration to an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
version_id (Tags) – Version ID of the Object.
tags (str | None) – Tags object.

Raises:

ValueError – If tags is not of type Tags.

from miniopy_async import Minio
from miniopy_async.commonconfig import Tags
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

tags = Tags.new_object_tags()
tags["Project"] = "Project One"
tags["User"] = "jsmith"

async def main():
    await client.set_object_tags("my-bucket", "my-object", tags)

asyncio.run(main())

Get object information and metadata of an object.

Parameters:

bucket_name (str) – Name of the bucket.
object_name (str) – Object name in the bucket.
ssec (SseCustomerKey | None) – Server-side encryption customer key.
version_id (str | None) – Version ID of the object.
request_headers (DictType | None) – Any additional headers to be added with GET request.
extra_query_params (DictType | None) – Extra query parameters for advanced usage.

Returns:

Object.

Return type:

Object

from miniopy_async import Minio
from miniopy_async.sse import SseCustomerKey
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True  # http for False, https for True
)

async def main():
    # Get object information.
    print('example one')
    result = await client.stat_object("my-bucket", "my-object")
    print(
        "status: last-modified: {0}, size: {1}".format(
            result.last_modified, result.size,
        ),
    )

    # Get object information of version-ID.
    print('example two')
    result = await client.stat_object(
        "my-bucket", "my-object",
        version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
    )
    print(
        "status: last-modified: {0}, size: {1}".format(
            result.last_modified, result.size,
        ),
    )

    # Get SSE-C encrypted object information.
    print('example three')
    result = await client.stat_object(
        "my-bucket", "my-object",
        ssec=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
    )
    print(
        "status: last-modified: {0}, size: {1}".format(
            result.last_modified, result.size,
        ),
    )

asyncio.run(main())

trace_off()[source]: Disable HTTP trace.

trace_on(stream: TextIO)[source]

Enable http trace.

Parameters:: stream (TextIO) – Stream for writing HTTP call tracing.

async upload_snowball_objects(bucket_name: str, object_list: Iterable[SnowballObject], metadata: Dict[str, str | List[str] | Tuple[str]] | None = None, sse: Sse | None = None, tags: Tags | None = None, retention: Retention | None = None, legal_hold: bool = False, staging_filename: str | None = None, compression: bool = False) → ObjectWriteResult[source]

Uploads multiple objects in a single put call. It is done by creating intermediate TAR file optionally compressed which is uploaded to S3 service.

Parameters:

bucket_name (str) – Name of the bucket.
object_list (Iterable[SnowballObject]) – An iterable containing SnowballObject object.
metadata (DictType | None) – Any additional metadata to be uploaded along with your PUT request.
sse (Sse | None) – Server-side encryption.
tags (Tags | None) – Tags for the object.
retention (Retention | None) – Retention configuration object.
legal_hold (bool) – Flag to set legal hold for the object.
staging_filename (str | None) – A staging filename to create intermediate tarball.
compression (bool) – Flag to compress TAR ball.

Returns:

ObjectWriteResult object.

Return type:

ObjectWriteResult

from miniopy_async import Minio
from miniopy_async.commonconfig import SnowballObject
import io
from datetime import datetime
import asyncio

client = Minio(
    "play.min.io",
    access_key="Q3AM3UQ867SPQQA43P2F",
    secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
    secure=True,  # http for False, https for True
)

async def main():
    result = await client.upload_snowball_objects(
        "my-bucket",
        [
            SnowballObject("my-object1", filename="filename"),
            SnowballObject(
                "my-object2", data=io.BytesIO(b"hello"), length=5,
            ),
            SnowballObject(
                "my-object3", data=io.BytesIO(b"world"), length=5,
                mod_time=datetime.now(),
            ),
        ],
    )

asyncio.run(main())

miniopy_async.api.random() → x in the interval [0, 1).

miniopy_async.commonconfig module

Common request/response configuration of S3 APIs.

class miniopy_async.commonconfig.AndOperator(prefix: str | None = None, tags: Tags | None = None)[source]

Bases: object

AND operator.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property prefix: str | None: Get prefix.

property tags: Tags | None: Get tags.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.commonconfig.BaseRule(rule_filter: Filter | None = None, rule_id: str | None = None)[source]

Bases: object

Base rule class for Replication and Lifecycle.

static parsexml(element: Element) → tuple[Filter | None, str | None][source]: Parse XML and return filter and ID.

property rule_filter: Filter | None: Get replication rule filter.

property rule_id: str | None: Get rule ID.

toxml(element: Element | None) → Element[source]: Convert to XML.

Bases: ObjectConditionalReadArgs

A source object definition for compose_object method.

build_headers(object_size: int, etag: str)[source]: Build headers.

property headers: dict[str, str]: Get headers.

property object_size: int | None: Get object size.

classmethod of(src: ObjectConditionalReadArgs) → F[source]: Create ComposeSource from another source.

Bases: ObjectConditionalReadArgs

A source object definition for copy_object method.

classmethod of(src: ObjectConditionalReadArgs) → E[source]: Create CopySource from another source.

class miniopy_async.commonconfig.Filter(and_operator: AndOperator | None = None, prefix: str | None = None, tag: Tag | None = None)[source]

Bases: object

Lifecycle rule filter.

property and_operator: AndOperator | None: Get AND operator.

classmethod fromxml(element: Element) → D[source]: Create new object with values from XML element.

property prefix: str | None: Get prefix.

property tag: Tag | None: Get tag.

toxml(element: Element | None) → Element[source]: Convert to XML.

Bases: object

Base argument class holds condition properties for reading object.

property bucket_name: str: Get bucket name.

gen_copy_headers() → dict[str, str][source]: Generate copy source headers.

property length: int | None: Get length.

property match_etag: str | None: Get match ETag condition.

property modified_since: datetime | None: Get modified since condition.

property not_match_etag: str | None: Get not-match ETag condition.

property object_name: str: Get object name.

property offset: int | None: Get offset.

property region: str | None: Get region.

property ssec: SseCustomerKey | None: Get SSE-C.

property unmodified_since: datetime | None: Get unmodified since condition.

property version_id: str | None: Get version ID.

class miniopy_async.commonconfig.SnowballObject(object_name: str, filename: str | None = None, data: IO[bytes] | None = None, length: int | None = None, mod_time: datetime | None = None)[source]

Bases: object

A source object definition for upload_snowball_objects method.

property data: IO[bytes] | None: Get data.

property filename: str | None: Get filename.

property length: int | None: Get length.

property mod_time: datetime | None: Get modification time.

property object_name: str: Get object name.

class miniopy_async.commonconfig.Tag(key: str, value: str)[source]

Bases: object

Tag.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

property key: str: Get key.

toxml(element: Element | None) → Element[source]: Convert to XML.

property value: str: Get value.

class miniopy_async.commonconfig.Tags(for_object: bool = False)[source]

Bases: dict

dict extended to bucket/object tags.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

classmethod new_bucket_tags() → A[source]: Create new bucket tags.

classmethod new_object_tags() → A[source]: Create new object tags.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.commonconfig.check_status(status: str)[source]: Validate status.

miniopy_async.crypto module

Cryptography to read and write encrypted MinIO Admin payload

class miniopy_async.crypto.DecryptReader(response: ClientResponse, header: bytes, secret: bytes)[source]

Bases: object

BufferedIOBase compatible reader represents decrypted data of MinioAdmin APIs.

close()[source]: Close response and release network resources.

async classmethod init_async(response: ClientResponse, secret: bytes)[source]: Initialize DecryptReader asynchronously.

readable()[source]: Return this is readable.

async stream(num_bytes=32768)[source]: Stream extracted payload from response data. Upon completion, caller should call self.close() to release network resources.

writeable()[source]: Return this is not writeable.

async miniopy_async.crypto.decrypt(response: ClientResponse, secret_key: str) → bytes[source]: Decrypt response data.

miniopy_async.crypto.encrypt(payload: bytes, password: str) → bytes[source]: Encrypt given payload.

miniopy_async.datatypes module

Response of ListBuckets, ListObjects, ListObjectsV2 and ListObjectVersions API.

class miniopy_async.datatypes.AsyncEventIterable(response: ClientResponse, session: ClientSession | RetryClient)[source]

Bases: object

Context manager friendly event iterable.

class miniopy_async.datatypes.Bucket(name: str, creation_date: datetime | None)[source]

Bases: object

Bucket information.

property creation_date: datetime | None: Get creation date.

property name: str: Get name.

class miniopy_async.datatypes.CompleteMultipartUploadResult(response: ClientResponse, response_data: str)[source]

Bases: object

CompleteMultipartUpload API result.

property bucket_name: str | None: Get bucket name.

property etag: str | None: Get etag.

property http_headers: CIMultiDictProxy[str]: Get HTTP headers.

property location: str | None: Get location.

property object_name: str | None: Get object name.

property version_id: str | None: Get version ID.

class miniopy_async.datatypes.DeleteErrors(client: Minio, bucket_name: str, delete_object_list: Iterable[DeleteObject], bypass_governance_mode: bool = False)[source]

Bases: object

async gen_iterator() → AsyncGenerator[DeleteError][source]

class miniopy_async.datatypes.ListAllMyBucketsResult(buckets: list[Bucket])[source]

Bases: object

LissBuckets API result.

property buckets: List[Bucket]: Get buckets.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

class miniopy_async.datatypes.ListMultipartUploadsResult(response_data: str)[source]

Bases: object

ListMultipartUploads API result.

property bucket_name: str | None: Get bucket name.

property encoding_type: str | None: Get encoding type.

property is_truncated: bool: Get is-truncated flag.

property key_marker: str | None: Get key marker.

property max_uploads: int | None: Get max uploads.

property next_key_marker: str | None: Get next key marker.

property next_upload_id_marker: str | None: Get next upload ID marker.

property upload_id_marker: str | None: Get upload ID marker.

property uploads: list[Upload]: Get uploads.

class miniopy_async.datatypes.ListObjects(client: Minio, bucket_name: str, prefix: str | None = None, recursive: bool = False, start_after: str | None = None, include_user_meta: bool = False, include_version: bool = False, use_api_v1: bool = False, use_url_encoding_type: bool = True, fetch_owner: bool = False, extra_headers: DictType | None = None, extra_query_params: DictType | None = None)[source]

Bases: object

gen_iterator() → AsyncGenerator[Object][source]

class miniopy_async.datatypes.ListPartsResult(response_data: str)[source]

Bases: object

ListParts API result.

property bucket_name: str | None: Get bucket name.

property initator_name: str | None: Get initiator name.

property initiator_id: str | None: Get initiator ID.

property is_truncated: bool: Get is-truncated flag.

property max_parts: int | None: Get max parts.

property next_part_number_marker: int | None: Get next part number marker.

property object_name: str | None: Get object name.

property owner_id: str | None: Get owner ID.

property owner_name: str | None: Get owner name.

property part_number_marker: str | None: Get part number marker.

property parts: list[Part]: Get parts.

property storage_class: str | None: Get storage class.

Bases: object

Object information.

property bucket_name: str: Get bucket name.

property content_type: str | None: Get content type.

property etag: str | None: Get etag.

classmethod fromxml(element: Element, bucket_name: str, is_delete_marker: bool = False, encoding_type: str | None = None) → B[source]: Create new object with values from XML element.

property is_delete_marker: bool: Get whether this key is a delete marker.

property is_dir: bool: Get whether this key is a directory.

property is_latest: str | None: Get is-latest flag.

property last_modified: datetime | None: Get last modified time.

property metadata: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None: Get metadata.

property object_name: str | None: Get object name.

property owner_id: str | None: Get owner ID.

property owner_name: str | None: Get owner name.

property size: int | None: Get size.

property storage_class: str | None: Get storage class.

property tags: Tags | None: Get the tags

property version_id: str | None: Get version ID.

class miniopy_async.datatypes.Part(part_number: int, etag: str, last_modified: datetime | None = None, size: int | None = None)[source]

Bases: object

Part information of a multipart upload.

property etag: str: Get etag.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property last_modified: datetime | None: Get last-modified.

property part_number: int: Get part number.

property size: int | None: Get size.

class miniopy_async.datatypes.PeerInfo(deployment_id: str, endpoint: str, bucket_bandwidth_limit: str, bucket_bandwidth_set: str)[source]

Bases: object

Site replication peer information.

property bucket_bandwidth_limit: str: Get bucket bandwidth limit.

property bucket_bandwidth_set: str: Get bucket bandwidth set.

property bucket_bandwidth_updated_at: datetime | None: Get bucket bandwidth updated at.

property deployment_id: str: Get deployment ID.

property endpoint: str: Get endpoint.

property name: str | None: Get name.

property sync_status: str | None: Get sync status.

to_dict()[source]: Converts peer information to dictionary.

class miniopy_async.datatypes.PeerSite(name: str, endpoint: str, access_key: str, secret_key: str)[source]

Bases: object

Represents a cluster/site to be added to the set of replicated sites.

to_dict() → dict[str, str][source]: Convert to dictionary.

class miniopy_async.datatypes.PostPolicy(bucket_name: str, expiration: datetime)[source]

Bases: object

Post policy information to be used to generate presigned post policy form-data. Condition elements and respective condition for Post policy is available at https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-HTTPPOSTConstructPolicy.html#sigv4-PolicyConditions

add_content_length_range_condition(lower_limit: int, upper_limit: int)[source]: Add content-length-range condition with lower and upper limits.

add_equals_condition(element: str, value: str)[source]: Add equals condition of an element and value.

add_starts_with_condition(element: str, value: str)[source]: Add starts-with condition of an element and value. Value set to empty string does matching any content condition.

property bucket_name: str: Get bucket name.

form_data(creds: Credentials, region: str) → dict[str, Any][source]: Return form-data of this post policy. The returned dict contains x-amz-algorithm, x-amz-credential, x-amz-security-token, x-amz-date, policy and x-amz-signature.

remove_content_length_range_condition()[source]: Remove previously set content-length-range condition.

remove_equals_condition(element: str)[source]: Remove previously set equals condition of an element.

remove_starts_with_condition(element: str)[source]: Remove previously set starts-with condition of an element.

class miniopy_async.datatypes.SiteReplicationStatusOptions[source]

Bases: object

Represents site replication status options.

class ENTITY_TYPE(value)

Bases: Enum

An enumeration.

BUCKET = 'bucket'

GROUP = 'group'

POLICY = 'policy'

USER = 'user'

property buckets: bool: Get buckets.

property entity: str | None: Get entity.

property entity_value: str | None: Get entity value.

property groups: bool: Get groups.

property metrics: bool: Get metrics.

property policies: bool: Get policies.

property show_deleted: bool: Get show deleted.

to_query_params() → dict[str, str][source]: Convert this options to query parameters.

property users: bool: Get users.

class miniopy_async.datatypes.Upload(element: Element, encoding_type: str | None = None)[source]

Bases: object

Upload information of a multipart upload.

property initator_name: str | None: Get initiator name.

property initiated_time: datetime | None: Get initiated time.

property initiator_id: str | None: Get initiator ID.

property object_name: str: Get object name.

property owner_id: str | None: Get owner ID.

property owner_name: str | None: Get owner name.

property storage_class: str | None: Get storage class.

property upload_id: str | None: Get upload ID.

miniopy_async.datatypes.parse_copy_object(response_data: str) → tuple[str, datetime | None][source]: Parse CopyObject/UploadPartCopy response.

miniopy_async.datatypes.parse_list_objects(response_data: str, bucket_name: str | None = None) → tuple[list[Object], bool, str | None, str | None][source]: Parse ListObjects/ListObjectsV2/ListObjectVersions response.

miniopy_async.deleteobjects module

Request/response of DeleteObjects API.

class miniopy_async.deleteobjects.DeleteError(code: str, message: str | None, name: str | None, version_id: str | None)[source]

Bases: object

Delete error information.

property code: str: Get error code.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

property message: str | None: Get error message.

property name: str | None: Get name.

property version_id: str | None: Get version ID.

class miniopy_async.deleteobjects.DeleteObject(name: str, version_id: str | None = None)[source]

Bases: object

Delete object request information.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.deleteobjects.DeleteRequest(object_list: list[DeleteObject], quiet: bool = False)[source]

Bases: object

Delete object request.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.deleteobjects.DeleteResult(object_list: list[DeletedObject], error_list: list[DeleteError])[source]

Bases: object

Delete object result.

property error_list: list[DeleteError]: Get error list.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property object_list: list[DeletedObject]: Get object list.

class miniopy_async.deleteobjects.DeletedObject(name: str, version_id: str | None, delete_marker: bool, delete_marker_version_id: str | None)[source]

Bases: object

Deleted object information.

property delete_marker: bool: Get delete marker.

property delete_marker_version_id: str | None: Get delete marker version ID.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property name: str: Get name.

property version_id: str | None: Get version ID.

miniopy_async.error module

This module provides custom exception classes for MinIO library and API specific errors.

exception miniopy_async.error.InvalidResponseError(code: int, content_type: str | None, body: str | None)[source]

Bases: MinioException

Raised to indicate that non-xml response from server.

exception miniopy_async.error.MinioAdminException(code: str, body: str)[source]

Bases: Exception

Raised to indicate admin API execution error.

exception miniopy_async.error.MinioException[source]

Bases: Exception

Base Minio exception.

Bases: MinioException

Raised to indicate that error response is received when executing S3 operation.

property code: str | None: Get S3 error code.

copy(code: str, message: str) → S3Error[source]: Make a copy with replace code and message.

classmethod fromxml(response: ClientResponse, response_data: str) → A[source]: Create new object with values from XML element.

property message: str | None: Get S3 error message.

property response: ClientResponse: Get HTTP response.

exception miniopy_async.error.ServerError(message: str, status_code: int)[source]

Bases: MinioException

Raised to indicate that S3 service returning HTTP server error.

property status_code: int: Get HTTP status code.

miniopy_async.helpers module

Helper functions.

class miniopy_async.helpers.BaseURL(endpoint: str, region: str | None)[source]

Bases: object

Base URL of S3 endpoint.

property accelerate_host_flag: bool: Check if URL points to AWS accelerate host.

property aws_s3_prefix: str | None: Get AWS S3 domain prefix.

build(method: str, region: str, bucket_name: str | None = None, object_name: str | None = None, query_params: Dict[str, str | List[str] | Tuple[str]] | None = None) → SplitResult[source]: Build URL for given information.

property dualstack_host_flag: bool: Check if URL points to AWS dualstack host.

property host: str: Get hostname.

property is_aws_host: bool: Check if URL points to AWS host.

property is_https: bool: Check if scheme is HTTPS.

property region: str | None: Get region.

property virtual_style_flag: bool: Check to use virtual style or not.

Bases: object

Result class of any APIs doing object creation.

property bucket_name: str: Get bucket name.

property etag: str | None: Get etag.

property http_headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]]: Get HTTP headers.

property last_modified: datetime | None: Get last-modified time.

property location: str | None: Get location.

property object_name: str: Get object name.

property version_id: str | None: Get version ID.

class miniopy_async.helpers.ProgressType(*args, **kwargs)[source]

Bases: Protocol

typing stub for Put/Get object progress.

async set_meta(object_name: str, total_length: int)[source]: Set process meta information.

async update(length: int)[source]: Set current progress length.

miniopy_async.helpers.check_bucket_name(bucket_name: str, strict: bool = False, s3_check: bool = False)[source]: Check whether bucket name is valid optional with strict check or not.

miniopy_async.helpers.check_non_empty_string(string: str | bytes)[source]: Check whether given string is not empty.

miniopy_async.helpers.check_object_name(object_name: str)[source]: Check whether given object name is valid.

miniopy_async.helpers.check_sse(sse: Sse | None)[source]: Check sse is Sse type or not.

miniopy_async.helpers.check_ssec(sse: SseCustomerKey | None)[source]: Check sse is SseCustomerKey type or not.

miniopy_async.helpers.genheaders(headers: Dict[str, str | List[str] | Tuple[str]] | None, sse: Sse | None, tags: dict[str, str] | None, retention, legal_hold: bool) → Dict[str, str | List[str] | Tuple[str]][source]: Generate headers for given parameters.

miniopy_async.helpers.get_part_info(object_size: int, part_size: int) → tuple[int, int][source]: Compute part information for object and part size.

miniopy_async.helpers.headers_to_strings(headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]], titled_key: bool = False) → str[source]: Convert HTTP headers to multi-line string.

miniopy_async.helpers.is_valid_policy_type(policy: str | bytes) → bool[source]

Validate if policy is type str

Parameters:: policy (str | bytes) – S3 style Bucket policy.
Returns:: True if policy parameter is of a valid type str.
Return type:: bool
Raise:: TypeError otherwise.

miniopy_async.helpers.makedirs(path: str)[source]: Wrapper of os.makedirs() ignores errno.EEXIST.

miniopy_async.helpers.md5sum_hash(data: str | bytes | None) → str | None[source]: Compute MD5 of data and return hash as Base64 encoded value.

miniopy_async.helpers.normalize_headers(headers: Dict[str, str | List[str] | Tuple[str]] | None) → Dict[str, str | List[str] | Tuple[str]][source]: Normalize headers by prefixing ‘X-Amz-Meta-’ for user metadata.

miniopy_async.helpers.queryencode(query: str, safe: str = '', encoding: str | None = None, errors: str | None = None) → str[source]: Encode query parameter value.

miniopy_async.helpers.quote(resource: str, safe: str = '/', encoding: str | None = None, errors: str | None = None) → str[source]: Wrapper to urllib.parse.quote() replacing back to ‘~’ for older python versions.

async miniopy_async.helpers.read_part_data(stream: BinaryIO | FileIOWrapperBase, size: int, part_data: bytes = b'', progress: ProgressType | None = None) → bytes[source]: Read part data of given size from stream.

miniopy_async.helpers.sha256_hash(data: str | bytes | None) → str[source]: Compute SHA-256 of data and return hash as hex encoded value.

miniopy_async.helpers.url_replace(url: SplitResult, scheme: str | None = None, netloc: str | None = None, path: str | None = None, query: str | None = None, fragment: str | None = None) → SplitResult[source]: Return new URL with replaced properties in given URL.

miniopy_async.legalhold module

Request/response of PutObjectLegalHold and GetObjectLegalHold S3 APIs.

class miniopy_async.legalhold.LegalHold(status: bool = False)[source]

Bases: object

Legal hold configuration.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property status: bool: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.lifecycleconfig module

Request/response of PutBucketLifecycleConfiguration and GetBucketLifecycleConfiguration APIs.

class miniopy_async.lifecycleconfig.AbortIncompleteMultipartUpload(days_after_initiation: int | None = None)[source]

Bases: object

Abort incomplete multipart upload.

property days_after_initiation: int | None: Get days after initiation.

classmethod fromxml(element: Element) → E[source]: Create new object with values from XML element.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.DateDays(date: datetime | None = None, days: int | None = None)[source]

Bases: object

Base class holds date and days of Transition and Expiration.

property date: datetime | None: Get date.

property days: int | None: Get days.

static parsexml(element: Element) → tuple[datetime | None, int | None][source]: Parse XML to date and days.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.Expiration(date: datetime | None = None, days: int | None = None, expired_object_delete_marker: bool | None = None)[source]

Bases: DateDays

Expiration.

property expired_object_delete_marker: bool | None: Get expired object delete marker.

classmethod fromxml(element: Element) → D[source]: Create new object with values from XML element.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.LifecycleConfig(rules: list[Rule])[source]

Bases: object

Lifecycle configuration.

classmethod fromxml(element: Element) → G[source]: Create new object with values from XML element.

property rules: list[Rule]: Get rules.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.NoncurrentVersionExpiration(noncurrent_days: int | None = None, newer_noncurrent_versions: int | None = None)[source]

Bases: object

Noncurrent version expiration.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property newer_noncurrent_versions: int | None: Get Newer Noncurrent versions.

property noncurrent_days: int | None: Get Noncurrent days.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.NoncurrentVersionTransition(noncurrent_days: int | None = None, storage_class: str | None = None, newer_noncurrent_versions: int | None = None)[source]

Bases: object

Noncurrent version transition.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

property newer_noncurrent_versions: int | None: Get Newer Noncurrent versions.

property noncurrent_days: int | None: Get Noncurrent days.

property storage_class: str | None: Get storage class.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.lifecycleconfig.Rule(status: str, abort_incomplete_multipart_upload: AbortIncompleteMultipartUpload | None = None, expiration: Expiration | None = None, rule_filter: Filter | None = None, rule_id: str | None = None, noncurrent_version_expiration: NoncurrentVersionExpiration | None = None, noncurrent_version_transition: NoncurrentVersionTransition | None = None, transition: Transition | None = None)[source]

Bases: BaseRule

Lifecycle rule.

property abort_incomplete_multipart_upload: AbortIncompleteMultipartUpload | None: Get abort incomplete multipart upload.

property expiration: Expiration | None: Get expiration.

classmethod fromxml(element: Element) → F[source]: Create new object with values from XML element.

property noncurrent_version_expiration: NoncurrentVersionExpiration | None: Get noncurrent version expiration.

property noncurrent_version_transition: NoncurrentVersionTransition | None: Get noncurrent version transition.

property status: str | None: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

property transition: Transition | None: Get transition.

class miniopy_async.lifecycleconfig.Transition(date: datetime | None = None, days: int | None = None, storage_class: str | None = None)[source]

Bases: DateDays

Transition.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property storage_class: str | None: Get storage class.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.minioadmin module

MinIO Admin wrapper using MinIO Client (mc) tool.

class miniopy_async.minioadmin.MinioAdmin(endpoint: str, credentials: Provider, region: str = '', secure: bool = True, cert_check: bool = True, client_session: Callable[[...], ClientSession | RetryClient] | None = None)[source]

Bases: object

Client to perform MinIO administration operations.

async add_service_account(access_key: str | None = None, secret_key: str | None = None, name: str | None = None, description: str | None = None, policy_file: str | None = None, expiration: str | None = None, status: str | None = None) → str[source]: Add a new service account with the given access key and secret key

async add_site_replication(peer_sites: list[PeerSite]) → str[source]: Add peer sites to site replication.

async attach_policy(policies: list[str], user: str | None = None, group: str | None = None) → str[source]: Attach builtin policies.

async attach_policy_ldap(policies: list[str], user: str | None = None, group: str | None = None) → str[source]: Attach policies for LDAP.

async bucket_quota_clear(bucket: str) → str[source]: Clear bucket quota configuration.

async bucket_quota_get(bucket: str) → str[source]: Get bucket quota configuration.

async bucket_quota_set(bucket: str, size: int) → str[source]: Set bucket quota configuration.

async config_get(key: str | None = None) → str[source]: Get configuration parameters.

async config_history() → str[source]: Get historic configuration changes.

async config_reset(key: str, name: str | None = None) → str[source]: Reset configuration parameters.

async config_restore(restore_id: str) → str[source]: Restore to a specific configuration history.

async config_set(key: str, config: dict[str, str] | None = None) → str[source]: Set configuration parameters.

async delete_service_account(access_key: str) → str[source]: Delete a service account

async detach_policy(policies: list[str], user: str | None = None, group: str | None = None) → str[source]: Detach builtin policies.

async detach_policy_ldap(policies: list[str], user: str | None = None, group: str | None = None) → str[source]: Detach policies for LDAP.

async edit_site_replication(peer_info: PeerInfo) → str[source]: Edit site replication with given peer information.

async get_data_usage_info()[source]: Get data usage info

async get_policy_entities(users: list[str], groups: list[str], policies: list[str]) → str[source]: Get builtin policy entities.

async get_service_account(access_key: str) → str[source]: Get information about service account

async get_site_replication_info() → str[source]: Get site replication information.

async get_site_replication_status(options: SiteReplicationStatusOptions) → str[source]: Get site replication information.

async group_add(group_name: str, members: str) → str[source]: Add users a new or existing group.

async group_disable(group_name: str) → str[source]: Disable group.

async group_enable(group_name: str) → str[source]: Enable group.

async group_info(group_name: str) → str[source]: Get group information.

async group_list() → str[source]: List groups.

async group_remove(group_name: str, members: str | None = None) → str[source]: Remove group or members from a group.

async info() → str[source]: Get MinIO server information.

async kms_key_create(key: str | None = None) → str[source]: Create a new KMS master key.

async kms_key_status(key: str | None = None) → str[source]: Get status information of a KMS master key.

async list_access_keys_ldap(user_dn: str, list_type: str) → str[source]: List service accounts belonging to the specified user.

async list_access_keys_ldap_bulk(users: list[str], list_type: str, all_users: bool) → str[source]: List access keys belonging to the given users or all users.

async list_service_account(user: str) → str[source]: List service accounts of user

async policy_add(policy_name: str, policy_file: str) → str[source]: Add new policy.

async policy_info(policy_name: str) → str[source]: Get policy information.

async policy_list() → str[source]: List policies.

async policy_remove(policy_name: str) → str[source]: Remove policy.

async policy_set(policy_name: str, user: str | None = None, group: str | None = None) → str[source]: Set IAM policy on a user or group.

async policy_unset(policy_name: str | list[str], user: str | None = None, group: str | None = None) → str[source]: Unset an IAM policy for a user or group.

async profile_start(profilers: tuple[str] = ()) → str[source]: Runs a system profile

async remove_site_replication(sites: str | None = None, all_sites: bool = False) → str[source]: Remove given sites or all sites from site replication.

async service_restart() → str[source]: Restart MinIO service.

async service_stop() → str[source]: Stop MinIO service.

set_app_info(app_name: str, app_version: str)[source]

Set your application name and version to user agent header.

Parameters:

app_name – Application name.
app_version – Application version.

Example::: client.set_app_info(‘my_app’, ‘1.0.2’)

async top_locks() → str[source]: Get a list of the 10 oldest locks on a MinIO cluster.

trace_off()[source]: Disable HTTP trace.

trace_on(stream: TextIO)[source]

Enable http trace.

Parameters:: stream – Stream for writing HTTP call tracing.

async update() → str[source]: Update MinIO.

async update_service_account(access_key: str, secret_key: str | None = None, name: str | None = None, description: str | None = None, policy_file: str | None = None, expiration: str | None = None, status: str | None = None) → str[source]: Update an existing service account

async user_add(access_key: str, secret_key: str) → str[source]: Create user with access and secret keys

async user_disable(access_key: str) → str[source]: Disable user.

async user_enable(access_key: str) → str[source]: Enable user.

async user_info(access_key: str) → str[source]: Get information about user

async user_list() → str[source]: List all users

async user_remove(access_key: str) → str[source]: Delete user

miniopy_async.notificationconfig module

Request/response of PutBucketNotificationConfiguration and GetBucketNotiicationConfiguration APIs.

class miniopy_async.notificationconfig.CloudFuncConfig(cloud_func: str, events: list[str], config_id: str | None = None, prefix_filter_rule: PrefixFilterRule | None = None, suffix_filter_rule: SuffixFilterRule | None = None)[source]

Bases: CommonConfig

Cloud function configuration.

property cloud_func: str: Get cloud function ARN.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.notificationconfig.CommonConfig(events: list[str], config_id: str | None, prefix_filter_rule: PrefixFilterRule | None, suffix_filter_rule: SuffixFilterRule | None)[source]

Bases: object

Common for cloud-function/queue/topic configuration.

property config_id: str | None: Get configuration ID.

property events: list[str]: Get events.

static parsexml(element: Element) → tuple[list[str], str | None, PrefixFilterRule | None, SuffixFilterRule | None][source]: Parse XML.

property prefix_filter_rule: PrefixFilterRule | None: Get prefix filter rule.

property suffix_filter_rule: SuffixFilterRule | None: Get suffix filter rule.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.notificationconfig.FilterRule(name: str, value: str)[source]

Bases: object

Filter rule.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property name: str: Get name.

toxml(element: Element | None) → Element[source]: Convert to XML.

property value: str: Get value.

class miniopy_async.notificationconfig.NotificationConfig(cloud_func_config_list: list[CloudFuncConfig] | None = None, queue_config_list: list[QueueConfig] | None = None, topic_config_list: list[TopicConfig] | None = None)[source]

Bases: object

Notification configuration.

property cloud_func_config_list: list[CloudFuncConfig] | None: Get cloud function configuration list.

classmethod fromxml(element: Element) → E[source]: Create new object with values from XML element.

property queue_config_list: list[QueueConfig] | None: Get queue configuration list.

property topic_config_list: list[TopicConfig] | None: Get topic configuration list.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.notificationconfig.PrefixFilterRule(value: str)[source]

Bases: FilterRule

Prefix filter rule.

class miniopy_async.notificationconfig.QueueConfig(queue: str, events: list[str], config_id: str | None = None, prefix_filter_rule: PrefixFilterRule | None = None, suffix_filter_rule: SuffixFilterRule | None = None)[source]

Bases: CommonConfig

Queue configuration.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property queue: str: Get queue ARN.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.notificationconfig.SuffixFilterRule(value: str)[source]

Bases: FilterRule

Suffix filter rule.

class miniopy_async.notificationconfig.TopicConfig(topic: str, events: list[str], config_id: str | None = None, prefix_filter_rule: PrefixFilterRule | None = None, suffix_filter_rule: SuffixFilterRule | None = None)[source]

Bases: CommonConfig

Get topic configuration.

classmethod fromxml(element: Element) → D[source]: Create new object with values from XML element.

property topic: str: Get topic ARN.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.objectlockconfig module

Request/response of PutObjectLockConfiguration and GetObjectLockConfiguration APIs.

class miniopy_async.objectlockconfig.ObjectLockConfig(mode: str | None, duration: int | None, duration_unit: str | None)[source]

Bases: object

Object lock configuration.

property duration: tuple[int | None, str | None]: Get duration and it’s unit.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property mode: str | None: Get mode.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.replicationconfig module

Request/response of PutBucketReplication and GetBucketReplication APIs.

class miniopy_async.replicationconfig.AccessControlTranslation(owner: str = 'Destination')[source]

Bases: object

Access control translation.

classmethod fromxml(element: Element) → G[source]: Create new object with values from XML element.

property owner: str: Get owner.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.DeleteMarkerReplication(status='Disabled')[source]

Bases: Status

Delete marker replication.

class miniopy_async.replicationconfig.Destination(bucket_arn: str, access_control_translation: AccessControlTranslation | None = None, account: str | None = None, encryption_config: EncryptionConfig | None = None, metrics: Metrics | None = None, replication_time: ReplicationTime | None = None, storage_class: str | None = None)[source]

Bases: object

Replication destination.

property access_control_translation: AccessControlTranslation | None: Get access control translation.

property account: str | None: Get account.

property bucket_arn: str: Get bucket ARN.

property encryption_config: EncryptionConfig | None: Get encryption configuration.

classmethod fromxml(element: Element) → H[source]: Create new object with values from XML element.

property metrics: Metrics | None: Get metrics.

property replication_time: ReplicationTime | None: Get replication time.

property storage_class: str | None: Get storage class.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.EncryptionConfig(replica_kms_key_id: str | None = None)[source]

Bases: object

Encryption configuration.

classmethod fromxml(element: Element) → F[source]: Create new object with values from XML element.

property replica_kms_key_id: str | None: Get replica KMS key ID.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.EventThreshold(minutes: None | int = 15)[source]

Bases: ReplicationTimeValue

Event threshold.

class miniopy_async.replicationconfig.ExistingObjectReplication(status: str)[source]

Bases: Status

Existing object replication.

class miniopy_async.replicationconfig.Metrics(event_threshold: EventThreshold, status: str)[source]

Bases: object

Metrics.

property event_threshold: EventThreshold: Get event threshold.

classmethod fromxml(element: Element) → E[source]: Create new object with values from XML element.

property status: str: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.ReplicationConfig(role: str, rules: list[Rule])[source]

Bases: object

Replication configuration.

classmethod fromxml(element: Element) → J[source]: Create new object with values from XML element.

property role: str: Get role.

property rules: list[Rule]: Get rules.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.ReplicationTime(time: Time, status: str)[source]

Bases: object

Replication time.

classmethod fromxml(element: Element) → D[source]: Create new object with values from XML element.

property status: str: Get status.

property time: Time: Get time value.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.ReplicationTimeValue(minutes: None | int = 15)[source]

Bases: object

Replication time value.

classmethod fromxml(element: Element) → C[source]: Create new object with values from XML element.

property minutes: int | None: Get minutes.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.Rule(destination: Destination, status: str, delete_marker_replication: DeleteMarkerReplication | None = None, existing_object_replication: ExistingObjectReplication | None = None, rule_filter: Filter | None = None, rule_id: str | None = None, prefix: str | None = None, priority: int | None = None, source_selection_criteria: SourceSelectionCriteria | None = None)[source]

Bases: BaseRule

Replication rule.

property delete_marker_replication: DeleteMarkerReplication | None: Get delete marker replication.

property destination: Destination: Get destination.

property existing_object_replication: ExistingObjectReplication | None: Get existing object replication.

classmethod fromxml(element: Element) → I[source]: Create new object with values from XML element.

property prefix: str | None: Get prefix.

property priority: int | None: Get priority.

property source_selection_criteria: SourceSelectionCriteria | None: Get source selection criteria.

property status: str: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.SourceSelectionCriteria(sse_kms_encrypted_objects: SseKmsEncryptedObjects | None = None)[source]

Bases: object

Source selection criteria.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

property sse_kms_encrypted_objects: SseKmsEncryptedObjects | None: Get SSE KMS encrypted objects.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.SseKmsEncryptedObjects(status: str)[source]

Bases: Status

SSE KMS encrypted objects.

class miniopy_async.replicationconfig.Status(status: str)[source]

Bases: object

Status.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property status: str: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.replicationconfig.Time(minutes: None | int = 15)[source]

Bases: ReplicationTimeValue

Time.

miniopy_async.retention module

Request/response of PutObjectRetention and GetObjectRetention APIs.

class miniopy_async.retention.Retention(mode: str, retain_until_date: datetime)[source]

Bases: object

Retention configuration.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property mode: str: Get mode.

property retain_until_date: datetime: Get retain util date.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.select module

Request/response of SelectObjectContent API.

Bases: InputSerialization

CSV input serialization.

toxml(element: Element | None) → Element[source]: Convert to XML.

Bases: object

CSV output serialization.

toxml(element: Element | None)[source]: Convert to XML.

class miniopy_async.select.InputSerialization(compression_type: str | None)[source]

Bases: object

Input serialization.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.select.JSONInputSerialization(compression_type: str | None = None, json_type: str | None = None)[source]

Bases: InputSerialization

JSON input serialization.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.select.JSONOutputSerialization(record_delimiter: str | None = None)[source]

Bases: object

JSON output serialization.

toxml(element: Element | None)[source]: Convert to XML.

class miniopy_async.select.ParquetInputSerialization[source]

Bases: InputSerialization

Parquet input serialization.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.select.SelectObjectReader(response: ClientResponse, session: ClientSession | RetryClient)[source]

Bases: object

BufferedIOBase compatible reader represents response data of Minio.select_object_content() API.

async close()[source]: Close response and release network resources.

readable() → bool[source]: Return this is readable.

stats() → Stats | None[source]: Get stats information.

async stream(num_bytes: int = 32768) → AsyncGenerator[bytes][source]: Stream extracted payload from response data. Upon completion, caller should call self.close() to release network resources.

writeable() → bool[source]: Return this is not writeable.

class miniopy_async.select.SelectRequest(expression: str, input_serialization: CSVInputSerialization | JSONInputSerialization | ParquetInputSerialization, output_serialization: CSVOutputSerialization | JSONOutputSerialization, request_progress: bool = False, scan_start_range: str | None = None, scan_end_range: str | None = None)[source]

Bases: object

Select object content request.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.select.Stats(data: bytes)[source]

Bases: object

Progress/Stats information.

property bytes_processed: str | None: Get bytes processed.

property bytes_returned: str | None: Get bytes returned.

property bytes_scanned: str | None: Get bytes scanned.

miniopy_async.signer module

This module implements all helpers for AWS Signature version ‘4’ support.

miniopy_async.signer.get_credential_string(access_key: str, date: datetime, region: str) → str[source]: Get credential string of given access key, date and region.

miniopy_async.signer.post_presign_v4(data: str, secret_key: str, date: datetime, region: str) → str[source]: Do signature V4 of given presign POST form-data.

miniopy_async.signer.presign_v4(method: str, url: SplitResult, region: str, credentials: Credentials, date: datetime, expires: int) → SplitResult[source]: Do signature V4 of given presign request.

miniopy_async.signer.sign_v4_s3(method: str, url: SplitResult, region: str, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]], credentials: Credentials, content_sha256: str, date: datetime) → Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]][source]: Do signature V4 of given request for S3 service.

miniopy_async.signer.sign_v4_sts(method: str, url: SplitResult, region: str, headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]], credentials: Credentials, content_sha256: str, date: datetime) → Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]][source]: Do signature V4 of given request for STS service.

miniopy_async.sse module

This module contains core API parsers.

class miniopy_async.sse.Sse[source]

Bases: object

Server-side encryption base class.

copy_headers() → dict[str, str][source]: Return copy headers.

abstract headers() → dict[str, str][source]: Return headers.

tls_required() → bool[source]: Return TLS required to use this server-side encryption.

class miniopy_async.sse.SseCustomerKey(key: bytes)[source]

Bases: Sse

Server-side encryption - customer key type.

copy_headers() → dict[str, str][source]: Return copy headers.

headers() → dict[str, str][source]: Return headers.

class miniopy_async.sse.SseKMS(key: str, context: dict[str, Any])[source]

Bases: Sse

Server-side encryption - KMS type.

headers() → dict[str, str][source]: Return headers.

class miniopy_async.sse.SseS3[source]

Bases: Sse

Server-side encryption - S3 type.

headers() → dict[str, str][source]: Return headers.

tls_required() → bool[source]: Return TLS required to use this server-side encryption.

miniopy_async.sseconfig module

Request/response of PutBucketEncryption and GetBucketEncryption APIs.

class miniopy_async.sseconfig.Rule(sse_algorithm: str, kms_master_key_id: str | None = None)[source]

Bases: object

Server-side encryption rule.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property kms_master_key_id: str | None: Get KMS master key ID.

classmethod new_sse_kms_rule(kms_master_key_id: str | None = None) → A[source]: Create new SSE-KMS rule.

classmethod new_sse_s3_rule() → A[source]: Create SSE-S3 rule.

property sse_algorithm: str: Get SSE algorithm.

toxml(element: Element | None) → Element[source]: Convert to XML.

class miniopy_async.sseconfig.SSEConfig(rule: Rule)[source]

Bases: object

server-side encryption configuration.

classmethod fromxml(element: Element) → B[source]: Create new object with values from XML element.

property rule: Rule: Get rule.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.tagging module

Tagging for bucket and object.

class miniopy_async.tagging.Tagging(tags: Tags | None)[source]

Bases: object

Tagging for buckets and objects.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property tags: Tags | None: Get tags.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.time module

Time formatter for S3 APIs.

miniopy_async.time.from_http_header(value: str) → datetime[source]: Parse HTTP header date formatted string to datetime.

miniopy_async.time.from_iso8601utc(value: str | None) → datetime | None[source]: Parse UTC ISO-8601 formatted string to datetime.

miniopy_async.time.to_amz_date(value: datetime) → str[source]: Format datetime into AMZ date formatted string.

miniopy_async.time.to_float(value: datetime) → float[source]: Convert datetime into float value.

miniopy_async.time.to_http_header(value: datetime) → str[source]: Format datatime into HTTP header date formatted string.

miniopy_async.time.to_iso8601utc(value: datetime | None) → str | None[source]: Format datetime into UTC ISO-8601 formatted string.

miniopy_async.time.to_signer_date(value: datetime) → str[source]: Format datetime into SignatureV4 date formatted string.

miniopy_async.time.utcnow() → datetime[source]: Timezone-aware wrapper to datetime.utcnow().

miniopy_async.versioningconfig module

Request/response of PutBucketVersioning and GetBucketVersioning APIs.

class miniopy_async.versioningconfig.VersioningConfig(status: str | None = None, mfa_delete: str | None = None, excluded_prefixes: list[str] | None = None, exclude_folders: bool = False)[source]

Bases: object

Versioning configuration.

property exclude_folders: bool: Get exclude folders.

property excluded_prefixes: list[str] | None: Get excluded prefixes.

classmethod fromxml(element: Element) → A[source]: Create new object with values from XML element.

property mfa_delete: str | None: Get MFA delete.

property status: str: Get status.

toxml(element: Element | None) → Element[source]: Convert to XML.

miniopy_async.xml module

XML utility module.

miniopy_async.xml.Element(tag: str, namespace: str = 'http://s3.amazonaws.com/doc/2006-03-01/') → Element[source]: Create ElementTree.Element with tag and namespace.

class miniopy_async.xml.FromXmlType(*args, **kwargs)[source]

Bases: Protocol

typing stub for class with fromxml method

classmethod fromxml(element: Element) → A[source]: Create python object with values from XML element.

miniopy_async.xml.SubElement(parent: Element, tag: str, text: str | None = None) → Element[source]: Create ElementTree.SubElement on parent with tag and text.

class miniopy_async.xml.ToXmlType(*args, **kwargs)[source]

Bases: Protocol

typing stub for class with toxml method

toxml(element: Element | None) → Element[source]: Convert python object to ElementTree.Element.

miniopy_async.xml.find(element: Element, name: str, strict: bool = False) → Element | None[source]: Namespace aware ElementTree.Element.find().

miniopy_async.xml.findall(element: Element, name: str) → list[Element][source]: Namespace aware ElementTree.Element.findall().

miniopy_async.xml.findtext(element: Element, name: str, strict: bool = False) → str | None[source]: Namespace aware ElementTree.Element.findtext() with strict flag raises ValueError if element name not exist.

miniopy_async.xml.getbytes(element: Element) → bytes[source]: Convert ElementTree.Element to bytes.

miniopy_async.xml.marshal(obj: ToXmlType) → bytes[source]: Get XML data as bytes of ElementTree.Element.

miniopy_async.xml.unmarshal(cls: Type[B], xmlstring: str) → B[source]: Unmarshal given XML string to an object of passed class.