"""The client facades: :class:`AsyncClient` (flagship) and :class:`Client` (sync).
Both fetch IIIF documents over HTTP and hand the body to
:func:`~iiisight.normalize.normalize`, returning the normalized model with its
diagnostics attached. All network I/O is explicit here — the model never fetches
behind attribute access. ``expand()`` resolves a collection's references, and
``resolve()`` fetches a single reference; both are the only ways to trigger the
follow-on requests a collection or reference implies.
The two facades share every pure helper in this module; only the ``httpx`` call
differs (async vs. sync). Class names mirror ``httpx`` so its users have no
surprise.
"""
import asyncio
from collections.abc import AsyncIterator, Iterator, Mapping
from dataclasses import replace
from typing import TypeVar
import httpx
from .auth import (
AccessTokenService,
ProbeResult,
ProbeService,
TokenResult,
parse_probe_result,
parse_token_result,
)
from .conformance import Check, ConformanceReport, evaluate, plan_checks
from .discovery import (
Activity,
ChangeCollection,
page_activities,
parse_activity_page,
parse_change_collection,
)
from .errors import FetchError, UnexpectedDocumentError
from .image import parse_info
from .model import (
AnnotationPage,
Canvas,
Collection,
Described,
ImageService,
Manifest,
Reference,
Service,
)
from .normalize import normalize
from .search import parse_search_response, search_url
from .transport import ACCEPT, DEFAULT_TIMEOUT, afetch_json, fetch_json
#: Default cap on concurrent fetches when expanding a collection (async only).
DEFAULT_CONCURRENCY = 8
_Doc = TypeVar("_Doc", Manifest, Collection)
def _image_service_of(target: ImageService | Canvas) -> ImageService:
"""Resolve an ImageService from a service or a canvas's primary image."""
if isinstance(target, ImageService):
return target
if isinstance(target, Canvas):
service = target.image_service
if service is None:
raise UnexpectedDocumentError("canvas has no image service to load")
return service
raise UnexpectedDocumentError(f"cannot load info for a {type(target).__name__}")
def _info_url(service: ImageService) -> str:
return f"{service.id.rstrip('/')}/info.json"
def _finalize(document: Manifest | Collection, final_url: str) -> Manifest | Collection:
"""Fill an absent document id from the URL it was fetched from."""
if not document.id:
return replace(document, id=final_url)
return document
def _expect(document: Manifest | Collection, kind: type[_Doc]) -> _Doc:
"""Return ``document`` if it is ``kind``, else raise UnexpectedDocumentError."""
if not isinstance(document, kind):
raise UnexpectedDocumentError(
f"expected a {kind.__name__} but fetched a {type(document).__name__}"
)
return document
[docs]
class AsyncClient:
"""An async IIIF client over a shared ``httpx.AsyncClient`` connection pool.
Use as an async context manager. Pass ``http_client`` to reuse an existing
``httpx.AsyncClient`` (e.g. with custom auth/caching); iiisight then does not
close it.
"""
def __init__(
self,
*,
http_client: httpx.AsyncClient | None = None,
timeout: float = DEFAULT_TIMEOUT,
concurrency: int = DEFAULT_CONCURRENCY,
):
self._owns_client = http_client is None
self._http = http_client or httpx.AsyncClient(timeout=timeout)
self._concurrency = concurrency
async def __aenter__(self) -> "AsyncClient":
return self
async def __aexit__(self, *exc_info: object) -> None:
await self.aclose()
[docs]
async def aclose(self) -> None:
"""Close the underlying httpx client, unless it was supplied externally."""
if self._owns_client:
await self._http.aclose()
[docs]
async def fetch(self, url: str) -> Manifest | Collection:
"""Fetch and normalize a manifest or collection (auto-detected)."""
body, final_url = await afetch_json(self._http, url)
return _finalize(normalize(body), final_url)
[docs]
async def fetch_manifest(self, url: str) -> Manifest:
"""Fetch a URL expected to be a Manifest."""
return _expect(await self.fetch(url), Manifest)
[docs]
async def fetch_collection(self, url: str) -> Collection:
"""Fetch a URL expected to be a Collection."""
return _expect(await self.fetch(url), Collection)
[docs]
async def resolve(self, reference: Reference) -> Manifest | Collection:
"""Fetch the document a reference points at."""
return await self.fetch(reference.id)
[docs]
async def expand(self, collection: Collection) -> list[Manifest | Collection]:
"""Resolve a collection's items one level deep, preserving order.
Embedded items are returned as-is; references are fetched (concurrently,
bounded by the client's ``concurrency``).
"""
semaphore = asyncio.Semaphore(self._concurrency)
async def _one(item: Manifest | Collection | Reference) -> Manifest | Collection:
if isinstance(item, Reference):
async with semaphore:
return await self.fetch(item.id)
return item
return list(await asyncio.gather(*(_one(item) for item in collection.items)))
[docs]
async def load_info(self, target: ImageService | Canvas) -> ImageService:
"""Fetch an image service's ``info.json`` and return the loaded service."""
service = _image_service_of(target)
body, _ = await afetch_json(self._http, _info_url(service))
return parse_info(body)
[docs]
async def fetch_region(
self, service: ImageService, region: str = "full", size: str = "max", **url_kwargs: str
) -> bytes:
"""Fetch the image bytes for a region/size from an image service."""
url = service.url(region=region, size=size, **url_kwargs)
try:
response = await self._http.get(url, follow_redirects=True)
response.raise_for_status()
return response.content
except httpx.HTTPError as exc:
raise FetchError(url, str(exc)) from exc
[docs]
async def search(
self,
target: Described | Service | str,
query: str,
params: Mapping[str, str] | None = None,
) -> AnnotationPage:
"""Run a Content Search against a resource's search service.
Args:
target: A manifest/canvas (its search service is discovered), a
:class:`Service`, or a search base URL.
query: The search query string.
params: Extra query parameters (e.g. ``motivation``, ``date``).
Returns:
An :class:`AnnotationPage` of hit annotations.
"""
body, _ = await afetch_json(self._http, search_url(target, query, params))
return parse_search_response(body)
[docs]
async def probe(self, probe_service: ProbeService, token: str | None = None) -> ProbeResult:
"""Call an auth probe service to check access, optionally with a token."""
headers = {"Accept": ACCEPT}
if token:
headers["Authorization"] = f"Bearer {token}"
try:
response = await self._http.get(
probe_service.id, headers=headers, follow_redirects=True
)
response.raise_for_status()
return parse_probe_result(response.json())
except httpx.HTTPError as exc:
raise FetchError(probe_service.id, str(exc)) from exc
except ValueError as exc:
raise FetchError(probe_service.id, f"probe response was not valid JSON: {exc}") from exc
[docs]
async def access_token(self, token_service: AccessTokenService) -> TokenResult:
"""Non-interactive token exchange against an access token service."""
body, _ = await afetch_json(self._http, token_service.id)
return parse_token_result(body)
async def _resolve_service(self, target: ImageService | Canvas | str) -> ImageService:
if isinstance(target, ImageService) and target.loaded:
return target
if isinstance(target, str):
base = target[:-10] if target.endswith("/info.json") else target.rstrip("/")
return await self.load_info(ImageService(id=base, api_version=3))
return await self.load_info(target)
[docs]
async def check_compliance(self, target: ImageService | Canvas | str) -> ConformanceReport:
"""Probe an image service for Image API conformance.
Loads ``info.json`` (if needed), then issues the requests appropriate to
the claimed compliance level and verifies the returned image dimensions.
Args:
target: an image service base URL, an :class:`ImageService`, or a
:class:`Canvas` (its primary image service is used).
"""
service = await self._resolve_service(target)
checks: list[Check] = []
for spec in plan_checks(service):
try:
response = await self._http.get(spec.url, follow_redirects=True)
checks.append(
evaluate(
spec,
response.status_code,
response.headers.get("content-type", ""),
response.content,
)
)
except httpx.HTTPError as exc:
checks.append(
Check(spec.name, spec.level, spec.url, False, f"request failed: {exc}")
)
return ConformanceReport(
service_id=service.id,
claimed_level=service.compliance_level,
checks=tuple(checks),
)
[docs]
async def fetch_changes(self, url: str) -> ChangeCollection:
"""Fetch a Change Discovery ``OrderedCollection`` (the activity stream)."""
body, _ = await afetch_json(self._http, url)
return parse_change_collection(body)
[docs]
async def iter_activities(
self,
source: ChangeCollection | str,
*,
newest_first: bool = True,
since: str | None = None,
) -> AsyncIterator[Activity]:
"""Walk a change stream, yielding activities across pages.
Args:
source: a change-collection URL or a fetched :class:`ChangeCollection`.
newest_first: walk from ``last`` backwards (recent changes first); set
``False`` for an oldest-first full sync from ``first``.
since: an ``endTime`` string; only strictly-newer activities are
yielded (and newest-first paging stops once it reaches older ones).
"""
collection = (
source if isinstance(source, ChangeCollection) else await self.fetch_changes(source)
)
if not collection.first and not collection.last: # inlined, single page
for activity in page_activities(collection, newest_first, since)[0]:
yield activity
return
page_url = collection.last if newest_first else collection.first
while page_url:
body, _ = await afetch_json(self._http, page_url)
page = parse_activity_page(body)
out, stop = page_activities(page, newest_first, since)
for activity in out:
yield activity
if stop:
return
page_url = page.prev if newest_first else page.next
[docs]
class Client:
"""A synchronous IIIF client — the same surface as :class:`AsyncClient`
without ``await``. Use as a context manager."""
def __init__(
self,
*,
http_client: httpx.Client | None = None,
timeout: float = DEFAULT_TIMEOUT,
):
self._owns_client = http_client is None
self._http = http_client or httpx.Client(timeout=timeout)
def __enter__(self) -> "Client":
return self
def __exit__(self, *exc_info: object) -> None:
self.close()
[docs]
def close(self) -> None:
"""Close the underlying httpx client, unless it was supplied externally."""
if self._owns_client:
self._http.close()
[docs]
def fetch(self, url: str) -> Manifest | Collection:
"""Fetch and normalize a manifest or collection (auto-detected)."""
body, final_url = fetch_json(self._http, url)
return _finalize(normalize(body), final_url)
[docs]
def fetch_manifest(self, url: str) -> Manifest:
"""Fetch a URL expected to be a Manifest."""
return _expect(self.fetch(url), Manifest)
[docs]
def fetch_collection(self, url: str) -> Collection:
"""Fetch a URL expected to be a Collection."""
return _expect(self.fetch(url), Collection)
[docs]
def resolve(self, reference: Reference) -> Manifest | Collection:
"""Fetch the document a reference points at."""
return self.fetch(reference.id)
[docs]
def expand(self, collection: Collection) -> list[Manifest | Collection]:
"""Resolve a collection's items one level deep, preserving order."""
return [
self.fetch(item.id) if isinstance(item, Reference) else item
for item in collection.items
]
[docs]
def load_info(self, target: ImageService | Canvas) -> ImageService:
"""Fetch an image service's ``info.json`` and return the loaded service."""
service = _image_service_of(target)
body, _ = fetch_json(self._http, _info_url(service))
return parse_info(body)
[docs]
def fetch_region(
self, service: ImageService, region: str = "full", size: str = "max", **url_kwargs: str
) -> bytes:
"""Fetch the image bytes for a region/size from an image service."""
url = service.url(region=region, size=size, **url_kwargs)
try:
response = self._http.get(url, follow_redirects=True)
response.raise_for_status()
return response.content
except httpx.HTTPError as exc:
raise FetchError(url, str(exc)) from exc
[docs]
def search(
self,
target: Described | Service | str,
query: str,
params: Mapping[str, str] | None = None,
) -> AnnotationPage:
"""Run a Content Search against a resource's search service."""
body, _ = fetch_json(self._http, search_url(target, query, params))
return parse_search_response(body)
[docs]
def probe(self, probe_service: ProbeService, token: str | None = None) -> ProbeResult:
"""Call an auth probe service to check access, optionally with a token."""
headers = {"Accept": ACCEPT}
if token:
headers["Authorization"] = f"Bearer {token}"
try:
response = self._http.get(probe_service.id, headers=headers, follow_redirects=True)
response.raise_for_status()
return parse_probe_result(response.json())
except httpx.HTTPError as exc:
raise FetchError(probe_service.id, str(exc)) from exc
except ValueError as exc:
raise FetchError(probe_service.id, f"probe response was not valid JSON: {exc}") from exc
[docs]
def access_token(self, token_service: AccessTokenService) -> TokenResult:
"""Non-interactive token exchange against an access token service."""
body, _ = fetch_json(self._http, token_service.id)
return parse_token_result(body)
def _resolve_service(self, target: ImageService | Canvas | str) -> ImageService:
if isinstance(target, ImageService) and target.loaded:
return target
if isinstance(target, str):
base = target[:-10] if target.endswith("/info.json") else target.rstrip("/")
return self.load_info(ImageService(id=base, api_version=3))
return self.load_info(target)
[docs]
def check_compliance(self, target: ImageService | Canvas | str) -> ConformanceReport:
"""Probe an image service for Image API conformance (see AsyncClient)."""
service = self._resolve_service(target)
checks: list[Check] = []
for spec in plan_checks(service):
try:
response = self._http.get(spec.url, follow_redirects=True)
checks.append(
evaluate(
spec,
response.status_code,
response.headers.get("content-type", ""),
response.content,
)
)
except httpx.HTTPError as exc:
checks.append(
Check(spec.name, spec.level, spec.url, False, f"request failed: {exc}")
)
return ConformanceReport(
service_id=service.id,
claimed_level=service.compliance_level,
checks=tuple(checks),
)
[docs]
def fetch_changes(self, url: str) -> ChangeCollection:
"""Fetch a Change Discovery ``OrderedCollection`` (the activity stream)."""
body, _ = fetch_json(self._http, url)
return parse_change_collection(body)
[docs]
def iter_activities(
self,
source: ChangeCollection | str,
*,
newest_first: bool = True,
since: str | None = None,
) -> Iterator[Activity]:
"""Walk a change stream, yielding activities across pages (see AsyncClient)."""
collection = source if isinstance(source, ChangeCollection) else self.fetch_changes(source)
if not collection.first and not collection.last:
yield from page_activities(collection, newest_first, since)[0]
return
page_url = collection.last if newest_first else collection.first
while page_url:
body, _ = fetch_json(self._http, page_url)
page = parse_activity_page(body)
out, stop = page_activities(page, newest_first, since)
yield from out
if stop:
return
page_url = page.prev if newest_first else page.next
# --- Module-level convenience -----------------------------------------------
[docs]
async def fetch(url: str) -> Manifest | Collection:
"""One-shot async fetch of a manifest or collection."""
async with AsyncClient() as client:
return await client.fetch(url)
[docs]
async def fetch_manifest(url: str) -> Manifest:
"""One-shot async fetch of a Manifest."""
async with AsyncClient() as client:
return await client.fetch_manifest(url)
[docs]
async def fetch_collection(url: str) -> Collection:
"""One-shot async fetch of a Collection."""
async with AsyncClient() as client:
return await client.fetch_collection(url)
[docs]
def fetch_sync(url: str) -> Manifest | Collection:
"""One-shot sync fetch of a manifest or collection."""
with Client() as client:
return client.fetch(url)
[docs]
def fetch_manifest_sync(url: str) -> Manifest:
"""One-shot sync fetch of a Manifest."""
with Client() as client:
return client.fetch_manifest(url)
[docs]
def fetch_collection_sync(url: str) -> Collection:
"""One-shot sync fetch of a Collection."""
with Client() as client:
return client.fetch_collection(url)