gitea.cli.watch.scopes

scopes

What a watch run watches, and how it reduces each of those to snapshots.

Two commands walk the same ground. watch list compares what is there now against the cache and reports the difference; watch advance records what is there now and reports nothing about it. They differ in what they do with the snapshots and in nothing else - the options naming the scopes, the way a project is resolved against a repository, the pages walked, and the fields kept are one behaviour, and live here so the two commands cannot come to disagree about which issues a scope holds or which key it is cached under.

Classes

gitea.cli.watch.scopes.Scope dataclass

Scope(
    key: str, repository: str | None, project_id: int | None
)

One thing being watched, and the key its snapshots are cached under.

Attributes:

Name	Type	Description
key	str	Key the scope's snapshots are recorded under in the cache.
repository	str | None	Repository the scope is narrowed to, or None for the owner.
project_id	int | None	Project the scope watches the board of, or None to watch the open issues of the repository instead.

Functions:

gitea.cli.watch.scopes.build_scopes

build_scopes(
    owner: str,
    repositories: list[str],
    project_ids: list[int],
    command_name: str,
) -> list[Scope]

Work out what a run watches from the options naming it.

Every repository named is a scope of its own, and so is every project, which is how one invocation reports what changed across several repositories and boards at once. A project is resolved the way every other project command resolves one: against the repository when one is named, and against the owner itself when none is.

Parameters:

Name	Type	Description	Default
owner	str	The user or organization owning what is watched.	required
repositories	list[str]	The repositories named, in the order they were named.	required
project_ids	list[int]	The projects named, in the order they were named.	required
command_name	str	The command being run, so a refusal names the invocation the user typed rather than whichever of the two this code is shared with.	required

Returns:

Type	Description
list[Scope]	One scope per repository and per project, repositories first, and one
list[Scope]	scope per key: naming the same repository twice watches it once, rather
list[Scope]	than fetching it twice and comparing the second fetch against the same
list[Scope]	recorded snapshots as the first.

Raises:

Type	Description
CommandError	If nothing was named to watch, or if projects were named alongside more than one repository, which leaves no single scope for them to be resolved against.

Source code in src/gitea/cli/watch/scopes.py

def build_scopes(owner: str, repositories: list[str], project_ids: list[int], command_name: str) -> list[Scope]:
    """Work out what a run watches from the options naming it.

    Every repository named is a scope of its own, and so is every project, which
    is how one invocation reports what changed across several repositories and
    boards at once. A project is resolved the way every other `project` command
    resolves one: against the repository when one is named, and against the owner
    itself when none is.

    Args:
        owner: The user or organization owning what is watched.
        repositories: The repositories named, in the order they were named.
        project_ids: The projects named, in the order they were named.
        command_name: The command being run, so a refusal names the invocation
            the user typed rather than whichever of the two this code is shared
            with.

    Returns:
        One scope per repository and per project, repositories first, and one
        scope per key: naming the same repository twice watches it once, rather
        than fetching it twice and comparing the second fetch against the same
        recorded snapshots as the first.

    Raises:
        CommandError: If nothing was named to watch, or if projects were named
            alongside more than one repository, which leaves no single scope for
            them to be resolved against.

    """
    from gitea.cli.utils.errors import CommandError  # noqa: PLC0415

    if not repositories and not project_ids:
        raise CommandError(
            f"'{command_name}' needs something to watch: pass --repository REPOSITORY for a repository's open "
            f"issues, --project-id ID for a project's board, or both. Either may be repeated to watch several."
        )

    if project_ids and len(repositories) > 1:
        raise CommandError(
            f"'{command_name}' cannot resolve --project-id against {len(repositories)} repositories: a project "
            f"belongs either to one repository or to the owner itself. Pass --project-id with at most one "
            f"--repository, or watch the repositories in a separate invocation."
        )

    scope_repository = repositories[0] if len(repositories) == 1 else None

    scopes = [Scope(key=f"repo:{owner}/{name}", repository=name, project_id=None) for name in repositories]
    scopes += [
        Scope(
            key=f"project:{owner}/{scope_repository}/{identifier}"
            if scope_repository
            else f"project:{owner}/{identifier}",
            repository=scope_repository,
            project_id=identifier,
        )
        for identifier in project_ids
    ]

    unique: dict[str, Scope] = {}
    for scope in scopes:
        unique.setdefault(scope.key, scope)
    return list(unique.values())

gitea.cli.watch.scopes.collect_snapshots

collect_snapshots(
    client: Any, owner: str, scopes: list[Scope]
) -> tuple[
    dict[str, dict[str, dict[str, Any]]], dict[str, Any]
]

Fetch every scope of a run and reduce each to the snapshots the cache holds.

Parameters:

Name	Type	Description	Default
client	Any	The API client.	required
owner	str	The owner the scopes were named with.	required
scopes	list[Scope]	The scopes the run watches.	required

Returns:

Type	Description
dict[str, dict[str, dict[str, Any]]]	A tuple of the snapshots of each scope, keyed by scope key, and the
dict[str, Any]	metadata of the last response - which is what the envelope reports the
tuple[dict[str, dict[str, dict[str, Any]]], dict[str, Any]]	status code of the run from.

Source code in src/gitea/cli/watch/scopes.py

def collect_snapshots(
    client: Any, owner: str, scopes: list[Scope]
) -> tuple[dict[str, dict[str, dict[str, Any]]], dict[str, Any]]:
    """Fetch every scope of a run and reduce each to the snapshots the cache holds.

    Args:
        client: The API client.
        owner: The owner the scopes were named with.
        scopes: The scopes the run watches.

    Returns:
        A tuple of the snapshots of each scope, keyed by scope key, and the
        metadata of the last response - which is what the envelope reports the
        status code of the run from.

    """
    recorded: dict[str, dict[str, dict[str, Any]]] = {}
    metadata: dict[str, Any] = {}

    for scope in scopes:
        issues, metadata = _scope_issues(client, owner, scope)
        recorded[scope.key] = _snapshots(client, owner, scope, issues)

    return recorded, metadata