gitea.watch.changes

changes

The snapshot a watch run takes of an issue, and the comparison of two of them.

An issue payload is far larger than the handful of things worth waking someone up for, and most of it changes for reasons nobody wants a line of output about. A snapshot keeps the four that a watch reports on - who it is assigned to, what it is labelled, which comments it carries, and whether it is there at all - plus the fields needed to name the issue in the report.

Comments are compared by hash rather than by count, so that a comment added and another deleted between two runs is two changes and not none, and so that the comparison needs nothing from the previous run except the hashes it recorded. comment_hash is stable across re-fetches: the same comment always hashes to the same value, and an edited one does not, which is what makes an edit show up as a change rather than disappear.

A comment is reported whether or not the issue carrying it was already known. An issue seen for the first time is compared against an empty snapshot rather than passed over, so a comment that was already on it is an addition and not a baseline - what it costs to pass over is a comment nobody is ever told about, since the run that passed over it also recorded it.

updated_at is recorded but is not itself compared. Gitea bumps it for every edit, including ones a watch has nothing to say about, so comparing it would report a body reworded as indistinguishable from a comment added. The consequence, stated plainly: an issue whose title or body alone was edited is not reported as changed.

Functions:

gitea.watch.changes.usable_identifier

usable_identifier(value: Any) -> int | None

Read a value that has to be a whole number to identify anything.

Every identifier a watch reads - an issue's global ID, its number, a column's ID - is read through here, so that a payload with a nonsense one in it is refused the same way whichever field it was in.

Parameters:

Name	Type	Description	Default
value	Any	The value the payload carries for the identifier.	required

Returns:

Type	Description
int | None	The identifier, or None when the value is not one. True is not one:
int | None	it is an int as far as Python is concerned, and would key every issue
int | None	whose ID came back as a boolean under the same entry.

Source code in src/gitea/watch/changes.py

def usable_identifier(value: Any) -> int | None:
    """Read a value that has to be a whole number to identify anything.

    Every identifier a watch reads - an issue's global ID, its number, a
    column's ID - is read through here, so that a payload with a nonsense one in
    it is refused the same way whichever field it was in.

    Args:
        value: The value the payload carries for the identifier.

    Returns:
        The identifier, or None when the value is not one. `True` is not one:
        it is an `int` as far as Python is concerned, and would key every issue
        whose ID came back as a boolean under the same entry.

    """
    return value if isinstance(value, int) and not isinstance(value, bool) else None

gitea.watch.changes.comment_hash

comment_hash(comment: dict[str, Any]) -> str

Hash a comment's identity, stably across re-fetches.

The digest is taken over the comment's ID, author, body and timestamps, serialized as JSON so that no value can be confused with the boundary between two of them - a body ending in the separator would otherwise hash as a different comment's body beginning with it.

Including the timestamps means an edited comment hashes differently from the comment it replaced, so an edit is reported as a change rather than passing for the comment already recorded.

The author is taken by ID and never by login, because a login is renameable and the digest has to survive a rename: hashing the login would turn every comment a renamed user ever wrote into a removal and an addition, on every issue being watched, although nothing about any of them changed. A comment's author cannot change, so the ID is only there to tell two comments apart when the payload carries no comment ID of its own - and a payload carrying no user ID contributes no author at all rather than falling back to the login, which would put the rename back.

Parameters:

Name	Type	Description	Default
comment	dict[str, Any]	The comment data returned by the API. A payload missing any of these fields hashes as if it carried them empty, rather than raising.	required

Returns:

Type	Description
str	The first _HASH_LENGTH hex characters of the SHA-256 digest.

Source code in src/gitea/watch/changes.py

def comment_hash(comment: dict[str, Any]) -> str:
    """Hash a comment's identity, stably across re-fetches.

    The digest is taken over the comment's ID, author, body and timestamps,
    serialized as JSON so that no value can be confused with the boundary
    between two of them - a body ending in the separator would otherwise hash as
    a different comment's body beginning with it.

    Including the timestamps means an edited comment hashes differently from the
    comment it replaced, so an edit is reported as a change rather than passing
    for the comment already recorded.

    The author is taken by ID and never by login, because a login is renameable
    and the digest has to survive a rename: hashing the login would turn every
    comment a renamed user ever wrote into a removal and an addition, on every
    issue being watched, although nothing about any of them changed. A comment's
    author cannot change, so the ID is only there to tell two comments apart
    when the payload carries no comment ID of its own - and a payload carrying
    no user ID contributes no author at all rather than falling back to the
    login, which would put the rename back.

    Args:
        comment: The comment data returned by the API. A payload missing any of
            these fields hashes as if it carried them empty, rather than
            raising.

    Returns:
        The first `_HASH_LENGTH` hex characters of the SHA-256 digest.

    """
    user = comment.get("user")
    author = usable_identifier(user.get("id")) if isinstance(user, dict) else None
    identity = [
        comment.get("id"),
        author,
        comment.get("body") if isinstance(comment.get("body"), str) else "",
        comment.get("created_at") if isinstance(comment.get("created_at"), str) else "",
        comment.get("updated_at") if isinstance(comment.get("updated_at"), str) else "",
    ]
    raw = json.dumps(identity, separators=(",", ":"), sort_keys=True, default=str)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:_HASH_LENGTH]

gitea.watch.changes.issue_key

issue_key(issue: dict[str, Any]) -> str | None

Build the key an issue is recorded under.

The global ID is used rather than the number shown in the web UI, because a project scope holds issues from several repositories and their numbers collide.

Parameters:

Name	Type	Description	Default
issue	dict[str, Any]	The issue data returned by the API.	required

Returns:

Type	Description
str | None	The key, or None when the payload carries no usable global ID.

Source code in src/gitea/watch/changes.py

def issue_key(issue: dict[str, Any]) -> str | None:
    """Build the key an issue is recorded under.

    The global ID is used rather than the number shown in the web UI, because a
    project scope holds issues from several repositories and their numbers
    collide.

    Args:
        issue: The issue data returned by the API.

    Returns:
        The key, or None when the payload carries no usable global ID.

    """
    identifier = usable_identifier(issue.get("id"))
    return None if identifier is None else str(identifier)

gitea.watch.changes.issue_snapshot

issue_snapshot(
    issue: dict[str, Any],
    comments: list[dict[str, Any]],
    repository: str | None = None,
) -> dict[str, Any]

Reduce an issue and its comments to what a watch compares and reports.

Parameters:

Name	Type	Description	Default
issue	dict[str, Any]	The issue data returned by the API.	required
comments	list[dict[str, Any]]	The issue's comments, as returned by the API. Pass an empty list for an issue whose comments were not fetched; its comment hashes are then empty and no comment change is ever reported for it.	required
repository	str | None	Full name of the repository holding the issue, used to name it in the report, or None when it could not be determined.	None

Returns:

Type	Description
dict[str, Any]	The snapshot of the issue.

Source code in src/gitea/watch/changes.py

def issue_snapshot(
    issue: dict[str, Any], comments: list[dict[str, Any]], repository: str | None = None
) -> dict[str, Any]:
    """Reduce an issue and its comments to what a watch compares and reports.

    Args:
        issue: The issue data returned by the API.
        comments: The issue's comments, as returned by the API. Pass an empty
            list for an issue whose comments were not fetched; its comment
            hashes are then empty and no comment change is ever reported for it.
        repository: Full name of the repository holding the issue, used to name
            it in the report, or None when it could not be determined.

    Returns:
        The snapshot of the issue.

    """
    title = issue.get("title")
    updated_at = issue.get("updated_at")

    return {
        "issue_id": usable_identifier(issue.get("id")),
        "number": usable_identifier(issue.get("number")),
        "title": title if isinstance(title, str) else "",
        "repository": repository,
        "updated_at": updated_at if isinstance(updated_at, str) else "",
        "assignees": _names(issue.get("assignees"), "login"),
        "labels": _names(issue.get("labels"), "name"),
        "comment_hashes": sorted({comment_hash(comment) for comment in comments if isinstance(comment, dict)}),
    }

gitea.watch.changes.detect_changes

detect_changes(
    current: dict[str, dict[str, Any]],
    previous: dict[str, dict[str, Any]] | None,
) -> list[dict[str, Any]]

Compare the snapshots of one scope against the ones recorded for it.

An issue that changed in more than one way contributes one record per way, as a digest reads better naming each change than one line naming three.

An issue the recorded scope has not seen is one of those: it contributes the new record naming it, and then the same records everything else contributes, taken against the empty snapshot it is being compared to - so the assignees, labels and comments it already carries are reported as added. Reporting only new would make them baseline, which loses a comment written between an issue being opened and the run that first saw it, and loses it permanently: the run that swallowed it records it and the next one has nothing left to compare against. It matters most to a consumer that reacts to a kind rather than to an issue - one acting on comments and not on new would never learn the comment was there.

Parameters:

Name	Type	Description	Default
current	dict[str, dict[str, Any]]	The snapshot of each issue in the scope now, keyed by issue_key.	required
previous	dict[str, dict[str, Any]] | None	The snapshots the last run recorded for the scope, or None when the scope has never been recorded. None baselines the scope: no change is reported, whatever is in it.	required

Returns:

Type	Description
list[dict[str, Any]]	The changes since the recorded snapshots, ordered by issue and then by
list[dict[str, Any]]	what changed. Empty when nothing changed, which is the whole point of
list[dict[str, Any]]	the command.

Source code in src/gitea/watch/changes.py

def detect_changes(
    current: dict[str, dict[str, Any]],
    previous: dict[str, dict[str, Any]] | None,
) -> list[dict[str, Any]]:
    """Compare the snapshots of one scope against the ones recorded for it.

    An issue that changed in more than one way contributes one record per way,
    as a digest reads better naming each change than one line naming three.

    An issue the recorded scope has not seen is one of those: it contributes the
    `new` record naming it, and then the same records everything else
    contributes, taken against the empty snapshot it is being compared to - so
    the assignees, labels and comments it already carries are reported as added.
    Reporting only `new` would make them baseline, which loses a comment written
    between an issue being opened and the run that first saw it, and loses it
    permanently: the run that swallowed it records it and the next one has
    nothing left to compare against. It matters most to a consumer that reacts
    to a kind rather than to an issue - one acting on `comments` and not on
    `new` would never learn the comment was there.

    Args:
        current: The snapshot of each issue in the scope now, keyed by
            `issue_key`.
        previous: The snapshots the last run recorded for the scope, or None
            when the scope has never been recorded. None baselines the scope:
            no change is reported, whatever is in it.

    Returns:
        The changes since the recorded snapshots, ordered by issue and then by
        what changed. Empty when nothing changed, which is the whole point of
        the command.

    """
    if previous is None:
        return []

    changes: list[dict[str, Any]] = []

    for key, snapshot in sorted(current.items(), key=_sort_key):
        before = previous.get(key)
        if before is None:
            changes.append(_change(snapshot, "new", "new issue", [], []))
            # An issue first seen here is compared against an empty snapshot
            # rather than skipped, so what it already carries is reported as
            # added. Skipping it would make the comments an issue was opened
            # with - or the ones written before the run that first saw it -
            # baseline instead of a change, and they would never be reported:
            # the one record that mentioned the issue would be `new`, which a
            # consumer filtering on the kinds it acts on may not act on at all.
            before = {}

        for field, kind in _FIELD_KINDS:
            added, removed = _delta(before.get(field, []), snapshot.get(field, []))
            if added or removed:
                changes.append(_change(snapshot, kind, _describe_names(added, removed), added, removed))

        added, removed = _delta(before.get("comment_hashes", []), snapshot.get("comment_hashes", []))
        if added or removed:
            changes.append(_change(snapshot, "comments", _describe_comments(added, removed), added, removed))

    gone = {key: snapshot for key, snapshot in previous.items() if key not in current}
    for _, snapshot in sorted(gone.items(), key=_sort_key):
        changes.append(_change(snapshot, "gone", "no longer listed", [], []))

    return changes

gitea.watch.changes.format_change

format_change(change: dict[str, Any]) -> str

Render one change as the line the human digest prints for it.

The issue is named the way it is written in a browser's address bar and in prose - owner/repo#15 - so a line can be read, pasted and grepped without consulting the scope it came from.

Parameters:

Name	Type	Description	Default
change	dict[str, Any]	The change record.	required

Returns:

Type	Description
str	One line naming the issue, what changed and what it changed to.

Source code in src/gitea/watch/changes.py

def format_change(change: dict[str, Any]) -> str:
    """Render one change as the line the human digest prints for it.

    The issue is named the way it is written in a browser's address bar and in
    prose - `owner/repo#15` - so a line can be read, pasted and grepped without
    consulting the scope it came from.

    Args:
        change: The change record.

    Returns:
        One line naming the issue, what changed and what it changed to.

    """
    number, repository = change.get("number"), change.get("repository")
    if isinstance(number, int):
        reference = f"{repository}#{number}" if repository else f"#{number}"
    else:
        reference = f"issue {change.get('issue_id')}"

    line = f"{reference} {change.get('kind')}: {change.get('detail')}"
    title = change.get("title")
    return f"{line} · {title}" if title else line