libp2p.observability.oso package

Submodules

libp2p.observability.oso.cli module

CLI entrypoint for OSO health reporting.

libp2p.observability.oso.cli.build_parser() → ArgumentParser: Build CLI parser.

libp2p.observability.oso.cli.main() → int: Run CLI and return process status code.

libp2p.observability.oso.dependency_graph module

Dependency graph extraction and normalization utilities.

libp2p.observability.oso.dependency_graph.build_dependency_graph(pyproject_path: Path, repository: str) → DependencyGraph: Build a normalized dependency graph from pyproject.toml.

libp2p.observability.oso.generators module

Dependency graph artifact generators used by operational scripts.

libp2p.observability.oso.generators.generate_direct_graph_artifacts(repo_root: Path, output_dir: Path | None = None, repository: str = 'https://github.com/libp2p/py-libp2p') → dict[str, Path]: Generate direct dependency artifacts (json/dot/mmd/md).

libp2p.observability.oso.metrics module

Metric calculators for OSO observability report.

libp2p.observability.oso.metrics.calculate_contributor_trend(commits: list[CommitInfo]) → ContributorTrendMetric: Calculate contributor trend metrics.

libp2p.observability.oso.metrics.calculate_dependency_topology(graph: DependencyGraph) → DependencyTopologyMetric: Calculate topology metrics from dependency graph.

libp2p.observability.oso.metrics.calculate_issue_responsiveness(issues: list[IssueInfo]) → IssueResponsivenessMetric: Calculate issue responsiveness metrics.

libp2p.observability.oso.metrics.calculate_popularity(popularity: PopularityInfo) → PopularityMetric: Normalize popularity metrics.

libp2p.observability.oso.metrics.calculate_release_cadence(releases: list[ReleaseInfo]) → ReleaseCadenceMetric: Calculate release cadence metrics.

libp2p.observability.oso.metrics.calculate_security_proxy(graph: DependencyGraph, vulnerable_packages: Iterable[str]) → SecurityProxyMetric: Calculate security proxy from dependency graph + vulnerability lookups.

libp2p.observability.oso.metrics.collect_rcmgr_baseline() → RcmgrSnapshot: Collect current rcmgr baseline capability status.

libp2p.observability.oso.models module

Typed models for OSO observability reporting.

class libp2p.observability.oso.models.ContributorTrendMetric(commits_considered: int, unique_contributors: int, weekly_commit_counts: dict[str, int], contributor_logins: list[str] = <factory>)

Bases: object

Contributor activity trend metric model.

commits_considered: int

contributor_logins: list[str]

unique_contributors: int

weekly_commit_counts: dict[str, int]

class libp2p.observability.oso.models.DependencyEntry(name: str, version_spec: str | None, condition: str | None, dependency_type: str)

Bases: object

A single dependency declaration entry.

condition: str | None

dependency_type: str

name: str

version_spec: str | None

class libp2p.observability.oso.models.DependencyGraph(project: ~libp2p.observability.oso.models.ProjectMetadata, dependencies: list[~libp2p.observability.oso.models.DependencyEntry], optional_dependencies: dict[str, list[~libp2p.observability.oso.models.DependencyEntry]], edges: list[tuple[str, str]] = <factory>)

Bases: object

Dependency graph model used for health analysis.

dependencies: list[DependencyEntry]

edges: list[tuple[str, str]]

optional_dependencies: dict[str, list[DependencyEntry]]

project: ProjectMetadata

class libp2p.observability.oso.models.DependencyTopologyMetric(direct_dependencies: int, optional_dependencies: int, total_dependency_entries: int, unique_packages: int, duplicate_packages: list[str], max_out_degree: int, most_connected_packages: list[str], runtime_package_versions: list[str] = <factory>)

Bases: object

Dependency graph topology metrics.

direct_dependencies: int

duplicate_packages: list[str]

max_out_degree: int

most_connected_packages: list[str]

optional_dependencies: int

runtime_package_versions: list[str]

total_dependency_entries: int

unique_packages: int

class libp2p.observability.oso.models.HealthMetrics(dependency_topology: DependencyTopologyMetric, release_cadence: ReleaseCadenceMetric, issue_responsiveness: IssueResponsivenessMetric, contributor_trend: ContributorTrendMetric, popularity: PopularityMetric, security_proxy: SecurityProxyMetric)

Bases: object

Aggregated V1 health metrics.

contributor_trend: ContributorTrendMetric

dependency_topology: DependencyTopologyMetric

issue_responsiveness: IssueResponsivenessMetric

popularity: PopularityMetric

release_cadence: ReleaseCadenceMetric

security_proxy: SecurityProxyMetric

class libp2p.observability.oso.models.HealthReport(generated_at: str, project: ProjectMetadata, source_status: SourceStatus, rcmgr_baseline: RcmgrSnapshot, metrics: HealthMetrics)

Bases: object

Top-level report model persisted to JSON/Markdown.

generated_at: str

metrics: HealthMetrics

project: ProjectMetadata

rcmgr_baseline: RcmgrSnapshot

source_status: SourceStatus

to_dict() → dict[str, Any]: Serialize report as a plain dict.

class libp2p.observability.oso.models.IssueResponsivenessMetric(issues_considered: int, median_hours_to_first_response: float | None, median_hours_to_close: float | None, open_issues: int, closed_issues: int)

Bases: object

Issue responsiveness metric model.

closed_issues: int

issues_considered: int

median_hours_to_close: float | None

median_hours_to_first_response: float | None

open_issues: int

class libp2p.observability.oso.models.PopularityMetric(stars: int | None, forks: int | None, watchers: int | None)

Bases: object

Popularity metric model.

forks: int | None

stars: int | None

watchers: int | None

class libp2p.observability.oso.models.ProjectMetadata(name: str, version: str, python_version: str, repository: str)

Bases: object

Project identity metadata.

name: str

python_version: str

repository: str

version: str

class libp2p.observability.oso.models.RcmgrSnapshot(metrics_available: bool, prometheus_available: bool, monitoring_available: bool, exported_metric_names: list[str] = <factory>)

Bases: object

Snapshot view of local rcmgr runtime-related metrics.

exported_metric_names: list[str]

metrics_available: bool

monitoring_available: bool

prometheus_available: bool

class libp2p.observability.oso.models.ReleaseCadenceMetric(releases_considered: int, average_days_between_releases: float | None, median_days_between_releases: float | None, last_release_at: str | None)

Bases: object

Release cadence metric model.

average_days_between_releases: float | None

last_release_at: str | None

median_days_between_releases: float | None

releases_considered: int

class libp2p.observability.oso.models.SecurityProxyMetric(duplicate_dependency_specs: list[str], osv_vulnerable_packages: list[str], dependency_count: int)

Bases: object

Security proxy metric model.

dependency_count: int

duplicate_dependency_specs: list[str]

osv_vulnerable_packages: list[str]

class libp2p.observability.oso.models.SourceStatus(github: str, oso: str, rcmgr: str, notes: list[str] = <factory>)

Bases: object

Data-source status for partial-failure reporting.

github: str

notes: list[str]

oso: str

rcmgr: str

libp2p.observability.oso.models.utc_now_iso() → str: Return an ISO8601 UTC timestamp.

libp2p.observability.oso.providers module

External data providers for GitHub and OSO.

class libp2p.observability.oso.providers.CommitInfo(committed_at: str, author_login: str | None)

Bases: object

Commit metadata for contributor trends.

author_login: str | None

committed_at: str

class libp2p.observability.oso.providers.GitHubProvider(repo_slug: str, token: str | None = None)

Bases: object

GitHub API provider with typed normalization.

fetch_commits(limit: int = 100) → list[CommitInfo]: Fetch latest commits.

fetch_issues(limit: int = 100) → list[IssueInfo]: Fetch issue metadata and first response proxy.

fetch_popularity() → PopularityInfo: Fetch repo popularity counters.

fetch_releases(limit: int = 30) → list[ReleaseInfo]: Fetch latest releases.

class libp2p.observability.oso.providers.IssueInfo(created_at: str, closed_at: str | None, first_response_at: str | None)

Bases: object

Issue metadata used for responsiveness metrics.

closed_at: str | None

created_at: str

first_response_at: str | None

class libp2p.observability.oso.providers.OsoProvider(token: str | None = None)

Bases: object

OSO GraphQL provider.

query(query: str, variables: dict[str, Any] | None = None) → Any: Run a GraphQL query against OSO endpoint.

class libp2p.observability.oso.providers.PopularityInfo(stars: int | None, forks: int | None, watchers: int | None)

Bases: object

Popularity counters from GitHub.

forks: int | None

stars: int | None

watchers: int | None

class libp2p.observability.oso.providers.ReleaseInfo(published_at: str)

Bases: object

Release metadata needed for cadence calculations.

published_at: str

libp2p.observability.oso.providers.get_installed_package_versions() → dict[str, str]: Return installed package versions from current Python environment.

libp2p.observability.oso.providers.parse_iso_datetime(value: str) → datetime: Parse GitHub/ISO timestamp with Z support.

libp2p.observability.oso.providers.query_osv_vulnerabilities(package_name: str, ecosystem: str = 'PyPI') → bool: Return whether OSV reports vulnerabilities for package/version.

libp2p.observability.oso.providers.query_osv_vulnerabilities_for_version(package_name: str, version: str, ecosystem: str = 'PyPI') → bool: Return whether OSV reports vulnerabilities for exact package version.

libp2p.observability.oso.report module

Report rendering and persistence helpers.

libp2p.observability.oso.report.render_markdown_report(report: HealthReport) → str: Render health report as concise markdown.

libp2p.observability.oso.report.write_json_report(report: HealthReport, output_path: Path) → None: Write health report as JSON.

libp2p.observability.oso.report.write_markdown_report(report: HealthReport, output_path: Path) → None: Write report in Markdown format.

libp2p.observability.oso.service module

Orchestration service for OSO health reports.

libp2p.observability.oso.service.collect_health_report(repo_root: Path, repo_slug: str = 'libp2p/py-libp2p', github_token: str | None = None, oso_token: str | None = None) → HealthReport: Collect a full health report for py-libp2p.

libp2p.observability.oso.transitive_graph module

Transitive dependency graph generation helpers.

libp2p.observability.oso.transitive_graph.build_transitive_graph(root_package: str, direct_dependencies: list[str], max_depth: int = 10) → dict[str, Any]: Build transitive graph by recursively resolving dependencies.

libp2p.observability.oso.transitive_graph.generate_transitive_graph_artifacts(repo_root: Path, output_dir: Path | None = None) → dict[str, Path]: Generate transitive dependency artifacts (json/dot/mmd).

libp2p.observability.oso.transitive_graph.get_installed_packages() → dict[str, str]: Get installed packages and versions from pip.

libp2p.observability.oso.transitive_graph.get_package_dependencies(package_name: str) → list[str]: Get direct dependencies of package from pip show.

Module contents

OSO observability package for py-libp2p.

class libp2p.observability.oso.HealthReport(generated_at: str, project: ProjectMetadata, source_status: SourceStatus, rcmgr_baseline: RcmgrSnapshot, metrics: HealthMetrics)

Bases: object

Top-level report model persisted to JSON/Markdown.

generated_at: str

metrics: HealthMetrics

project: ProjectMetadata

rcmgr_baseline: RcmgrSnapshot

source_status: SourceStatus

to_dict() → dict[str, Any]: Serialize report as a plain dict.

libp2p.observability.oso.collect_health_report(repo_root: Path, repo_slug: str = 'libp2p/py-libp2p', github_token: str | None = None, oso_token: str | None = None) → HealthReport: Collect a full health report for py-libp2p.

libp2p.observability.oso.generate_direct_graph_artifacts(repo_root: Path, output_dir: Path | None = None, repository: str = 'https://github.com/libp2p/py-libp2p') → dict[str, Path]: Generate direct dependency artifacts (json/dot/mmd/md).

libp2p.observability.oso.generate_transitive_graph_artifacts(repo_root: Path, output_dir: Path | None = None) → dict[str, Path]: Generate transitive dependency artifacts (json/dot/mmd).