"""IIIF Change Discovery API 1.0: consume a repository's activity stream.
A publisher exposes a W3C Activity Streams ``OrderedCollection`` of paged
activities (``Create`` / ``Update`` / ``Delete`` / …), each referencing a IIIF
resource that changed. Harvesters walk the pages to discover changed resources
and re-fetch them. This module models and parses that stream; the client
(:meth:`iiisight.AsyncClient.fetch_changes` / ``iter_activities``) walks it.
Activities are ordered oldest → newest across the collection, so a harvester
looking for *recent* changes walks from ``last`` backwards via ``prev`` (the
default), while a full sync walks from ``first`` forwards via ``next``.
"""
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from .model import Reference
[docs]
@dataclass(frozen=True)
class Activity:
"""A single change activity referencing the resource that changed."""
type: str # Create, Update, Delete, Move, Add, Remove, Refresh
object: Reference | None
end_time: str | None = None # xsd:dateTime, the change timestamp
id: str | None = None
actor: str | None = None
@property
def reference(self) -> Reference | None:
"""The changed resource as a :class:`Reference` (hand to ``client.resolve``)."""
return self.object
[docs]
@dataclass(frozen=True)
class ActivityPage:
"""An ``OrderedCollectionPage`` of activities, with paging links."""
id: str
items: tuple[Activity, ...] = ()
next: str | None = None
prev: str | None = None
part_of: str | None = None
start_index: int | None = None
[docs]
@dataclass(frozen=True)
class ChangeCollection:
"""An ``OrderedCollection`` — the entry point to an activity stream."""
id: str
total_items: int | None = None
first: str | None = None # first page URL
last: str | None = None # last page URL
items: tuple[Activity, ...] = () # inlined activities, when there are no pages
[docs]
def parse_change_collection(document: Mapping[str, Any]) -> ChangeCollection:
"""Parse an ``OrderedCollection`` change-discovery document."""
return ChangeCollection(
id=str(document.get("id") or document.get("@id") or ""),
total_items=document.get("totalItems"),
first=_id_of(document.get("first")),
last=_id_of(document.get("last")),
items=tuple(_activity(item) for item in _as_list(document.get("orderedItems"))),
)
[docs]
def parse_activity_page(document: Mapping[str, Any]) -> ActivityPage:
"""Parse an ``OrderedCollectionPage`` of activities."""
return ActivityPage(
id=str(document.get("id") or document.get("@id") or ""),
items=tuple(_activity(item) for item in _as_list(document.get("orderedItems"))),
next=_id_of(document.get("next")),
prev=_id_of(document.get("prev")),
part_of=_id_of(document.get("partOf")),
start_index=document.get("startIndex"),
)
[docs]
def page_activities(
page: ActivityPage | ChangeCollection, newest_first: bool, since: str | None
) -> tuple[list[Activity], bool]:
"""Return the activities to yield from a page and whether to stop paging.
Ordering within the stream is oldest → newest; ``newest_first`` reverses each
page. ``since`` (an ``endTime`` string) filters to strictly-newer activities —
when walking newest-first, reaching an older activity means paging can stop.
"""
items = list(page.items)
if newest_first:
items.reverse()
out: list[Activity] = []
for activity in items:
if since is not None and activity.end_time is not None and activity.end_time <= since:
if newest_first:
return out, True # everything further back is older; stop
continue
out.append(activity)
return out, False
def _activity(document: Any) -> Activity:
if not isinstance(document, Mapping):
return Activity(type="", object=None)
return Activity(
type=str(document.get("type") or document.get("@type") or ""),
object=_reference(document.get("object")),
end_time=document.get("endTime"),
id=_id_of(document.get("id") or document.get("@id")),
actor=_id_of(document.get("actor")),
)
def _reference(value: Any) -> Reference | None:
if isinstance(value, str):
return Reference(id=value, type="")
if isinstance(value, Mapping):
ref_id = value.get("id") or value.get("@id")
if not ref_id:
return None
return Reference(id=str(ref_id), type=str(value.get("type") or value.get("@type") or ""))
return None
def _id_of(value: Any) -> str | None:
if isinstance(value, str):
return value
if isinstance(value, Mapping):
found = value.get("id") or value.get("@id")
return str(found) if found else None
return None
def _as_list(value: Any) -> list[Any]:
if value is None:
return []
if isinstance(value, list):
return value
return [value]