buckethead.config

Pydantic configuration models.

buckethead.config

Configuration subsystem.

Aggregates Pydantic data models, env-driven settings, persistent user config, and the secret-reference resolver registry.

BucketConfig

Bases: BaseModel

Where BucketHead stores snapshots and filestore blobs.

for_r2 classmethod

for_r2(
    *,
    account_id: str,
    bucket: str,
    access_key_id: str,
    secret_access_key: str,
    key: str | None = None,
    files_prefix: str | None = None,
) -> Self

Construct a BucketConfig pointed at Cloudflare R2.

Fills in the account-specific S3 endpoint and the mandatory region='auto'.

Source code in src/buckethead/config/models.py

@classmethod
def for_r2(
    cls,
    *,
    account_id: str,
    bucket: str,
    access_key_id: str,
    secret_access_key: str,
    key: str | None = None,
    files_prefix: str | None = None,
) -> Self:
    """Construct a BucketConfig pointed at Cloudflare R2.

    Fills in the account-specific S3 endpoint and the mandatory region='auto'.
    """
    kwargs: dict[str, str] = {
        "bucket": bucket,
        "endpoint_url": f"https://{account_id}.r2.cloudflarestorage.com",
        "region": "auto",
        "access_key_id": access_key_id,
        "secret_access_key": secret_access_key,
    }
    if key is not None:
        kwargs["key"] = key
    if files_prefix is not None:
        kwargs["files_prefix"] = files_prefix
    return cls(**kwargs)

SnapshotConfig

Bases: BaseModel

How often BucketHead flushes and what it keeps.

BucketEnv

Bases: BaseModel

S3-compatible bucket addressing + credentials.

BucketHeadSettings

Bases: BaseSettings

to_bucket_config

to_bucket_config() -> BucketConfig

Resolve runtime config into a BucketConfig.

Raises ValueError when required fields aren't set. Derives the R2 endpoint from cloudflare.account_id when bucket.endpoint_url is unset and cloud == 'cloudflare-r2'.

Source code in src/buckethead/config/settings.py

def to_bucket_config(self) -> BucketConfig:
    """Resolve runtime config into a BucketConfig.

    Raises ``ValueError`` when required fields aren't set. Derives the
    R2 endpoint from ``cloudflare.account_id`` when ``bucket.endpoint_url``
    is unset and ``cloud == 'cloudflare-r2'``.
    """
    missing: list[str] = []
    if not self.bucket.name:
        missing.append("BUCKETHEAD_BUCKET__NAME")
    if not self.bucket.access_key_id:
        missing.append("BUCKETHEAD_BUCKET__ACCESS_KEY_ID")
    if not self.bucket.secret_access_key:
        missing.append("BUCKETHEAD_BUCKET__SECRET_ACCESS_KEY")
    if missing:
        raise ValueError(f"missing required env vars: {', '.join(missing)}")

    endpoint = self.bucket.endpoint_url
    if (
        endpoint is None
        and self.cloud == "cloudflare-r2"
        and self.cloudflare.account_id
    ):
        endpoint = f"https://{self.cloudflare.account_id}.r2.cloudflarestorage.com"

    return BucketConfig(
        bucket=self.bucket.name,
        key=self.snapshot.key,
        endpoint_url=endpoint,
        region=self.bucket.region,
        access_key_id=self.bucket.access_key_id,
        secret_access_key=self.bucket.secret_access_key,
        initial_branch=self.snapshot.branch,
    )

CloudflareEnv

Bases: BaseModel

Cloudflare-specific knobs.

account_id is dual-purpose: provisioning uses it for API calls, and runtime uses it to derive the S3 endpoint when bucket.endpoint_url is unset and cloud == 'cloudflare-r2'.

OnePasswordEnv

Bases: BaseModel

1Password-specific knobs used by provisioning.

SnapshotEnv

Bases: BaseModel

Where the snapshot lives inside the bucket.

UserConfig

Bases: BaseSettings

load classmethod

load() -> Self

Construct from the TOML file + defaults (read-only view).

Source code in src/buckethead/config/user.py

@classmethod
def load(cls) -> Self:
    """Construct from the TOML file + defaults (read-only view)."""
    return cls()

forget_project

forget_project(
    project: str, *, path: Path | None = None
) -> None

Remove a project mapping (primary + share) and persist.

Source code in src/buckethead/config/user.py

def forget_project(self, project: str, *, path: Path | None = None) -> None:
    """Remove a project mapping (primary + share) and persist."""
    if project in self.projects:
        del self.projects[project]
        unset_project(project, path=path)

bucket_for_project

bucket_for_project(
    project: str, *, path: Path | None = None
) -> str

Return the primary bucket for project; generate and persist on miss.

Source code in src/buckethead/config/user.py

def bucket_for_project(self, project: str, *, path: Path | None = None) -> str:
    """Return the primary bucket for `project`; generate and persist on miss."""
    entry = self.projects.get(project)
    if entry is None:
        bucket = f"{project}-{token_hex(3)}"
        self.projects[project] = bucket
        set_project(project, primary=bucket, path=path)
        return bucket
    if isinstance(entry, str):
        return entry
    return entry["primary"]

configure_share_for_project

configure_share_for_project(
    project: str,
    *,
    share_bucket: str,
    mode: str,
    public_base_url: str | None = None,
    path: Path | None = None,
) -> None

Attach or update share config on an existing project entry.

Raises KeyError if the project isn't present — call bucket_for_project first to create the primary.

Source code in src/buckethead/config/user.py

def configure_share_for_project(
    self,
    project: str,
    *,
    share_bucket: str,
    mode: str,
    public_base_url: str | None = None,
    path: Path | None = None,
) -> None:
    """Attach or update share config on an existing project entry.

    Raises KeyError if the project isn't present — call
    `bucket_for_project` first to create the primary.
    """
    if mode not in ("public", "protected"):
        raise ValueError(f"mode must be 'public' or 'protected', got {mode!r}")
    entry = self.projects.get(project)
    if entry is None:
        raise KeyError(f"project {project!r} not in user config")

    base: dict[str, str] = (
        {"primary": entry} if isinstance(entry, str) else dict(entry)
    )
    base["share"] = share_bucket
    base["share_mode"] = mode
    if public_base_url is None:
        base.pop("share_public_base_url", None)
    else:
        base["share_public_base_url"] = public_base_url
    self.projects[project] = base

    set_project(
        project,
        share=share_bucket,
        share_mode=mode,
        share_public_base_url=public_base_url,
        clear_public_base_url=public_base_url is None,
        path=path,
    )

forget_share_for_project

forget_share_for_project(
    project: str, *, path: Path | None = None
) -> None

Detach the share bucket; demote to string form if only primary remains.

Source code in src/buckethead/config/user.py

def forget_share_for_project(
    self, project: str, *, path: Path | None = None
) -> None:
    """Detach the share bucket; demote to string form if only `primary` remains."""
    entry = self.projects.get(project)
    if not isinstance(entry, dict):
        return
    remaining = {k: v for k, v in entry.items() if k not in _SHARE_FIELDS}
    self.projects[project] = (
        remaining["primary"] if list(remaining.keys()) == ["primary"] else remaining
    )
    unset_share_for_project(project, path=path)

register_resolver

register_resolver(
    scheme: str, resolver: Callable[[str], str]
) -> None

Register a resolver for URL-shaped refs with the given scheme.

The resolver receives the full ref (without the secret: prefix) and returns the resolved secret value.

Source code in src/buckethead/config/secret_refs.py

def register_resolver(scheme: str, resolver: Callable[[str], str]) -> None:
    """Register a resolver for URL-shaped refs with the given scheme.

    The resolver receives the full ref (without the `secret:` prefix)
    and returns the resolved secret value.
    """
    _RESOLVERS[scheme] = resolver

resolve_value

resolve_value(value: str) -> str

Resolve a TOML value. Only secret:<scheme>://... strings are resolved — everything else (including bare op://... or https://) passes through as a literal.

Source code in src/buckethead/config/secret_refs.py

def resolve_value(value: str) -> str:
    """Resolve a TOML value. Only `secret:<scheme>://...` strings are
    resolved — everything else (including bare `op://...` or `https://`)
    passes through as a literal."""
    if not value.startswith(_SECRET_PREFIX):
        return value
    ref = value[len(_SECRET_PREFIX) :]
    match = _SCHEME_RE.match(ref)
    if match is None:
        raise ValueError(
            f"value starts with {_SECRET_PREFIX!r} but is not a "
            f"`<scheme>://...` ref: {value!r}"
        )
    scheme = match.group(1)
    resolver = _RESOLVERS.get(scheme)
    if resolver is None:
        raise ValueError(
            f"no resolver registered for scheme {scheme!r} "
            f"(ref: {ref!r}); register one with "
            f"buckethead.config.secret_refs.register_resolver()"
        )
    return resolver(ref)