py/facett_demo/client.py
"""Thin gRPC client over the korp-server control plane (:50051).
ONE seam. Every pane reads live server data through a :class:`KorpClient` — no
mocks, no local scans. Each method wraps one KorpControl RPC and returns plain
Python dataclasses so the Streamlit panes stay free of protobuf types.
Honest states (LAW #6 — never a blank false-green): every call either returns
data or raises :class:`KorpUnavailable` carrying the server's own error string,
which the panes render as a loud "server unreachable / empty" panel.
**The korp wire is an OPTIONAL extra** (``pip install 'facett-demo[korp]'``, which
brings ``grpcio`` + ``protobuf``). Without it this module still imports, a
:class:`KorpClient` still constructs, and every RPC raises :class:`KorpNotInstalled`
— a :class:`KorpUnavailable` that names the extra — so the app runs on the gallery's
generated/fixture data and says why the korp tabs are not live.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
#: The install line that brings the korp wire.
KORP_EXTRA_HINT = "pip install 'facett-demo[korp]'"
try:
import grpc
from .proto import korp_pb2 as pb
from .proto import korp_pb2_grpc as rpc
#: Why the korp wire is missing (``None`` when it is installed).
KORP_IMPORT_ERROR: str | None = None
except ImportError as _e: # the [korp] extra is not installed
KORP_IMPORT_ERROR = f"{type(_e).__name__}: {_e}"
pb = rpc = None
class grpc: # noqa: N801 — a stand-in so `except grpc.RpcError` clauses still evaluate
class RpcError(Exception):
"""Never raised: there is no grpc to raise it."""
def korp_available() -> bool:
"""Whether the korp wire (``grpcio`` + ``protobuf``, the ``[korp]`` extra) imports."""
return KORP_IMPORT_ERROR is None
DEFAULT_TARGET = os.environ.get("KORP_SERVER", "localhost:50051")
DEFAULT_TIMEOUT = float(os.environ.get("KORP_TIMEOUT", "30"))
# A Sweden map tile at lod 12 is a 6.75 MB MapQuery reply (measured 2026-09-19); grpc's
# default receive cap is 4 MB and the pane then said "could not read the region
# GeoParquet" for what was a CLIENT-side limit. The server sets no cap of its own.
GRPC_CHANNEL_OPTIONS = [
("grpc.max_receive_message_length", 256 * 1024 * 1024),
("grpc.max_send_message_length", 64 * 1024 * 1024),
]
class KorpUnavailable(RuntimeError):
"""Raised when an RPC cannot be served — carries the honest reason."""
class KorpNotInstalled(KorpUnavailable):
"""The korp wire is not installed (the `[korp]` extra) — not a server failure.
A subclass so every existing `except KorpUnavailable` still catches it, and a
pane that wants to say "install the extra" rather than "the server is down" can
tell the two apart.
"""
# ── plain-Python return shapes (no protobuf leaks into the panes) ────────────
@dataclass
class Health:
status: str
version: str
uptime_seconds: int
@dataclass
class MapPoint:
lon: float
lat: float
tags: str
osm_id: int
@dataclass
class MapWay:
coords: list[float] # flat [lon,lat,lon,lat,...]
tags: str
osm_id: int
@dataclass
class MapScene:
asset: str
ways: list[MapWay]
points: list[MapPoint]
row_groups_total: int
row_groups_scanned: int
truncated: bool
status: str
@dataclass
class GraphNode:
id: int
label: str
key: str
@dataclass
class GraphEdge:
src: int
dst: int
rel: str
@dataclass
class GraphScene:
nodes: list[GraphNode]
edges: list[GraphEdge]
status: str
@dataclass
class IcebergTable:
namespace: str
name: str
rows: int
snapshots: int
latest_commit_ms: int
columns: list[tuple[str, str]] # (name, data_type)
@dataclass
class TablePreview:
table: str
columns: list[str]
column_types: list[str]
rows: list[list[str]]
total_rows: int
snapshots: int
@dataclass
class AggTable:
title: str
columns: list[str]
rows: list[list[str]]
@dataclass
class ChartSeries:
label: str
bars: list[tuple[str, float]] # (key, value)
@dataclass
class Population:
warehouse: str
total_rows: int
aggregates: list[AggTable] = field(default_factory=list)
series: list[ChartSeries] = field(default_factory=list)
class KorpClient:
"""Blocking gRPC client. Cheap to construct; safe to cache per-session."""
def __init__(self, target: str = DEFAULT_TARGET, timeout: float = DEFAULT_TIMEOUT):
self.target = target
self.timeout = timeout
# Without the [korp] extra there is no channel; every RPC refuses BY NAME
# (KorpNotInstalled) through `_stub` rather than the constructor, so the
# sidebar can still build a client and show WHY nothing is live.
if korp_available():
self._channel = grpc.insecure_channel(target, options=GRPC_CHANNEL_OPTIONS)
self._korp_stub = rpc.KorpControlStub(self._channel)
else:
self._channel = None
self._korp_stub = None
@property
def _stub(self):
if self._korp_stub is None:
raise KorpNotInstalled(
f"live korp data needs the korp wire, which is not installed ({KORP_IMPORT_ERROR}) — {KORP_EXTRA_HINT}"
)
return self._korp_stub
# -- helpers --------------------------------------------------------------
def _guard(self, exc: grpc.RpcError):
code = exc.code() if hasattr(exc, "code") else None
detail = exc.details() if hasattr(exc, "details") else str(exc)
raise KorpUnavailable(f"{self.target}: {code.name if code else '?'}: {detail}")
# -- Health (sidebar: which build backs the demo) ------------------------
def health(self) -> Health:
try:
h = self._stub.Health(pb.HealthRequest(), timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
return Health(status=h.status, version=h.version, uptime_seconds=h.uptime_seconds)
# -- Map: Assets + MapQuery ----------------------------------------------
def assets(self) -> tuple[list[dict], str]:
try:
a = self._stub.Assets(pb.AssetsRequest(), timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
out = [
{
"name": x.name,
"ways_url": x.ways_url,
"way_count": x.way_count,
"node_count": x.node_count,
"bbox": x.bbox,
}
for x in a.assets
]
return out, a.error
def map_query(
self,
asset: str = "sweden",
bbox: tuple[float, float, float, float] = (11.0, 55.0, 24.0, 69.5),
lod: int = 6,
max_ways: int = 4000,
max_points: int = 4000,
) -> MapScene:
req = pb.MapQueryRequest(
asset=asset,
min_lon=bbox[0],
min_lat=bbox[1],
max_lon=bbox[2],
max_lat=bbox[3],
lod=lod,
max_ways=max_ways,
max_points=max_points,
)
try:
m = self._stub.MapQuery(req, timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
if m.error:
raise KorpUnavailable(f"MapQuery({asset}): {m.error}")
return MapScene(
asset=asset,
ways=[MapWay(coords=list(w.coords), tags=w.tags, osm_id=w.osm_id) for w in m.ways],
points=[MapPoint(lon=p.lon, lat=p.lat, tags=p.tags, osm_id=p.osm_id) for p in m.points],
row_groups_total=m.row_groups_total,
row_groups_scanned=m.row_groups_scanned,
truncated=m.truncated,
status=m.status,
)
# -- Graph ----------------------------------------------------------------
def graph(
self,
# EMPTY on purpose. korp-server decodes exactly ONE result shape — its own
# scene tail (`id(a) AS src, id(b) AS dst, type(r) AS rel, … a.lon AS slon …`,
# knut-graphdb `RETURN_TAIL`). An empty query makes the server run that scene
# read; any other RETURN shape decodes to zero rows and the server then reports
# `korp-graph-9 … holds 0 nodes` — which is what this pane said against a
# FalkorDB holding 1 212 122 nodes (measured on loki 2026-09-19, korp `24e96fe`).
# The old default here was `MATCH (n)-[e]->(m) RETURN n,e,m LIMIT 200`.
query: str = "",
graph: str = "",
) -> GraphScene:
try:
g = self._stub.Graph(pb.GraphRequest(query=query, graph=graph), timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
if g.error:
raise KorpUnavailable(f"Graph: {g.error}")
return GraphScene(
nodes=[GraphNode(id=n.id, label=n.label, key=n.key) for n in g.nodes],
edges=[GraphEdge(src=e.src, dst=e.dst, rel=e.rel) for e in g.edges],
status=g.status,
)
# -- Analysis: Iceberg browser + Population -------------------------------
def iceberg_tables(self) -> list[IcebergTable]:
try:
t = self._stub.IcebergTables(pb.IcebergTablesRequest(), timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
if t.error:
raise KorpUnavailable(f"IcebergTables: {t.error}")
return [
IcebergTable(
namespace=x.namespace,
name=x.name,
rows=x.rows,
snapshots=x.snapshots,
latest_commit_ms=x.latest_commit_ms,
columns=[(c.name, c.data_type) for c in x.columns],
)
for x in t.tables
]
def table_preview(self, table: str, limit: int = 50) -> TablePreview:
try:
p = self._stub.TablePreview(
pb.TablePreviewRequest(table=table, limit=limit), timeout=self.timeout
)
except grpc.RpcError as e:
self._guard(e)
if p.error:
raise KorpUnavailable(f"TablePreview({table}): {p.error}")
return TablePreview(
table=p.table,
columns=list(p.columns),
column_types=list(p.column_types),
rows=[list(r.cells) for r in p.rows],
total_rows=p.total_rows,
snapshots=p.snapshots,
)
def population(self) -> Population:
try:
p = self._stub.Population(pb.PopulationRequest(), timeout=self.timeout)
except grpc.RpcError as e:
self._guard(e)
if p.error:
raise KorpUnavailable(f"Population: {p.error}")
return Population(
warehouse=p.warehouse,
total_rows=p.total_rows,
aggregates=[
AggTable(title=a.title, columns=list(a.columns), rows=[list(r.cells) for r in a.rows])
for a in p.aggregates
],
series=[
ChartSeries(label=s.label, bars=[(b.key, b.value) for b in s.bars]) for s in p.series
],
)