`buckethead.provisioning`¶

One-time setup utilities: create an R2 bucket, issue credentials, and store them in a secret store (1Password today). Driven by the buckethead provision CLI — the Python surface here is primarily for programmatic reuse and is best treated as internal until it stabilizes.

buckethead.provisioning ¶

Provisioning utilities.

Three layers:

Protocols (cloud/base.BucketProvider, secrets/base.SecretStore) — vendor-agnostic contracts.
Vendors (cloud/cloudflare_r2, secrets/onepassword) — concrete implementations that satisfy the Protocols.
Orchestration (orchestrator.provision_bucket, orchestrator.register_bucket) — pure functions that only touch Protocols, so they're trivially testable and swappable.

CLI and any library caller wire them together via factories.make_*, the single registry of vendor name → concrete class.

BucketCredentials ¶

Bases: BaseModel

S3-compatible credentials + bucket addressing info.

The shape is storage-agnostic — Cloudflare R2, AWS S3, MinIO all fit. region defaults to "auto" which R2 accepts; S3 requires a real region.

CloudflareAuth ¶

Bases: BaseModel

Inputs required for Cloudflare API calls.

destroy_project ¶

destroy_project(
    *,
    bucket: str,
    access_key_id: str,
    item_title: str,
    cloud: BucketProvider,
    secrets: SecretStore,
) -> None

Tear down what provision_bucket created, in the reverse-safe order.

Revoke the credentials first so a zombie token can't survive if a later step fails.
Delete the bucket (must be empty; CF errors otherwise).
Delete the secret-store item last — losing the 1P entry before we've used it to find the access_key_id would strand the token.

Each step is idempotent on its implementation, so a partial prior run doesn't block a retry.

Source code in src/buckethead/provisioning/orchestrator.py

def destroy_project(
    *,
    bucket: str,
    access_key_id: str,
    item_title: str,
    cloud: BucketProvider,
    secrets: SecretStore,
) -> None:
    """Tear down what provision_bucket created, in the reverse-safe order.

    1. Revoke the credentials first so a zombie token can't survive if a
       later step fails.
    2. Delete the bucket (must be empty; CF errors otherwise).
    3. Delete the secret-store item last — losing the 1P entry before
       we've used it to find the access_key_id would strand the token.

    Each step is idempotent on its implementation, so a partial prior run
    doesn't block a retry.
    """
    cloud.delete_credentials(access_key_id)
    cloud.delete_bucket(bucket)
    secrets.delete_item(item_title)

provision_bucket ¶

provision_bucket(
    *,
    project: str,
    bucket: str,
    item_title: str,
    cloud: BucketProvider,
    secrets: SecretStore,
) -> str

Create a bucket + scoped credentials, persist creds in the secret store.

Returns the secret-store reference string for the stored credentials.

Source code in src/buckethead/provisioning/orchestrator.py

def provision_bucket(
    *,
    project: str,
    bucket: str,
    item_title: str,
    cloud: BucketProvider,
    secrets: SecretStore,
) -> str:
    """Create a bucket + scoped credentials, persist creds in the secret store.

    Returns the secret-store reference string for the stored credentials.
    """
    cloud.create_bucket(bucket)
    creds = cloud.create_credentials(bucket=bucket, label=f"buckethead/{project}")
    return secrets.write_credentials(title=item_title, creds=creds)

register_bucket ¶

register_bucket(
    *,
    bucket: str,
    access_key_id: str,
    secret_access_key: str,
    endpoint_url: str,
    region: str,
    item_title: str,
    secrets: SecretStore,
) -> str

Persist BYO S3 credentials in the secret store. No cloud calls.

Source code in src/buckethead/provisioning/orchestrator.py

def register_bucket(
    *,
    bucket: str,
    access_key_id: str,
    secret_access_key: str,
    endpoint_url: str,
    region: str,
    item_title: str,
    secrets: SecretStore,
) -> str:
    """Persist BYO S3 credentials in the secret store. No cloud calls."""
    creds = BucketCredentials(
        access_key_id=access_key_id,
        secret_access_key=SecretStr(secret_access_key),
        endpoint_url=endpoint_url,
        bucket=bucket,
        region=region,
    )
    return secrets.write_credentials(title=item_title, creds=creds)

buckethead.provisioning¶