89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
"""Build read-only release dashboard snapshots."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
from .catalog import collect_catalog_snapshot
|
|
from .contracts import collect_contracts
|
|
from .git_state import collect_repository_snapshot, read_pyproject_version
|
|
from .migrations import collect_migration_snapshots
|
|
from .model import DashboardSummary, ReleaseDashboard, RepositorySnapshot
|
|
from .workspace import META_ROOT, core_root, load_repository_specs, resolve_workspace_root
|
|
|
|
|
|
def build_dashboard(
|
|
*,
|
|
workspace_root: Path | str | None = None,
|
|
target_version: str | None = None,
|
|
online: bool = False,
|
|
check_remote_tags: bool = False,
|
|
check_public_catalog: bool | None = None,
|
|
include_migrations: bool = False,
|
|
include_website: bool = False,
|
|
include_contracts: bool = True,
|
|
channel: str = "stable",
|
|
) -> ReleaseDashboard:
|
|
resolved_workspace = resolve_workspace_root(workspace_root)
|
|
if check_public_catalog is None:
|
|
check_public_catalog = online
|
|
if target_version is None:
|
|
target_version = read_pyproject_version(core_root(resolved_workspace))
|
|
target_tag = f"v{target_version}" if target_version else None
|
|
|
|
repositories = tuple(
|
|
collect_repository_snapshot(spec, workspace_root=resolved_workspace, target_tag=target_tag, online=check_remote_tags)
|
|
for spec in load_repository_specs(include_website=include_website)
|
|
)
|
|
catalog = collect_catalog_snapshot(workspace_root=resolved_workspace, channel=channel, check_public=check_public_catalog)
|
|
migrations = collect_migration_snapshots() if include_migrations else ()
|
|
contracts = collect_contracts(repositories) if include_contracts else ()
|
|
return ReleaseDashboard(
|
|
generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
|
|
meta_root=str(META_ROOT),
|
|
workspace_root=str(resolved_workspace),
|
|
target_version=target_version,
|
|
target_tag=target_tag,
|
|
online=online or check_remote_tags or check_public_catalog,
|
|
include_migrations=include_migrations,
|
|
summary=summarize(repositories, target_tag=target_tag),
|
|
repositories=repositories,
|
|
catalog=catalog,
|
|
migrations=migrations,
|
|
contracts=contracts,
|
|
)
|
|
|
|
|
|
def summarize(repositories: tuple[RepositorySnapshot, ...], *, target_tag: str | None) -> DashboardSummary:
|
|
missing_count = sum(1 for repo in repositories if not repo.exists or not repo.is_git)
|
|
dirty_count = sum(1 for repo in repositories if repo.dirty)
|
|
ahead_count = sum(1 for repo in repositories if repo.ahead)
|
|
behind_count = sum(1 for repo in repositories if repo.behind)
|
|
no_head_count = sum(1 for repo in repositories if repo.exists and repo.is_git and not repo.has_head)
|
|
error_count = sum(1 for repo in repositories if repo.errors)
|
|
safe_directory_count = sum(1 for repo in repositories if repo.safe_directory_required)
|
|
local_target_tag_missing_count = (
|
|
sum(1 for repo in repositories if repo.local_target_tag_exists is False and repo.versions.primary == target_tag.removeprefix("v"))
|
|
if target_tag
|
|
else 0
|
|
)
|
|
if missing_count or behind_count or safe_directory_count:
|
|
status = "blocked"
|
|
elif dirty_count or ahead_count or no_head_count or error_count or local_target_tag_missing_count:
|
|
status = "attention"
|
|
else:
|
|
status = "ready"
|
|
return DashboardSummary(
|
|
repository_count=len(repositories),
|
|
missing_count=missing_count,
|
|
dirty_count=dirty_count,
|
|
ahead_count=ahead_count,
|
|
behind_count=behind_count,
|
|
no_head_count=no_head_count,
|
|
error_count=error_count,
|
|
safe_directory_count=safe_directory_count,
|
|
local_target_tag_missing_count=local_target_tag_missing_count,
|
|
status=status,
|
|
)
|