Skip to content

graflo.object_storage.bucket

Ensure S3 buckets exist (idempotent).

Attributes

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

ensure_bucket_exists(client, bucket)

Create bucket if it does not exist. Return bucket.

Safe to call repeatedly. Raises :exc:botocore.exceptions.ClientError for non-recoverable API errors.

Source code in graflo/object_storage/bucket.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def ensure_bucket_exists(client: Any, bucket: str) -> str:
    """Create *bucket* if it does not exist. Return *bucket*.

    Safe to call repeatedly. Raises :exc:`botocore.exceptions.ClientError`
    for non-recoverable API errors.
    """
    try:
        client.head_bucket(Bucket=bucket)
        logger.info("S3 bucket %r already exists", bucket)
        return bucket
    except ClientError as e:
        code = str(e.response.get("Error", {}).get("Code", ""))
        http = e.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
        missing = code in ("404", "NoSuchBucket", "NotFound") or http == 404
        if not missing:
            raise

    try:
        client.create_bucket(Bucket=bucket)
        logger.info("Created S3 bucket %r", bucket)
    except ClientError as e:
        code = str(e.response.get("Error", {}).get("Code", ""))
        if code in ("BucketAlreadyOwnedByYou", "BucketAlreadyExists"):
            logger.info(
                "S3 bucket %r already exists (race or concurrent create)", bucket
            )
        else:
            raise

    return bucket

ensure_staging_bucket_for_config(cfg=None)

Ensure :attr:~graflo.object_storage.config.S3EndpointConfig.bucket exists.

If cfg is None, loads :meth:~graflo.object_storage.config.S3EndpointConfig.from_docker_env.

Source code in graflo/object_storage/bucket.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def ensure_staging_bucket_for_config(cfg: S3EndpointConfig | None = None) -> str:
    """Ensure :attr:`~graflo.object_storage.config.S3EndpointConfig.bucket` exists.

    If *cfg* is ``None``, loads :meth:`~graflo.object_storage.config.S3EndpointConfig.from_docker_env`.
    """
    from graflo.object_storage.config import S3EndpointConfig
    from graflo.object_storage.s3_client import boto3_s3_client_from_endpoint

    resolved = cfg or S3EndpointConfig.from_docker_env()
    client = boto3_s3_client_from_endpoint(resolved)
    try:
        return ensure_bucket_exists(client, resolved.bucket)
    except EndpointConnectionError as e:
        raise RuntimeError(
            f"Cannot connect to {resolved.endpoint_url!r}. "
            "Start the S3 service (e.g. `docker compose --profile graflo.seaweedfs up -d` "
            "in `docker/seaweedfs`) and ensure `S3_API_PORT` / `S3_ENDPOINT` matches the "
            "published S3 API port (default host 8333)."
        ) from e