buckethead.observability.status

Structured status reporting for the lifecycle. Pass a StatusReporter subclass (e.g. RichStatusReporter) as BucketSQLite(..., status_reporter=...) to receive live updates on restore, flushes, errors, and shutdown.

buckethead.observability.status

Observational status for BucketHead projects and running instances.

Two layers:

1. Structured models + library functions — ProjectStatus, ConnectionInfo, ActivitySummary carry the data. project_status(...) and list_project_statuses(...) return them. Purely observational: no DB startup, no mutation, no data reads.

2. Reporter Protocol + implementations — StatusReporter fires from BucketSQLite.start() / stop() for embedded consumers. RichStatusReporter is the default (TTY-aware); SilentReporter suppresses output. Both consume the same structured models.

The CLI (buckethead status) is a thin wrapper: construct backends via factories, call the library functions, hand results to RichStatusReporter for rendering.

ConnectionInfo

Bases: BaseModel

Fields a running BucketSQLite can report at start-time.

ActivitySummary

Bases: BaseModel

Fields a running BucketSQLite can report at stop-time.

ProjectStatus

Bases: BaseModel

Observational snapshot of a provisioned project.

Every probe field is bool | None: None means the probe wasn't attempted (e.g. because a prerequisite failed), True/False means the probe ran and returned a definite answer.

SilentReporter

No-op reporter. Use when embedding BucketSQLite in code that should not print to stdout.

RichStatusReporter

RichStatusReporter(
    *,
    project: str | None = None,
    console: Console | None = None,
    quiet: bool | None = None,
)

Rich-formatted reporter. Default for BucketSQLite.

quiet suppresses all output (falls back to BUCKETHEAD_QUIET=1 when unset). project enables the "project not provisioned" nudge at start-time by consulting user config; leave it None for library callers that don't map to a BucketHead project.

Source code in src/buckethead/observability/status.py

def __init__(
    self,
    *,
    project: str | None = None,
    console: Console | None = None,
    quiet: bool | None = None,
) -> None:
    self._project = project
    self._console = console or Console()
    if quiet is None:
        quiet = os.environ.get("BUCKETHEAD_QUIET") == "1"
    self._quiet = quiet

project_status

project_status(
    project: str,
    *,
    cloud: BucketProvider,
    secrets: SecretStore,
    user_config: UserConfig | None = None,
    item_title_prefix: str = "BucketHead R2",
) -> ProjectStatus

Non-destructive status probe for a single project.

Checks (in order, short-circuiting on each failure): 1. Project name present in user config's projects map. 2. Secret-store item exists at <prefix> <project>. 3. Bucket exists in the cloud provider.

Source code in src/buckethead/observability/status.py

def project_status(
    project: str,
    *,
    cloud: BucketProvider,
    secrets: SecretStore,
    user_config: UserConfig | None = None,
    item_title_prefix: str = "BucketHead R2",
) -> ProjectStatus:
    """Non-destructive status probe for a single project.

    Checks (in order, short-circuiting on each failure):
    1. Project name present in user config's `projects` map.
    2. Secret-store item exists at `<prefix> <project>`.
    3. Bucket exists in the cloud provider.
    """
    cfg = user_config or UserConfig.load()
    status = ProjectStatus(
        project=project,
        in_user_config=project in cfg.projects,
    )
    if not status.in_user_config:
        return status

    bucket = cfg.bucket_for_project(project)
    item_title = f"{item_title_prefix} {project}"
    status.bucket = bucket
    status.item_title = item_title

    try:
        status.credentials_exist = secrets.item_exists(item_title)
    except Exception as e:
        status.errors.append(f"secret store probe failed: {e}")
        return status

    try:
        status.bucket_reachable = cloud.bucket_exists(bucket)
    except Exception as e:
        status.errors.append(f"bucket probe failed: {e}")
    return status

list_project_statuses

list_project_statuses(
    *,
    cloud: BucketProvider,
    secrets: SecretStore,
    user_config: UserConfig | None = None,
    item_title_prefix: str = "BucketHead R2",
) -> list[ProjectStatus]

Probe every project in user config, sorted by project name.

Returns an empty list when no projects are configured.

Source code in src/buckethead/observability/status.py

def list_project_statuses(
    *,
    cloud: BucketProvider,
    secrets: SecretStore,
    user_config: UserConfig | None = None,
    item_title_prefix: str = "BucketHead R2",
) -> list[ProjectStatus]:
    """Probe every project in user config, sorted by project name.

    Returns an empty list when no projects are configured.
    """
    cfg = user_config or UserConfig.load()
    return [
        project_status(
            p,
            cloud=cloud,
            secrets=secrets,
            user_config=cfg,
            item_title_prefix=item_title_prefix,
        )
        for p in sorted(cfg.projects)
    ]