# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Read a repository's metrics from whichever forge hosts it.
Answers one question for one repository: how many accounts follow it, when it
was created, when it last shipped, and when it was last touched. GitHub, GitLab
(on any instance) and Forgejo or Gitea (likewise) each expose that through a
different API, and {func}`repo_metrics` picks the right one from the URL's
host.
One call per repository on every forge, which is what lets a single sampler
collect every metric {mod}`repomatic.metrics.store` records rather than one call per
metric family.
"""
from __future__ import annotations
import urllib.parse
from dataclasses import dataclass
from enum import Enum
from functools import reduce
from operator import getitem
from ..github.gh import gh_graphql
from ..http import get_json_soft
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterator, Mapping
from typing import Any
[docs]
class Forge(Enum):
"""Forge software a host runs, which owns how to read its API.
Supporting a fourth forge means a new member plus its entry in
{data}`REST_ENDPOINTS`, which keeps the config-side validation and the
dispatch from drifting: {func}`forge_of` resolves declared values through
the member lookup, so a `[tool.repomatic.metrics] forges` entry naming
anything else fails there rather than at sampling time.
"""
FORGEJO = "forgejo"
GITHUB = "github"
GITLAB = "gitlab"
[docs]
def metrics(self, host: str, path: str) -> ForgeMetrics | None:
"""Read one repository through this forge's API.
:param host: The instance's hostname. Unused on GitHub, whose one
host `gh` already addresses.
:param path: The repository's `owner/name` or namespace path.
:return: The repository's metrics, or `None` when unreadable.
:raises RuntimeError: When a GitHub call fails.
"""
if self is Forge.GITHUB:
return github_metrics(path)
return rest_metrics(REST_ENDPOINTS[self], host, path)
[docs]
@dataclass(frozen=True)
class RestEndpoints:
"""Where the REST API of a forge answers each reading of one repository.
GitLab and Forgejo answer the same four questions in four calls each: the
repository object, then the newest release, tag and commit. Only the
paths and the field names differ, so one reader,
{func}`rest_metrics`, serves both through this table.
"""
repository: str
"""URL of the repository object, with ``{host}`` and ``{path}`` fields."""
quote_path: bool
"""Whether the path travels URL-encoded, as one segment."""
stars: str
"""Field of the repository object holding its star count."""
releases: str
"""Suffix of the endpoint listing the newest release first, one per page."""
release_date: tuple[str, ...]
"""Keys leading to the date of a listed release."""
tags: str
"""Suffix of the endpoint listing the newest tag first, one per page."""
tag_date: tuple[str, ...]
"""Keys leading to the date of a listed tag."""
commits: str
"""Suffix of the endpoint listing the newest commit first, one per page."""
commit_date: tuple[str, ...]
"""Keys leading to the date of a listed commit."""
FORGE_APIS: dict[str, Forge] = {
"codeberg.org": Forge.FORGEJO,
"github.com": Forge.GITHUB,
"gitlab.com": Forge.GITLAB,
}
"""Forge software each known host runs, which is what selects the API to call.
Never guessed from the host name: an unknown host raises instead, so a subject
landing on a fourth kind of forge has to declare how to read it rather than
silently sampling nothing. Extend it through
`[tool.repomatic.metrics] forges`, which a repository uses to name the
self-hosted instances it tracks (`salsa.debian.org` runs GitLab,
`gitlab.archlinux.org` too).
"""
GITHUB_HOST = "github.com"
"""The one host whose deep collector exists.
The star-curve reconstruction reads a star-history endpoint only GitHub
serves, so a subject elsewhere is skipped by it rather than failed.
"""
GITHUB_METRICS_QUERY = """
query($owner: String!, $name: String!) {
repository(owner: $owner, name: $name) {
createdAt
stargazerCount
latestRelease { publishedAt }
defaultBranchRef { target { ... on Commit { committedDate } } }
refs(refPrefix: "refs/tags/", first: 1,
orderBy: {field: TAG_COMMIT_DATE, direction: DESC}) {
nodes {
target {
... on Commit { committedDate }
... on Tag { target { ... on Commit { committedDate } } }
}
}
}
}
}
"""
"""Reads a repository's whole metric set in one call.
One call where REST needs four, and correct where REST is not: `/tags` answers
in an order nobody should assume, so a fallback trusting it can date a live
project a decade into the past. Ordering on `TAG_COMMIT_DATE` states the
question instead of hoping the default matches it.
The commit date is read off the default branch rather than from the
repository's `pushedAt`, which any push to any branch bumps.
"""
[docs]
@dataclass(frozen=True)
class ForgeMetrics:
"""One repository's metrics, as any forge reports them."""
stars: int
"""Count of accounts following the repository on its own forge."""
created: str | None = None
"""ISO date the repository was opened.
The one date a star count is known to be zero, which is what gives a
history an origin and a by-age chart something to align on.
"""
release: str | None = None
"""ISO date of the newest release or tag, `None` when a project has neither."""
release_source: str | None = None
"""Where {attr}`release` came from: a `release` object, or a bare `tag`.
Recorded because the two are not the same claim. A release is something the
project announced; a tag is only the newest thing it labelled, which is the
closest available answer for the many projects that never cut a release.
"""
commit: str | None = None
"""ISO date of the newest commit on the default branch.
The half of the activity reading that stays true for a rolling repository.
A widely used package archive can go a decade without tagging a release
while being committed to several times a day: a release date alone would
report it as long dead.
"""
[docs]
def readings(self) -> Iterator[tuple[str, str]]:
"""Yield each metric this reading carries, as `(metric id, value)`.
The bridge between a typed forge answer and the metric store, which
holds every value as text. A metric the forge did not answer yields
nothing rather than an empty string, so a project with no release adds
no row instead of a blank one.
{attr}`created` is deliberately absent: it is not a metric but the
origin of one, recorded by the sampler as a `stars` reading of zero on
that date.
"""
yield "stars", str(self.stars)
if self.commit:
yield "commit", self.commit
if self.release:
yield "release", self.release
yield "release_source", self.release_source or "release"
[docs]
def split_repo_url(url: str) -> tuple[str, str]:
"""Split a repository URL into its host and its owner/name path.
:param url: An `https://host/owner/name` repository URL.
:return: The `(host, path)` pair.
:raises ValueError: When the URL carries no host or no owner/name path.
"""
host, _, path = url.removeprefix("https://").removeprefix("http://").partition("/")
path = path.rstrip("/").removesuffix(".git")
owner, _, name = path.partition("/")
if not host or not owner or not name:
msg = f"Not an https://host/owner/name repository URL: {url!r}"
raise ValueError(msg)
return host, path
[docs]
def canonical_url(subject: str) -> str:
"""Normalize a configured subject into the URL the store keys on.
A bare `owner/name` is GitHub, which is what a repository declaring a
handful of peers writes. Anything else is already a URL and only needs its
trailing decoration removed. One spelling in the store keeps a subject
addressable whichever way its configuration named it.
:param subject: An `owner/name` slug or a full repository URL.
:return: The canonical `https://host/owner/name` URL.
:raises ValueError: When neither shape parses.
"""
if "://" not in subject:
subject = f"https://{GITHUB_HOST}/{subject.strip('/')}"
host, path = split_repo_url(subject)
return f"https://{host}/{path}"
[docs]
def forge_of(url: str, extra_forges: Mapping[str, str] | None = None) -> Forge:
"""Name the forge software running the host of *url*.
:param url: An `https://host/owner/name` repository URL.
:param extra_forges: Host-to-forge entries a repository declared for the
self-hosted instances it tracks, merged over {data}`FORGE_APIS`.
:return: The {class}`Forge` the host runs.
:raises ValueError: When the host is not declared anywhere, which is
deliberate: a silently unsampled subject is worse than a loud one.
"""
host, _path = split_repo_url(url)
forges: dict[str, Forge | str] = {**FORGE_APIS, **(extra_forges or {})}
forge = forges.get(host)
if forge is None:
known = ", ".join(sorted(forges))
msg = (
f"Unknown forge host {host!r}. Declare it in "
f"[tool.repomatic.metrics] forges. Known hosts: {known}."
)
raise ValueError(msg)
try:
return Forge(forge)
except ValueError:
supported = ", ".join(sorted(member.value for member in Forge))
msg = f"Unsupported forge {forge!r} for {host!r}: pick one of {supported}."
raise ValueError(msg) from None
[docs]
def newest_dated(
release: str | None,
tag: str | None,
) -> tuple[str | None, str | None]:
"""Pick whichever of a project's newest release and newest tag is more recent.
Not a preference for releases: plenty of projects carry a tag newer than
their latest release object, some by close to a year, so always reading the
release would report them as idle. ISO dates compare as strings, which is
the whole of the arithmetic here.
:param release: ISO date of the newest release, or `None`.
:param tag: ISO date of the newest tag, or `None`.
:return: The `(date, source)` pair, both `None` when a project has neither.
"""
if release and tag:
return (release, "release") if release >= tag else (tag, "tag")
if release:
return release, "release"
if tag:
return tag, "tag"
return None, None
[docs]
def forge_json(url: str) -> Any | None:
"""Read one JSON document from a forge's public API.
Covers every forge but GitHub, whose authentication `gh` already carries.
The instances read here (GitLab and Forgejo) serve their project metadata
to anonymous callers, so no token is involved and none is asked for.
The request identifies itself with {data}`repomatic.http.USER_AGENT`, like
every fetch of {mod}`repomatic.http`. Several self-hosted GitLab instances
answer a browser identity with an HTML page instead of the JSON payload,
and nothing then errors: the subject only goes missing from the readings.
:param url: The API endpoint to read.
:return: The parsed payload, or `None` when the call or the parse failed.
"""
return get_json_soft(url, f"Could not read {url}")
[docs]
def github_metrics(path: str) -> ForgeMetrics:
"""Read a GitHub repository through {data}`GITHUB_METRICS_QUERY`.
:param path: The repository's `owner/name` path.
:return: The repository's metrics.
:raises RuntimeError: When the `gh` call fails.
"""
owner, _, name = path.partition("/")
repo = gh_graphql(GITHUB_METRICS_QUERY, owner=owner, name=name)["repository"]
release = repo["latestRelease"]
tag = None
for node in repo["refs"]["nodes"]:
# A lightweight tag points straight at its commit, an annotated one at
# a tag object wrapping it, and the query asks for both shapes.
target = node["target"]
nested = target.get("target") or {}
committed = target.get("committedDate") or nested.get("committedDate")
if committed:
tag = committed[:10]
dated, source = newest_dated(release["publishedAt"][:10] if release else None, tag)
branch = repo["defaultBranchRef"]
return ForgeMetrics(
stars=repo["stargazerCount"],
created=str(repo["createdAt"])[:10],
release=dated,
release_source=source,
commit=branch["target"]["committedDate"][:10] if branch else None,
)
REST_ENDPOINTS: dict[Forge, RestEndpoints] = {
Forge.FORGEJO: RestEndpoints(
repository="https://{host}/api/v1/repos/{path}",
quote_path=False,
stars="stars_count",
releases="/releases?limit=1",
release_date=("published_at",),
tags="/tags?limit=1",
tag_date=("commit", "created"),
commits="/commits?limit=1",
commit_date=("commit", "committer", "date"),
),
Forge.GITLAB: RestEndpoints(
repository="https://{host}/api/v4/projects/{path}",
quote_path=True,
stars="star_count",
releases="/releases?per_page=1",
release_date=("released_at",),
# Ordering spelled out rather than inherited: it is GitLab's current
# default for tags, but the whole point of asking is to not depend on
# that.
tags="/repository/tags?per_page=1&order_by=updated&sort=desc",
tag_date=("commit", "created_at"),
commits="/repository/commits?per_page=1",
commit_date=("committed_date",),
),
}
"""How to read each forge served over REST, which is every forge but GitHub.
Gitea instances read as Forgejo, whose API they share.
"""
[docs]
def rest_metrics(endpoints: RestEndpoints, host: str, path: str) -> ForgeMetrics | None:
"""Read a repository through the REST API of a forge, on any instance.
:param endpoints: Where that forge answers each reading.
:param host: The instance's hostname.
:param path: The repository's `owner/name` or namespace path.
:return: The repository's metrics, or `None` when unreadable.
"""
repository = endpoints.repository.format(
host=host,
path=urllib.parse.quote(path, safe="") if endpoints.quote_path else path,
)
payload = forge_json(repository)
if not payload:
return None
def newest(suffix: str, keys: tuple[str, ...]) -> str | None:
"""The date of the newest item a listing holds, `None` when empty."""
items = forge_json(f"{repository}{suffix}")
if not items:
return None
# Indexed rather than looked up with a default: an answer of a shape
# nobody anticipated raises, and costs its subject one row.
timestamp: str = reduce(getitem, keys, items[0])
return timestamp[:10]
dated, source = newest_dated(
newest(endpoints.releases, endpoints.release_date),
newest(endpoints.tags, endpoints.tag_date),
)
return ForgeMetrics(
stars=payload[endpoints.stars],
created=str(payload["created_at"])[:10] if payload.get("created_at") else None,
release=dated,
release_source=source,
commit=newest(endpoints.commits, endpoints.commit_date),
)
[docs]
def repo_metrics(
url: str,
extra_forges: Mapping[str, str] | None = None,
) -> ForgeMetrics | None:
"""Read one repository, through whichever API its host speaks.
:param url: An `https://host/owner/name` repository URL.
:param extra_forges: Host-to-forge entries for self-hosted instances.
:return: The repository's metrics, or `None` when the forge could not be
read.
:raises ValueError: When the host declares no forge.
:raises RuntimeError: When a GitHub call fails.
"""
host, path = split_repo_url(url)
return forge_of(url, extra_forges).metrics(host, path)