"""Quanor's feed for an API key — the reference client. Keeps a correct, current copy of the feed for one or more selections: reads each selection's window from ``GET /v1/feed``, listens on the feed stream from the cursor it read, and answers every message the stream sends as the protocol says (https://api.quanor.com/api/docs/stream): a catch-up from ``POST /v1/feed/since`` with the selections it subscribed with, a resync, a change of access, a refetch of one item, processing rows. It holds each item at its latest position, a removal's included, so a frame that arrives after a later one of its item changes nothing; a flash row the stream sent ahead stays until an item frame holds it or lists it as superseded, since one built before the row was written arrives after it; and a ``ready`` under another access than its copy was read under (the key's access changed while it was away) reads its windows again. Given filters (``GET /v1/feed``'s, by its names), it reads its windows and catch-ups with them, keeps every item the stream sends and shows those that pass them (``passes``, ``shown``): what a window read now would list. An item that does not pass stays in its copy, since its next flash row may make it pass; only a ``hidden`` or ``withdrawn`` the stream sent removes one, or a refetch the API answers 404 under the access the copy was read under (under another, it reads its windows again). A flash row sent ahead ages only while the copy hears the stream live and the builder is not silent: while either is not so, its item's frame is late, not missing. A processing row no resolution reaches goes after ``PROCESSING_ROW_S``. ``on_change(copy, frame)`` hears each frame that changed what the copy shows, and ``{"type": "copy_replaced", "cause": ...}`` — the client's own, never the stream's — when the windows were read and the copy replaced by them: at the start (``start``), on a ``resync`` (``resync``), and under another access (``access``). Redraw from the copy then. It keeps its socket alive (the reference's "Liveness and back-off"): it pings the stream every ``PING_EVERY_S`` and takes the socket as dead when nothing at all comes within ``PONG_DEADLINE_S`` of a ping, or nothing for ``silent_after`` the stream's ping interval (``ready.ping_interval_ms``), and then reconnects from the cursor held. Every failure to connect or to stay connected — a refused connection, any HTTP answer to the upgrade (503 while the stream restarts or is full, 404 while none answers the path), a close such as 4408 ``idle`` — is retried with exponential backoff and full jitter, which starts again from its first wait only once a connection has stayed up ``STEADY_S``. A key that is refused (4401, 4403) or a message the stream will not read (4400) stops it: reconnecting would be refused again — but for a 4400 whose reason is ``no_subscribe``, which names no message: no subscribe reached the stream within its wait from the socket opening, a moment's trouble. So does a read the API refuses as the key's (401 or 403 in its error shape) or as a request it will not take (413, 422); any other failed read — a load balancer's page among them — is retried, no sooner than its ``Retry-After`` asks, up to the backoff's longest wait. A restart's 1012 suggests a wait, in its reason and in a ``reconnect`` frame just before it (the frame for a WebSocket stack that drops a close's reason); the client waits it. A 4008 (too slow to read) reconnects at once after a connection that stayed up, and waits like any close when it comes sooner. Needs Python 3.11+, ``httpx`` and ``websockets`` (14 or later):: import asyncio from reference_client import FeedListener listener = FeedListener("qnr_...", {"mine": {"companies": [28, 706]}}, filters={"inside_information": True}, on_change=lambda copy, frame: print(frame["type"], frame.get("feed_item_id"))) asyncio.run(listener.run()) This file is the one the documentation shows; the API's tests run it against the API and a stream that refuses, drains and closes it. """ from __future__ import annotations import asyncio import json import random import re import time from dataclasses import dataclass, field from typing import Any, Callable, Optional import httpx from websockets.asyncio.client import connect from websockets.exceptions import ConnectionClosed, InvalidHandshake #: The API's host: every path the client reads begins ``/v1``. API = "https://api.quanor.com" STREAM = "wss://api.quanor.com/v1/stream/feed" #: Closes after which this key reconnecting would be refused again. STOP = {4400: "a message the stream will not read", 4401: "the key is not valid", 4403: "the key may not stream (revoked, rotated, expired, or no can_websocket)"} #: The one 4400 that names no message: no subscribe reached the stream within #: its wait from the socket opening (the protocol's ``limits.subscribe_within_s``). NO_SUBSCRIBE = "no_subscribe" #: The longest wait a 1012 suggests that a client takes, in seconds: the #: protocol's ``limits.reconnect_after_ms_max``. RECONNECT_AFTER_MAX_S = 5.0 _HINT = re.compile(r"reconnect_after_ms=(\d{1,9})") #: Removals a copy remembers, the oldest let go first: a frame comes out of #: order by moments (a position the stream sent above a gap, the gap's after #: it), never by thousands of removals. MAX_REMOVED = 5_000 #: The part kinds whose relevance the relevance floor reads: an announcement's #: flash rows and its story (the protocol's ``filters.relevance_floor``). A #: part of any other kind is not read for one, whatever fields it carries. RELEVANCE_PART_KINDS = ("flash", "story") #: How long a flash row the stream sent ahead is kept while no item frame has #: held it or listed it as superseded, in seconds of the copy hearing the #: stream live (the protocol's ``limits.ahead_kept_s``): its item's next frame #: comes within moments of a live builder, and a row none ever accounts for is #: not kept for good. AHEAD_KEPT_S = 600.0 #: The item kinds version 1 names (the selection's ``item_kinds``); an item or #: part frame of any other kind is ignored, as the protocol says of every kind #: a reader does not know. ITEM_KINDS = frozenset({"announcement", "market_alert", "earnings_call", "pre_earnings_analysis", "overview_audio", "insider_transaction"}) #: How long a processing row is kept with no resolution heard, in seconds: #: the protocol's ``limits.processing_row_s``, after which the row is resolved #: ``expired`` — a resolution that can be lost before it reaches the stream. PROCESSING_ROW_S = 120.0 #: The stream's ping interval where its ``ready`` does not say one, in #: seconds: the protocol's ``limits.ping_every_s``. PING_INTERVAL_S = 30.0 #: The stream writes a message queued for a listener within this, in seconds, #: or closes the socket 4008: the protocol's ``limits.send_deadline_s``. SEND_DEADLINE_S = 10.0 #: How often this client pings the stream, in seconds, and how long after its #: ping it waits for any message — the pong comes after the frames queued #: before it, so any message says the socket is alive — before taking the #: socket as dead: twice the send deadline, as the stream may be writing a #: batch when the ping comes, and a large frame takes its time over a slow #: link. PING_EVERY_S = 15.0 PONG_DEADLINE_S = 2 * SEND_DEADLINE_S #: The error ``code`` values that say the API refused the key itself. KEY_REFUSED = ("unauthenticated", "forbidden") #: How long a connection stays up after its ``ready`` before the backoff #: starts again from its first wait, in seconds: a failure that always #: follows a ready waits longer each time. STEADY_S = 30.0 #: What ``on_change`` is told when the copy was replaced by its windows read #: again: the client's own, never a frame of the stream's. COPY_REPLACED = "copy_replaced" def silent_after(ping_interval_s: float) -> float: """How long a socket may be heard saying nothing — no frame, no ping — before a client takes it as dead, in seconds: three of the stream's ping intervals, or two and the time it takes to write a message queued behind others, whichever is longer.""" return max(3 * ping_interval_s, 2 * ping_interval_s + SEND_DEADLINE_S) def suggested_wait(reason: str, said: Optional[float]) -> float: """The wait a 1012 suggests, in seconds: its reason's — the whole reason, as the protocol writes it — else the ``reconnect`` frame's sent just before it, else none; never more than the protocol's bound.""" hint = _HINT.fullmatch(reason or "") wait = int(hint.group(1)) / 1000 if hint else (said or 0.0) return min(max(wait, 0.0), RECONNECT_AFTER_MAX_S) class Stopped(Exception): """The stream refused this key or a message, or the API a request: fix it before listening again.""" def _detail(answer: httpx.Response) -> Optional[dict[str, Any]]: """The API's own error body, or none for an answer in another shape (a load balancer's page, a proxy's).""" try: body = answer.json() except ValueError: return None return body if isinstance(body, dict) and "detail" in body else None def passes(frame: dict[str, Any], filters: Optional[dict[str, Any]]) -> bool: """Whether an item passes the filters a window was read with, from its own fields ("Filtering live frames" in the reference): an item frame, or the item a copy holds — its frame's fields, and its parts with the flash rows sent ahead of it. The stream sends every frame; a client narrows them itself.""" if not filters: return True reports = filters.get("reports") or "include" kinds = filters.get("announcement_kind") if frame.get("item_kind") != "announcement": # What only an announcement answers: reports only, inside information, its kind. return not (reports == "only" or filters.get("inside_information") or kinds) teaser = frame.get("access") == "teaser" # shows none of what a filter reads head = frame.get("head") or {} gone = set(frame.get("superseded_parts") or ()) parts = [p for p in frame.get("parts") or () if p.get("part_id") not in gone] story = next((p for p in parts if p.get("kind") == "story"), None) floor = filters.get("relevance_floor") or 1 if floor > 1: # The highest its flash rows (every one, or the list's first) and its # story show; a report story is material, whatever its relevance. A # part of a kind this version does not name is ignored, as the # protocol says of every kind a reader does not know. known = [p["relevance"] for p in parts if p.get("kind") in RELEVANCE_PART_KINDS and isinstance(p.get("relevance"), (int, float))] report = story is not None and story.get("report") is not None if teaser or not (report or (known and max(known) >= floor)): return False if kinds and (teaser or head.get("announcement_kind") not in (None, *kinds)): return False treatments = set(filters.get("story_treatment") or ()) if treatments >= {"full", "brief"}: treatments = set() # both is every story: no filter # A report story is shown as its report, never as a Story or a Brief. if treatments and (teaser or (story is not None and story.get("report") is None and story.get("treatment") not in treatments)): return False if reports == "only" and (teaser or head.get("route") != "report"): return False if reports == "exclude" and (teaser or head.get("route") == "report"): return False # a route not known yet passes; a teaser shows none # Inside information: every announcement but a report is known to be; # reports follow `reports` alone (a report's own flag is unreliable). inside = head.get("route") == "report" or head.get("inside_information") is True return not filters.get("inside_information") or (not teaser and inside) @dataclass class FeedCopy: """The client's copy of the feed: items by id, the position at which each item it no longer holds was hidden or withdrawn, processing rows by edition, the cursor to resume from and the access it was read under.""" items: dict[str, dict[str, Any]] = field(default_factory=dict) removed: dict[str, int] = field(default_factory=dict) processing: dict[int, dict[str, Any]] = field(default_factory=dict) cursor: Optional[str] = None access_version: Optional[str] = None builder: str = "live" #: Per item, the parts its item frame held, and the flash rows the stream #: sent ahead (part frames) that no item frame has yet held or listed as #: superseded, with when each came: an item's parts are the two together. snapshots: dict[str, list[dict[str, Any]]] = field(default_factory=dict) ahead: dict[str, dict[str, tuple[dict[str, Any], float]]] = field(default_factory=dict) #: The selections a ``hidden`` named (``m``), for an item it removed. removed_from: dict[str, frozenset[str]] = field(default_factory=dict) #: The clock the rows sent ahead are aged on, while the copy is live. clock: Callable[[], float] = time.monotonic #: The seconds the copy was live until ``_live_since``, and when it last #: became live (none while it is not): a row sent ahead ages by them. _aged: float = field(default=0.0, init=False, repr=False) _live_since: Optional[float] = field(default=None, init=False, repr=False) #: When each processing row came, by the same clock. _processing_at: dict[int, float] = field(default_factory=dict, init=False, repr=False) def __post_init__(self) -> None: self._live_since = self.clock() def age(self) -> float: """The seconds the copy has been live: the clock rows sent ahead age on.""" since = self._live_since return self._aged + (self.clock() - since if since is not None else 0.0) def set_live(self, live: bool) -> None: """Whether the copy hears its stream live with a builder that is not silent. While it does not, a row sent ahead does not age: its item's frame is late, and comes once the stream is heard again.""" self._aged = self.age() self._live_since = self.clock() if live else None def apply(self, frame: dict[str, Any]) -> bool: """Apply a frame, as the reference's "Reading the feed" says; whether it applied. The stream sends a position as soon as it is accounted for, so an item's frames can arrive out of order — one held above a gap, then the gap's: a frame at or below the position the copy holds the item at, or last removed it at, is older, and changes nothing.""" kind = frame.get("type") if kind in ("item", "part") and frame.get("item_kind") not in ITEM_KINDS: return False # an item kind this version does not name if kind == "item": return self._item(frame) if kind == "part": return self._part(frame) if kind in ("hidden", "withdrawn"): return self._remove(frame) if kind == "processing": self.processing[frame["edition_id"]] = frame self._processing_at.setdefault(frame["edition_id"], self.clock()) elif kind == "processing_resolved": self._processing_at.pop(frame["edition_id"], None) # A row not held, or one this copy already let go of: nothing changes. return self.processing.pop(frame["edition_id"], None) is not None else: return False # A frame type this client does not know is ignored, as are fields. return True def _older(self, frame: dict[str, Any]) -> bool: feed_item_id, seq = frame["feed_item_id"], frame["seq"] held = self.items.get(feed_item_id) if held is not None: return held["seq"] is not None and seq <= held["seq"] removed = self.removed.get(feed_item_id) if removed is None or seq != removed: return removed is not None and seq < removed # At the position a ``hidden`` removed it: an item that left one of # this client's selections for another is sent ``hidden`` for the one # and, at that same position, the item for the other. elsewhere = set(frame.get("m") or ()) - self.removed_from.get(feed_item_id, frozenset(frame.get("m") or ())) return not (frame["type"] == "item" and elsewhere) def _item(self, frame: dict[str, Any]) -> bool: feed_item_id = frame["feed_item_id"] if self._older(frame): return False superseded = set(frame.get("superseded_parts") or ()) parts = [part for part in frame.get("parts") or () if part["part_id"] not in superseded] holds = {part["part_id"]: part["part_version"] for part in parts} # A flash row the stream sent ahead stays until an item frame holds it # at its version or later, or lists it as superseded: a frame built # before the row was written arrives after it, and says nothing of it. kept = {part_id: row for part_id, row in self.ahead.get(feed_item_id, {}).items() if part_id not in superseded and holds.get(part_id, -1) < row[0]["part_version"]} self._forget(feed_item_id) self.snapshots[feed_item_id] = parts if kept: self.ahead[feed_item_id] = kept self.items[feed_item_id] = {**frame, "parts": self._parts(feed_item_id)} return True def _parts(self, feed_item_id: str) -> list[dict[str, Any]]: """An item's parts: those its item frame held — each in the later copy the stream sent ahead, where it did — then the rows sent ahead that the frame does not hold.""" ahead = {part_id: part for part_id, (part, _) in self.ahead.get(feed_item_id, {}).items()} parts = [ahead.pop(part["part_id"], part) for part in self.snapshots.get(feed_item_id, ())] return parts + list(ahead.values()) def _part(self, frame: dict[str, Any]) -> bool: feed_item_id, part = frame["feed_item_id"], frame["part"] held = self.items.get(feed_item_id) if held is None: if feed_item_id in self.removed: return False # only a later item frame brings back an item hidden or withdrawn # A flash row ahead of its item's first snapshot: the item as the # frame's head draws it, until the snapshot comes. item = {key: value for key, value in frame.items() if key not in ("type", "part", "provisional")} held = self.items[feed_item_id] = {**item, "type": "item", "seq": None, "parts": []} elif part["part_id"] in (held.get("superseded_parts") or ()): return False # a row the item frame held says a later one took the place of old = next((p for p in held["parts"] if p["part_id"] == part["part_id"]), None) if old is not None and part["part_version"] <= old["part_version"]: return False self.ahead.setdefault(feed_item_id, {})[part["part_id"]] = (part, self.age()) held["parts"] = self._parts(feed_item_id) return True def sweep(self) -> list[str]: """Drop the flash rows sent ahead that no item frame has held or listed as superseded within ``AHEAD_KEPT_S`` of the copy being live — a row its item's builds never included — and say the items they went from. An item made from such rows alone goes with its last.""" now, changed = self.age(), [] for feed_item_id, rows in list(self.ahead.items()): kept = {part_id: row for part_id, row in rows.items() if now - row[1] < AHEAD_KEPT_S} if len(kept) == len(rows): continue self.ahead[feed_item_id] = kept if not kept: del self.ahead[feed_item_id] held = self.items[feed_item_id] held["parts"] = self._parts(feed_item_id) if held["seq"] is None and not held["parts"]: del self.items[feed_item_id] changed.append(feed_item_id) return changed def _remove(self, frame: dict[str, Any]) -> bool: feed_item_id = frame["feed_item_id"] if self._older(frame): return False self.items.pop(feed_item_id, None) self._forget(feed_item_id) # remembered as the newest removal self.removed[feed_item_id] = frame["seq"] if frame["type"] == "hidden" and frame.get("m"): self.removed_from[feed_item_id] = frozenset(frame["m"]) while len(self.removed) > MAX_REMOVED: self._forget(next(iter(self.removed))) return True def gone(self, feed_item_id: str) -> Optional[int]: """An item the API no longer shows this key (a refetch answered 404): dropped, and remembered as removed at the position it was held at, so a frame of it no later than that, arriving late, does not bring it back, nor does a flash row; a later item frame does. The position it was held at, none for one held from flash rows alone.""" held = self.items.pop(feed_item_id, None) seq = held["seq"] if held is not None else None removed = self.removed.get(feed_item_id) self._forget(feed_item_id) self.removed[feed_item_id] = max(seq or 0, removed or 0) while len(self.removed) > MAX_REMOVED: self._forget(next(iter(self.removed))) return seq def _forget(self, feed_item_id: str) -> None: """What the copy remembers beside an item's own frame: its removal, the parts its frame held and the rows sent ahead of it.""" self.removed.pop(feed_item_id, None) self.removed_from.pop(feed_item_id, None) self.snapshots.pop(feed_item_id, None) self.ahead.pop(feed_item_id, None) def ready(self, message: dict[str, Any]) -> None: """A ``ready``'s processing rows: they replace the copy's when the stream says it holds every row pending (``processing_complete``); a stream still warming may not, and its rows join the ones kept.""" rows = {row["edition_id"]: row for row in message.get("processing", ())} if message.get("processing_complete"): self.processing = rows else: self.processing.update(rows) now = self.clock() self._processing_at = {edition_id: self._processing_at.get(edition_id, now) for edition_id in self.processing} def clear_processing(self) -> None: """Drop every processing row: what an older access was sent.""" self.processing.clear() self._processing_at.clear() def expire_processing(self) -> list[int]: """Drop the processing rows held ``PROCESSING_ROW_S`` with no resolution heard, and say their editions: the stream resolves such a row ``expired`` by then, and a resolution lost on its way would leave it for good.""" now = self.clock() expired = [edition_id for edition_id, came in self._processing_at.items() if now - came >= PROCESSING_ROW_S] for edition_id in expired: self.processing.pop(edition_id, None) del self._processing_at[edition_id] return expired def replace(self, frames: list[dict[str, Any]], access_version: Optional[str]) -> None: """The copy as windows read under ``access_version`` list it. What the older copy remembered of removals goes with it: a frame's seq is the position its own access's rendering last changed at, so one access's positions say nothing of another's.""" self.items.clear() self.removed.clear() self.removed_from.clear() self.snapshots.clear() self.ahead.clear() self.access_version = access_version for frame in frames: self.apply(frame) class Backoff: """Exponential backoff with full jitter: a random wait up to a ceiling that doubles on each failure, from ``first`` to ``most`` seconds.""" def __init__(self, first: float = 1.0, most: float = 60.0) -> None: self.first, self.most, self.ceiling = first, most, first def reset(self) -> None: self.ceiling = self.first async def wait(self, at_least: float = 0.0) -> None: delay = max(at_least, random.uniform(0, self.ceiling)) self.ceiling = min(self.most, self.ceiling * 2) await asyncio.sleep(delay) class FeedListener: def __init__(self, key: str, selections: dict[str, dict[str, Any]], lang: str = "en", *, filters: Optional[dict[str, Any]] = None, api: str = API, stream: str = STREAM, backoff: Optional[Backoff] = None, on_change: Callable[[FeedCopy, dict[str, Any]], None] = lambda copy, frame: None) -> None: if not selections: raise ValueError("a listener needs at least one selection: with none, it is sent no item") self.selections, self.lang, self.filters = selections, lang, filters or {} self.api = httpx.AsyncClient(base_url=api, headers={"Authorization": f"Bearer {key}"}, timeout=30.0) self.stream, self.headers = stream, {"Authorization": f"Bearer {key}"} self.backoff = backoff or Backoff() self.copy, self.on_change = FeedCopy(), on_change #: The wait the connection's last ``reconnect`` frame suggested, in seconds. self.reconnect_after: Optional[float] = None #: The stream's ping interval, as its last ``ready`` said it, in seconds. self.ping_interval = PING_INTERVAL_S #: When the socket was last heard, or the client last finished with a #: message; when the connection's ``ready`` came, until it is steady; #: whether the client is handling a message (reading over HTTP), when #: the socket's silence is the client's, not the stream's. self._heard = 0.0 self._ready_at: Optional[float] = None self._handling = False #: Whether this connection has had its ``ready``, and whether it #: stayed up ``STEADY_S`` after it. self._answered = self._steady = False async def run(self) -> None: """Listen until the stream refuses the key (``Stopped``).""" try: while True: try: if self.copy.cursor is None: await self.read_windows(cause="start") await self._listen() # The socket closed as a normal end (1000, 1001, or no # status): none of the stream's own closes, so something # between it and the client — and, as any close the # reference does not name, reconnected after a wait. # Reconnected at once, a proxy that accepts the upgrade # and ends it would be asked again in a tight loop. await self.backoff.wait(suggested_wait("", self.reconnect_after)) except ConnectionClosed as closed: code = closed.rcvd.code if closed.rcvd else 1006 reason = closed.rcvd.reason if closed.rcvd else "" if code in STOP and (code, reason) != (4400, NO_SUBSCRIBE): raise Stopped(f"{code}: {STOP[code]} ({reason})") from None if code == 4008 and self._steady: continue # too slow to read: reconnect at once, and read faster # A 4008 soon after a connection's ready again and again is # a client too slow for its stream: it waits, as on any close. await self.backoff.wait(suggested_wait(reason, self.reconnect_after)) except _Transient as failed: # A read the API could not answer now: no sooner than its # Retry-After asks, within the backoff's longest wait. await self.backoff.wait(min(failed.retry_after, self.backoff.most)) except (InvalidHandshake, OSError, TimeoutError, httpx.TransportError): # An HTTP answer to the upgrade (503 while the stream # restarts or is full), or no answer at all. await self.backoff.wait() finally: await self.api.aclose() async def _listen(self) -> None: self.reconnect_after, self.ping_interval, self._ready_at = None, PING_INTERVAL_S, None self._answered = self._steady = False async with connect(self.stream, additional_headers=self.headers, open_timeout=10) as socket: loop = asyncio.get_running_loop() self._heard, self._handling = loop.time(), False watch = asyncio.create_task(self._watch(socket)) try: await self._subscribe(socket, self.copy.cursor) async for text in socket: self._heard, self._handling = loop.time(), True await self._handle(socket, json.loads(text)) self._heard, self._handling = loop.time(), False if self._ready_at is not None and self._heard - self._ready_at >= STEADY_S: self.backoff.reset() # a connection that stayed up self._ready_at, self._steady = None, True finally: watch.cancel() self.copy.set_live(False) async def _handle(self, socket: Any, message: dict[str, Any]) -> None: kind = message.get("type") if kind == "ready": await self._ready(message) elif kind == "ping": await socket.send(json.dumps({"type": "pong", "t": message["t"]})) self._sweep() elif kind == "catch_up": await self._subscribe(socket, await self.catch_up(self.copy.cursor)) elif kind == "resync": await self._subscribe(socket, await self.read_windows(cause="resync")) elif kind == "access_changed": await self._supersede() elif kind == "refetch": await self.read_item(message["feed_item_id"]) elif kind == "status": self._builder(message["builder"]) elif kind == "reconnect": # The stream closes 1012 next: the wait it suggests. self.reconnect_after = message["reconnect_after_ms"] / 1000 elif kind != "selection_ack": self._apply(message) if message.get("cursor") and kind != "selection_ack": # A selection_ack's cursor is a bound to read the mirror to — the # highest position the stream had heard — and may stand above a # position not sent yet: never one to resume from. self.copy.cursor = message["cursor"] async def _ready(self, message: dict[str, Any]) -> None: self._ready_at = asyncio.get_running_loop().time() interval = message.get("ping_interval_ms") if isinstance(interval, (int, float)) and not isinstance(interval, bool) and interval > 0: self.ping_interval = interval / 1000 if message["access_version"] != self.copy.access_version: # The key's access changed while this client was away, or after # its windows were read: an item that did not change since is not # resent, so what was read under the older access is superseded # here or never. await self._supersede() self.copy.ready(message) self._answered = True self._builder("live") # every ready says the builder is live; a status follows if not def _builder(self, state: str) -> None: """The builder's state, as ``ready`` or ``status`` says it: rows sent ahead age only while it is live (a state this version does not name is not) and the connection has had its ``ready``.""" self.copy.builder = state self.copy.set_live(state == "live" and self._answered) async def _watch(self, socket: Any) -> None: """Close a socket that stopped delivering, so the client reconnects from its cursor: nothing at all within ``PONG_DEADLINE_S`` of this client's ping, or nothing for ``silent_after`` the stream's ping interval. The client's own reads over HTTP are its silence, not the stream's: they hold no deadline, and the pings go on through them, so the stream does not take the client for idle.""" loop = asyncio.get_running_loop() next_ping, waiting_since = loop.time() + PING_EVERY_S, None try: while True: now = loop.time() if waiting_since is not None and self._heard >= waiting_since: waiting_since = None # something came after the ping late = waiting_since is not None and now - waiting_since >= PONG_DEADLINE_S if not self._handling and (late or now - self._heard >= silent_after(self.ping_interval)): await socket.close(1000, "no message in time") return if now >= next_ping: await socket.send(json.dumps({"type": "ping", "t": round(time.time() * 1000)})) next_ping = now + PING_EVERY_S if waiting_since is None: waiting_since = now due = [next_ping] if not self._handling: # a deadline passed while handling waits for its end due.append(self._heard + silent_after(self.ping_interval)) if waiting_since is not None: due.append(waiting_since + PONG_DEADLINE_S) # Waking at least every second, it reads a ping interval a # ``ready`` changed meanwhile. await asyncio.sleep(min(max(min(due) - loop.time(), 0.0) + 0.001, 1.0)) except ConnectionClosed: return except Exception: # A check that failed leaves no socket unwatched: the client # reconnects, as from any socket closed under it. await socket.close(1011, "liveness check failed") async def _supersede(self) -> None: """What was read under an older access is superseded: its processing rows go and the windows are read again; the stream goes on from its own cursor.""" self.copy.clear_processing() await self.read_windows(resume=False, cause="access") async def _subscribe(self, socket: Any, since: Optional[str]) -> None: await socket.send(json.dumps({"type": "subscribe", "v": 1, "since": since, "lang": self.lang, "selections": self.selections})) async def _get(self, path: str, **params: Any) -> dict[str, Any]: return self._answer(await self.api.get(path, params=params)) async def _post(self, path: str, body: dict[str, Any], **params: Any) -> dict[str, Any]: return self._answer(await self.api.post(path, json=body, params=params)) @staticmethod def _answer(answer: httpx.Response) -> dict[str, Any]: """A read's answer; ``Stopped`` where asking again would be refused again — the key refused by the API itself (401 or 403 in its error shape, its ``code``, where it has one, naming the key), a request it will not take (413, 422) — and ``_Transient`` for any other failure, a load balancer's answer among them.""" status, error = answer.status_code, _detail(answer) refused = status in (401, 403) and error is not None and error.get("code") in (None, *KEY_REFUSED) if refused or (status in (413, 422) and error is not None): raise Stopped(f"{status}: {error['detail']}") if status >= 400: raise _Transient(status, _retry_after(answer)) return answer.json() async def read_windows(self, *, resume: bool = True, cause: str = "start") -> Optional[str]: """Each selection's newest page, replacing the copy, which is under the access they were read under — none when the access changed between them, so the stream's next ``ready`` or ``access_changed`` reads them again — and telling ``on_change`` so, with why (``cause``); with ``resume``, the cursor to subscribe from: the lowest any of them was read at.""" frames, throughs, accesses = [], [], set() for selection in self.selections.values(): page = await self._get("/v1/feed", selection=json.dumps(selection), lang=self.lang, **self.filters) frames += page["frames"] throughs.append(page["through"]) accesses.add(page["access_version"]) self.copy.replace(frames, accesses.pop() if len(accesses) == 1 else None) if resume: self.copy.cursor = min(throughs, key=lambda c: tuple(int(x) for x in c.split("."))) self.on_change(self.copy, {"type": COPY_REPLACED, "cause": cause}) return self.copy.cursor async def catch_up(self, since: str) -> str: """Every change past ``since`` for the selections subscribed with, page by page until complete; the cursor to subscribe from.""" while True: # The selections in the body, as the subscribe sent them (eight of # 2,000 companies each are more than a URL carries), and the # windows' filters beside them, by the same names. found = await self._post("/v1/feed/since", {"since": since, "selections": self.selections, **self.filters}, lang=self.lang) if found["resync"]: return await self.read_windows(cause="resync") for frame in found["frames"]: self._apply(frame) since = found["through"] if found["complete"]: self.copy.cursor = since return since async def read_item(self, feed_item_id: str) -> None: """Answer a ``refetch``: the item as the key is shown it now, applied by its ``seq``; the API's 404 under the access the copy was read under says the key is shown it no longer, a removal ``on_change`` hears as ``hidden`` where the item was shown, and under another access that the key's access changed: the windows are read again. Another 404 — a load balancer's page — is a read that failed.""" answer = await self.api.get(f"/v1/feed/items/{feed_item_id}", params={"lang": self.lang}) error = _detail(answer) if answer.status_code == 404 and error is not None: # the API's, not a load balancer's if error.get("access_version") not in (None, self.copy.access_version): await self._supersede() return was_shown = self._shows(feed_item_id) seq = self.copy.gone(feed_item_id) if was_shown: self.on_change(self.copy, {"type": "hidden", "feed_item_id": feed_item_id, "seq": seq}) return self._apply(self._answer(answer)["frame"]) def _sweep(self) -> None: """Drop the rows sent ahead past their time (``FeedCopy.sweep``), telling ``on_change`` each item shown that they changed: as it is now held, or ``hidden`` where it is no longer shown; and the processing rows no resolution reached (``FeedCopy.expire_processing``), each told as resolved ``expired``.""" for edition_id in self.copy.expire_processing(): self.on_change(self.copy, {"type": "processing_resolved", "edition_id": edition_id, "resolution": "expired"}) shown = {feed_item_id for feed_item_id in self.copy.ahead if self._shows(feed_item_id)} for feed_item_id in self.copy.sweep(): if self._shows(feed_item_id): self.on_change(self.copy, self.copy.items[feed_item_id]) elif feed_item_id in shown: held = self.copy.items.get(feed_item_id) self.on_change(self.copy, {"type": "hidden", "feed_item_id": feed_item_id, "seq": held["seq"] if held is not None else None}) def shown(self) -> dict[str, dict[str, Any]]: """The items of the copy that pass the filters: what a window read now would list.""" return {feed_item_id: item for feed_item_id, item in self.copy.items.items() if passes(item, self.filters)} def _shows(self, feed_item_id: Optional[str]) -> bool: item = self.copy.items.get(feed_item_id) return item is not None and passes(item, self.filters) def _apply(self, frame: dict[str, Any]) -> None: """Apply a frame to the copy, which holds every item the stream sends whatever the filters make of it, and tell ``on_change`` each frame that applied — an item's or a flash row's as it came when the item passes the filters now, as ``hidden`` at the item's position (none, for one held from flash rows alone) when the item was shown and passes no longer, and not at all when it neither was shown nor is. An item the filters leave out is not removed: the copy keeps it with the flash rows sent ahead of it, so its next row can make it pass, where one the stream removed (``hidden``, ``withdrawn``) comes back with a later item frame alone.""" feed_item_id = frame.get("feed_item_id") was_shown = self._shows(feed_item_id) if not self.copy.apply(frame): return if frame["type"] in ("item", "part") and not self._shows(feed_item_id): if not was_shown: return frame = {"type": "hidden", "feed_item_id": feed_item_id, "seq": self.copy.items[feed_item_id]["seq"]} self.on_change(self.copy, frame) def _retry_after(answer: httpx.Response) -> float: """The wait an answer's ``Retry-After`` asks for, in seconds; none for a form other than the whole seconds the API writes.""" said = answer.headers.get("Retry-After", "").strip() return float(said) if said.isascii() and said.isdigit() else 0.0 class _Transient(Exception): """A read the API could not answer now: back off and reconnect, no sooner than its ``Retry-After`` asks, within the backoff's longest wait.""" def __init__(self, status: int, retry_after: float = 0.0) -> None: super().__init__(status) self.retry_after = retry_after