From 54e6a3b17c0eaf0036e903e5fd68923326201460 Mon Sep 17 00:00:00 2001 From: Hein Date: Wed, 30 Sep 2026 22:18:01 +0200 Subject: [PATCH] feat(resolvespec-python): add Python client for ResolveSpec, HeaderSpec, FunctionSpec and WebSocketSpec --- resolvespec-python/.gitignore | 6 + resolvespec-python/README.md | 142 ++++++++ resolvespec-python/pyproject.toml | 24 ++ .../src/resolvespec/__init__.py | 43 +++ .../src/resolvespec/funcspec.py | 197 ++++++++++ .../src/resolvespec/headerspec.py | 336 ++++++++++++++++++ resolvespec-python/src/resolvespec/http.py | 76 ++++ .../src/resolvespec/resolvespec.py | 137 +++++++ resolvespec-python/src/resolvespec/types.py | 166 +++++++++ .../src/resolvespec/websocket.py | 335 +++++++++++++++++ resolvespec-python/tests/test_funcspec.py | 132 +++++++ resolvespec-python/tests/test_headerspec.py | 236 ++++++++++++ resolvespec-python/tests/test_resolvespec.py | 210 +++++++++++ resolvespec-python/tests/test_websocket.py | 152 ++++++++ resolvespec-python/todo.md | 28 +- 15 files changed, 2206 insertions(+), 14 deletions(-) create mode 100644 resolvespec-python/.gitignore create mode 100644 resolvespec-python/README.md create mode 100644 resolvespec-python/pyproject.toml create mode 100644 resolvespec-python/src/resolvespec/__init__.py create mode 100644 resolvespec-python/src/resolvespec/funcspec.py create mode 100644 resolvespec-python/src/resolvespec/headerspec.py create mode 100644 resolvespec-python/src/resolvespec/http.py create mode 100644 resolvespec-python/src/resolvespec/resolvespec.py create mode 100644 resolvespec-python/src/resolvespec/types.py create mode 100644 resolvespec-python/src/resolvespec/websocket.py create mode 100644 resolvespec-python/tests/test_funcspec.py create mode 100644 resolvespec-python/tests/test_headerspec.py create mode 100644 resolvespec-python/tests/test_resolvespec.py create mode 100644 resolvespec-python/tests/test_websocket.py diff --git a/resolvespec-python/.gitignore b/resolvespec-python/.gitignore new file mode 100644 index 0000000..27e6129 --- /dev/null +++ b/resolvespec-python/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.egg-info/ +.venv/ +dist/ +.pytest_cache/ +.coverage diff --git a/resolvespec-python/README.md b/resolvespec-python/README.md new file mode 100644 index 0000000..ff60cd2 --- /dev/null +++ b/resolvespec-python/README.md @@ -0,0 +1,142 @@ +# resolvespec (Python) + +Python client for ResolveSpec REST, HeaderSpec (restheadspec), FunctionSpec and WebSocketSpec. Port of `resolvespec-js`. + +- Python >= 3.11, `httpx` (REST, sync + async), `websockets` (WS, async) +- Options/filters/sorts are plain dicts using the wire key names (`TypedDict` hints in `resolvespec.types`) + +``` +pip install resolvespec +``` + +## Clients + +| Protocol | Sync | Async | Transport | +|---|---|---|---| +| ResolveSpec | `ResolveSpecClient` | `AsyncResolveSpecClient` | POST + JSON body `{operation, id, data, options}` | +| HeaderSpec | `HeaderSpecClient` | `AsyncHeaderSpecClient` | GET/POST/PUT/DELETE, options as `X-*` headers | +| FunctionSpec | `FuncSpecClient` | `AsyncFuncSpecClient` | user-defined SQL endpoints; params via query string + `X-*` headers | +| WebSocketSpec | - | `WebSocketClient` | WebSocket JSON messages | + +Constructor (REST): `Client(base_url, token=None, headers=None, timeout=30.0)` + +- `token` -> `Authorization: Bearer`; wins over `headers` +- `headers`: custom headers, merged case-insensitively; snapshot at construction +- Sync: context manager / `close()`. Async: `async with` / `await aclose()` +- Cached sync factories: `get_resolvespec_client()`, `get_headerspec_client()` (same args -> same instance) + +## ResolveSpec + +URL: `{base}/{schema}/{entity}[/{id}]` + +| Method | Signature | +|---|---| +| `get_metadata` | `(schema, entity)` (GET) | +| `read` | `(schema, entity, id=None, options=None)` | +| `create` | `(schema, entity, data, options=None)` | +| `update` | `(schema, entity, data, id=None, options=None)` | +| `delete` | `(schema, entity, id)` | + +`id`: int/str -> URL path; `list[str]` -> body `id`. +Returns `{"success", "data", "metadata"?, "error"?}`. + +## HeaderSpec + +| Method | HTTP | Signature | +|---|---|---| +| `read` | GET | `(schema, entity, id=None, options=None)` | +| `create` | POST | `(schema, entity, data, options=None)` | +| `update` | PUT | `(schema, entity, id, data, options=None)` | +| `delete` | DELETE | `(schema, entity, id)` | + +Response metadata derived from `Content-Range` (`offset-end/total`) and `X-Limit`. +`build_headers(options)`, `encode_header_value()` / `decode_header_value()` (`ZIP_` / `__` base64) are exported. + +### Option -> header + +| Option | Header | +|---|---| +| `columns` / `omit_columns` | `X-Select-Fields` / `X-Not-Select-Fields` | +| filter `eq` + AND | `X-FieldFilter-{col}` | +| filter AND / OR | `X-SearchOp-{op}-{col}` / `X-SearchOr-{op}-{col}` | +| spatial (`st_*`, `bbox`) / vector (`*_within`) filter | `X-SpatialFilter-{col}` / `X-VectorFilter-{col}` (JSON) | +| `sort` | `X-Sort` (`+col,-col`) | +| `limit` / `offset` | `X-Limit` / `X-Offset` | +| `cursor_forward` / `cursor_backward` | `X-Cursor-Forward` / `X-Cursor-Backward` | +| `preload` | `X-Preload` (`Rel:c1,c2\|Rel2`), `X-Preload-Where`, `X-Preload-{n}[-Where]` | +| `expand` | `X-Expand` | +| `custom_sql_joins` / `custom_sql_or` | `X-Custom-SQL-Join` / `X-Custom-SQL-Or` | +| `search_columns` | `X-SearchCols` | +| `advanced_sql` | `X-AdvSQL-{col}` | +| `computedColumns` | `X-CQL-SEL-{name}` | +| `customOperators` | `X-Custom-SQL-W` (AND-joined) | +| `vector_search` | `X-Vector-Search-{col}`, `-Vector`, `-As`, `-Dir` | +| `fetch_row_number` | `X-Fetch-RowNumber` | +| `clean_json`, `distinct`, `skip_count`, `skip_cache`, `atomic_transaction`, `single_record_as_object` | `X-Clean-JSON`, `X-Distinct`, `X-SkipCount`, `X-SkipCache`, `X-Transaction-Atomic`, `X-Single-Record-As-Object` | +| `pk_row` | `X-PKRow` | +| `response_format` (`simple`/`detail`/`syncfusion`) | `X-SimpleApi` / `X-DetailApi` / `X-Syncfusion` | +| `xfiles` | `X-Files` (`ZIP_` base64 JSON) | + +Filter operator -> header op: `eq equals`, `neq notequals`, `gt greaterthan`, `gte greaterthanorequal`, `lt lessthan`, `lte lessthanorequal`, `like/ilike/contains contains`, `startswith beginswith`, `endswith`, `in`, `between`, `between_inclusive betweeninclusive`, `is_null empty`, `is_not_null notempty`. + +## FunctionSpec + +Routes are defined by the server app, so calls take a `path`. The server never reads a request body. + +| Method | Server handler | Result | +|---|---|---| +| `query(path, params=None, options=None, *, method="GET")` | `SqlQuery` (single record) | `{success, data}` | +| `query_list(path, params=None, options=None, *, method="GET")` | `SqlQueryList` | `{success, data, metadata}` (from `Content-Range: items a-b/total`) | + +- `params` -> query string. `bool` -> `true/false`, `None` skipped, `list` -> repeated key (server: `IN` filter). `p-` prefixed names are substituted into the SQL. +- `options` -> `X-*` headers. Query values override headers of the same name. +- 206 Partial Content (more rows than returned) is treated as success. + +| Option | Header | +|---|---| +| `filters` (`eq`+AND) | `X-FieldFilter-{col}` | +| `filters` (other) | `X-SearchOp-{op}-{col}` / `X-SearchOr-{op}-{col}` | +| `search_filters` `{col: text}` | `X-SearchFilter-{col}` (ILIKE) | +| `custom_sql_where` / `custom_sql_or` | `X-Custom-SQL-W` / `X-Custom-SQL-Or` | +| `sort` | `X-Sort` as SQL terms: `col ASC,col DESC` | +| `limit` / `offset` | `X-Limit` / `X-Offset` | +| `distinct`, `skip_count`, `skip_cache` | `X-Distinct`, `X-SkipCount`, `X-SkipCache` | +| `response_format` | `X-SimpleApi` / `X-DetailApi` / `X-Syncfusion` (`data` shape changes: array / `{items,...}` / `{result,count}`) | + +Server limits: +- `sort` goes verbatim into `ORDER BY`; `-col` (restheadspec style) does **not** mean DESC. +- `X-Select-Fields` / `X-Not-Select-Fields` are no-ops server-side, so not exposed. +- One search operator per column; same column twice keeps the last. +- Values starting with `ZIP_` / `__` are base64-decoded by the server; such plaintext cannot be sent. +- Non-ASCII / control-char values are sent `ZIP_`-encoded automatically. + +## WebSocketSpec + +`WebSocketClient(url, *, reconnect=True, reconnect_interval=3.0, max_reconnect_attempts=10, heartbeat_interval=30.0, request_timeout=30.0, subscribe_timeout=10.0, headers=None)` + +| Method | Notes | +|---|---| +| `connect()` / `close()` | also `async with` | +| `request(operation, entity, *, schema, record_id, data, options)` | returns response `data` | +| `read(entity, *, schema, record_id, filters, columns, sort, preload, limit, offset)` | | +| `create(entity, data, *, schema)` | | +| `update(entity, id, data, *, schema)` | | +| `delete(entity, id, *, schema)` | | +| `meta(entity, *, schema)` | | +| `subscribe(entity, callback, *, schema, filters)` | returns subscription id; callback gets notification dict (sync or async) | +| `unsubscribe(subscription_id)` | | +| `on(event, cb)` / `off(event)` | events: `connect`, `disconnect`, `error`, `message`, `state_change` | +| `state`, `is_connected()`, `get_subscriptions()` | | + +Auto-reconnect does not restore subscriptions; re-subscribe on `connect`. + +## Errors + +`ResolveSpecError(message, status_code, code, details)` on non-2xx (REST) or failed response / timeout / not connected (WS). + +## Dev + +``` +pip install -e '.[dev]' +pytest +``` diff --git a/resolvespec-python/pyproject.toml b/resolvespec-python/pyproject.toml new file mode 100644 index 0000000..b5aa80b --- /dev/null +++ b/resolvespec-python/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "resolvespec" +version = "1.0.0" +description = "Python client for ResolveSpec REST, HeaderSpec and WebSocket APIs" +readme = "README.md" +requires-python = ">=3.11" +license = { text = "MIT" } +authors = [{ name = "Hein (Warkanum) Puth" }] +keywords = ["resolvespec", "headerspec", "websocket", "rest-client", "api-client"] +dependencies = ["httpx>=0.27", "websockets>=13"] + +[project.optional-dependencies] +dev = ["pytest>=8", "pytest-asyncio>=0.23", "pytest-cov"] + +[tool.hatch.build.targets.wheel] +packages = ["src/resolvespec"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +asyncio_mode = "auto" diff --git a/resolvespec-python/src/resolvespec/__init__.py b/resolvespec-python/src/resolvespec/__init__.py new file mode 100644 index 0000000..f273fe4 --- /dev/null +++ b/resolvespec-python/src/resolvespec/__init__.py @@ -0,0 +1,43 @@ +"""ResolveSpec Python client: REST (ResolveSpec), HeaderSpec and WebSocketSpec.""" +from typing import Mapping, Optional + +from .headerspec import ( + AsyncHeaderSpecClient, + HeaderSpecClient, + build_headers, + decode_header_value, + encode_header_value, +) +from .funcspec import AsyncFuncSpecClient, FuncSpecClient +from .http import ResolveSpecError, merge_headers +from .resolvespec import AsyncResolveSpecClient, ResolveSpecClient +from .types import * # noqa: F401,F403 +from .websocket import Subscription, WebSocketClient + + +def _cache_key(base_url: str, token: Optional[str], headers: Optional[Mapping[str, str]]): + return ( + base_url, + token, + tuple(sorted((k.lower(), v) for k, v in (headers or {}).items())), + ) + + +_resolvespec: dict = {} +_headerspec: dict = {} + + +def get_resolvespec_client(base_url: str, token: Optional[str] = None, headers: Optional[Mapping[str, str]] = None) -> ResolveSpecClient: + """Cached sync client, keyed by base_url + token + headers (case-insensitive names).""" + key = _cache_key(base_url, token, headers) + if key not in _resolvespec: + _resolvespec[key] = ResolveSpecClient(base_url, token, headers) + return _resolvespec[key] + + +def get_headerspec_client(base_url: str, token: Optional[str] = None, headers: Optional[Mapping[str, str]] = None) -> HeaderSpecClient: + """Cached sync client, keyed by base_url + token + headers (case-insensitive names).""" + key = _cache_key(base_url, token, headers) + if key not in _headerspec: + _headerspec[key] = HeaderSpecClient(base_url, token, headers) + return _headerspec[key] diff --git a/resolvespec-python/src/resolvespec/funcspec.py b/resolvespec-python/src/resolvespec/funcspec.py new file mode 100644 index 0000000..35131fd --- /dev/null +++ b/resolvespec-python/src/resolvespec/funcspec.py @@ -0,0 +1,197 @@ +"""FunctionSpec client: calls user-defined SQL endpoints (Go pkg/funcspec). + +Routes are defined by the server application, so calls take a `path`. +Parameters are sent as query string values and/or `X-*` headers; the server never +reads a request body. Query-string values override headers of the same name. + +Server behaviour worth knowing (pkg/funcspec): + - `sort` is inserted raw into ORDER BY, so it must be SQL (`col DESC`), not `-col`. + - Field selection (`X-Select-Fields`) is a no-op server-side, so it is not exposed. + - Only one search operator per column is kept. + - Values starting with `ZIP_` or `__` are base64-decoded by the server (even after our + own encoding), so such plaintext values cannot be sent faithfully. +""" +from __future__ import annotations + +import re +from typing import Any, Dict, List, Mapping, Optional + +import httpx + +from .headerspec import _OPERATOR_MAP, _bool, _filter_value, encode_header_value +from .http import client_headers, error_from, merge_headers, parse_json +from .types import APIResponse, FuncSpecOptions + +Params = Mapping[str, Any] + +_CONTENT_RANGE = re.compile(r"(\d+)-(\d+)/(\d+)") + + +def _safe(value: str) -> str: + """Encode values that are unsafe as raw header/query text (non-ASCII, control chars, edge spaces).""" + if not value.isascii() or not value.isprintable() or value != value.strip(): + return encode_header_value(value) + return value + + +def build_headers(options: Mapping[str, Any]) -> Dict[str, str]: + """Build the X-* headers understood by funcspec.ParseParameters.""" + h: Dict[str, str] = {} + o = options + + for f in o.get("filters") or []: + operator = f["operator"] + logic = f.get("logic_operator") or "AND" + value = _safe(_filter_value(f)) + if operator == "eq" and logic == "AND": + h[f"X-FieldFilter-{f['column']}"] = value + else: + kind = "X-SearchOr" if logic == "OR" else "X-SearchOp" + h[f"{kind}-{_OPERATOR_MAP.get(operator, operator)}-{f['column']}"] = value + + for col, text in (o.get("search_filters") or {}).items(): + h[f"X-SearchFilter-{col}"] = _safe(str(text)) # CAST(col AS TEXT) ILIKE %text% + + if o.get("custom_sql_where"): + h["X-Custom-SQL-W"] = _safe(o["custom_sql_where"]) + if o.get("custom_sql_or"): + h["X-Custom-SQL-Or"] = _safe(o["custom_sql_or"]) + + if o.get("sort"): + h["X-Sort"] = _safe(",".join(_sort_term(s) for s in o["sort"])) + if o.get("limit") is not None: + h["X-Limit"] = str(o["limit"]) + if o.get("offset") is not None: + h["X-Offset"] = str(o["offset"]) + + for name, key in (("X-Distinct", "distinct"), ("X-SkipCount", "skip_count"), ("X-SkipCache", "skip_cache")): + if o.get(key) is not None: + h[name] = _bool(o[key]) + + fmt = o.get("response_format") + if fmt: + h[{"simple": "X-SimpleApi", "detail": "X-DetailApi", "syncfusion": "X-Syncfusion"}[fmt]] = "true" + return h + + +def _sort_term(s: Mapping[str, str]) -> str: + # funcspec puts this verbatim into ORDER BY + return f"{s['column']} {'DESC' if s.get('direction', 'asc').upper() == 'DESC' else 'ASC'}" + + +def build_query(params: Optional[Params]) -> Dict[str, Any]: + """Query-string values: bools -> true/false, lists -> repeated keys (server: IN filter).""" + out: Dict[str, Any] = {} + for k, v in (params or {}).items(): + if v is None: + continue + if isinstance(v, (list, tuple)): + out[k] = [_safe(_q(x)) for x in v] + else: + out[k] = _safe(_q(v)) + return out + + +def _q(v: Any) -> str: + return _bool(v) if isinstance(v, bool) else str(v) + + +def _metadata(response: httpx.Response, options: Optional[Mapping[str, Any]]) -> Dict[str, int]: + """Content-Range is `items {offset}-{offset+len}/{total}`.""" + m = _CONTENT_RANGE.search(response.headers.get("content-range", "")) + start, end, total = (int(x) for x in m.groups()) if m else (0, 0, 0) + return { + "total": total, + "count": end - start, + "filtered": total, + "offset": start, + "limit": int((options or {}).get("limit") or 0), + } + + +def _wrap(response: httpx.Response, options: Optional[Mapping[str, Any]], with_metadata: bool) -> APIResponse: + data = parse_json(response) + if not response.is_success: # 206 Partial Content is success + raise error_from(response, data) + result: APIResponse = {"success": True, "data": data} + if with_metadata: + result["metadata"] = _metadata(response, options) + return result + + +class _Base: + def __init__( + self, + base_url: str, + token: Optional[str] = None, + headers: Optional[Mapping[str, str]] = None, + timeout: Optional[float] = 30.0, + ): + self.base_url = base_url + self.token = token + self.headers = dict(headers or {}) # snapshot + self.timeout = timeout + + def _req(self, method: str, path: str, params: Optional[Params], options: Optional[Mapping[str, Any]]): + url = f"{self.base_url.rstrip('/')}/{path.lstrip('/')}" + headers = merge_headers( + client_headers(self.token, self.headers), + build_headers(options) if options else {}, + ) + return method.upper(), url, headers, build_query(params) + + +class FuncSpecClient(_Base): + """Synchronous client. Use as a context manager or call close().""" + + def __init__(self, *args: Any, transport: Optional[httpx.BaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.Client(timeout=self.timeout, transport=transport) + + def close(self) -> None: + self._http.close() + + def __enter__(self) -> "FuncSpecClient": + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + def _send(self, req, options, with_metadata) -> APIResponse: + method, url, headers, query = req + return _wrap(self._http.request(method, url, headers=headers, params=query), options, with_metadata) + + def query(self, path: str, params: Optional[Params] = None, options: Optional[FuncSpecOptions] = None, *, method: str = "GET") -> APIResponse: + """Single-record endpoint (Handler.SqlQuery). `data` is the row object.""" + return self._send(self._req(method, path, params, options), options, False) + + def query_list(self, path: str, params: Optional[Params] = None, options: Optional[FuncSpecOptions] = None, *, method: str = "GET") -> APIResponse: + """List endpoint (Handler.SqlQueryList). Adds `metadata` from Content-Range.""" + return self._send(self._req(method, path, params, options), options, True) + + +class AsyncFuncSpecClient(_Base): + """Asyncio client. Use as an async context manager or await aclose().""" + + def __init__(self, *args: Any, transport: Optional[httpx.AsyncBaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.AsyncClient(timeout=self.timeout, transport=transport) + + async def aclose(self) -> None: + await self._http.aclose() + + async def __aenter__(self) -> "AsyncFuncSpecClient": + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.aclose() + + async def _send(self, req, options, with_metadata) -> APIResponse: + method, url, headers, query = req + return _wrap(await self._http.request(method, url, headers=headers, params=query), options, with_metadata) + + async def query(self, path: str, params: Optional[Params] = None, options: Optional[FuncSpecOptions] = None, *, method: str = "GET") -> APIResponse: + return await self._send(self._req(method, path, params, options), options, False) + + async def query_list(self, path: str, params: Optional[Params] = None, options: Optional[FuncSpecOptions] = None, *, method: str = "GET") -> APIResponse: + return await self._send(self._req(method, path, params, options), options, True) diff --git a/resolvespec-python/src/resolvespec/headerspec.py b/resolvespec-python/src/resolvespec/headerspec.py new file mode 100644 index 0000000..831c772 --- /dev/null +++ b/resolvespec-python/src/resolvespec/headerspec.py @@ -0,0 +1,336 @@ +"""HeaderSpec client: query options sent as HTTP headers (Go restheadspec). + +Methods: GET=read, POST=create, PUT=update, DELETE=delete. +""" +from __future__ import annotations + +import base64 +import json +import re +from typing import Any, Dict, Mapping, Optional + +import httpx + +from .http import build_url, client_headers, error_from, merge_headers, parse_json +from .types import APIResponse, FilterOption, HeaderSpecOptions + +_PREFIXES = ("ZIP_", "__") + +_OPERATOR_MAP = { + "eq": "equals", + "neq": "notequals", + "gt": "greaterthan", + "gte": "greaterthanorequal", + "lt": "lessthan", + "lte": "lessthanorequal", + "like": "contains", + "ilike": "contains", + "contains": "contains", + "startswith": "beginswith", + "endswith": "endswith", + "in": "in", + "between": "between", + "between_inclusive": "betweeninclusive", + "is_null": "empty", + "is_not_null": "notempty", +} + + +def encode_header_value(value: str) -> str: + """Base64 (UTF-8) with ZIP_ prefix, for complex header values.""" + return "ZIP_" + base64.b64encode(value.encode("utf-8")).decode("ascii") + + +def decode_header_value(value: str) -> str: + """Decode a value that may carry a ZIP_ or __ base64 prefix (nested allowed).""" + code = value + for prefix in _PREFIXES: + if code.startswith(prefix): + b64 = re.sub(r"[\n\r ]", "", code[len(prefix):]) + b64 += "=" * (-len(b64) % 4) + code = base64.b64decode(b64).decode("utf-8") + break + if code.startswith(_PREFIXES): + code = decode_header_value(code) + return code + + +def _geo_header(operator: str) -> Optional[str]: + op = operator.lower() + if op.endswith("_within"): + return "X-VectorFilter-" + if op.startswith("st_") or op in ("bbox", "&&"): + return "X-SpatialFilter-" + return None + + +def _filter_value(f: FilterOption) -> str: + v = f.get("value") + if v is None: + return "" + if isinstance(v, (list, tuple)): + return ",".join(_scalar(x) for x in v) + return _scalar(v) + + +def _scalar(v: Any) -> str: + if isinstance(v, bool): # match JS String(true) + return "true" if v else "false" + return str(v) + + +def _bool(v: bool) -> str: + return "true" if v else "false" + + +def _preload_spec(p: Mapping[str, Any]) -> str: + cols = p.get("columns") + return f"{p['relation']}:{','.join(cols)}" if cols else p["relation"] + + +def build_headers(options: HeaderSpecOptions) -> Dict[str, str]: + """Build restheadspec HTTP headers from options. See README for the mapping.""" + h: Dict[str, str] = {} + o = options + + if o.get("columns"): + h["X-Select-Fields"] = ",".join(o["columns"]) + if o.get("omit_columns"): + h["X-Not-Select-Fields"] = ",".join(o["omit_columns"]) + + for f in o.get("filters") or []: + logic = f.get("logic_operator") or "AND" + operator = f["operator"] + op = _OPERATOR_MAP.get(operator, operator) + value = _filter_value(f) + geo = _geo_header(operator) + if geo: + payload: Dict[str, Any] = {"op": operator, "value": f.get("value")} + if logic == "OR": + payload["logic"] = "or" + h[f"{geo}{f['column']}"] = json.dumps(payload, separators=(",", ":")) + elif operator == "eq" and logic == "AND": + h[f"X-FieldFilter-{f['column']}"] = value + elif logic == "OR": + h[f"X-SearchOr-{op}-{f['column']}"] = value + else: + h[f"X-SearchOp-{op}-{f['column']}"] = value + + if o.get("sort"): + h["X-Sort"] = ",".join( + ("-" if s["direction"].upper() == "DESC" else "+") + s["column"] for s in o["sort"] + ) + + if o.get("limit") is not None: + h["X-Limit"] = str(o["limit"]) + if o.get("offset") is not None: + h["X-Offset"] = str(o["offset"]) + if o.get("cursor_forward"): + h["X-Cursor-Forward"] = o["cursor_forward"] + if o.get("cursor_backward"): + h["X-Cursor-Backward"] = o["cursor_backward"] + + if o.get("preload"): + # Go applies X-Preload-Where to every preload in the matching X-Preload header, + # so preloads are grouped by where clause. + groups: Dict[str, list] = {} + for p in o["preload"]: + groups.setdefault(p.get("where") or "", []).append(_preload_spec(p)) + n = 0 + for where, specs in groups.items(): + if not where: + h["X-Preload"] = "|".join(specs) + elif "" not in groups and n == 0: + # X-Preload-Where would also apply to a where-less X-Preload, so only use it alone + h["X-Preload"] = "|".join(specs) + h["X-Preload-Where"] = where + n += 1 + else: + n += 1 + h[f"X-Preload-{n}"] = "|".join(specs) + h[f"X-Preload-{n}-Where"] = where + + if o.get("expand"): + h["X-Expand"] = "|".join(_preload_spec(e) for e in o["expand"]) + if o.get("custom_sql_joins"): + h["X-Custom-SQL-Join"] = "|".join(o["custom_sql_joins"]) + if o.get("custom_sql_or"): + h["X-Custom-SQL-Or"] = " OR ".join(o["custom_sql_or"]) + if o.get("search_columns"): + h["X-SearchCols"] = ",".join(o["search_columns"]) + for col, sql in (o.get("advanced_sql") or {}).items(): + h[f"X-AdvSQL-{col}"] = sql + + vs = o.get("vector_search") + if vs: + h[f"X-Vector-Search-{vs['column']}"] = vs.get("metric") or "l2" + h["X-Vector-Search-Vector"] = json.dumps(vs["vector"], separators=(",", ":")) + if vs.get("as"): + h["X-Vector-Search-As"] = vs["as"] + if vs.get("direction"): + h["X-Vector-Search-Dir"] = vs["direction"] + + for name, key in ( + ("X-Clean-JSON", "clean_json"), + ("X-Distinct", "distinct"), + ("X-SkipCount", "skip_count"), + ("X-SkipCache", "skip_cache"), + ("X-Transaction-Atomic", "atomic_transaction"), + ("X-Single-Record-As-Object", "single_record_as_object"), + ): + if o.get(key) is not None: + h[name] = _bool(o[key]) + + if o.get("pk_row"): + h["X-PKRow"] = o["pk_row"] + + fmt = o.get("response_format") + if fmt: + h[{"simple": "X-SimpleApi", "detail": "X-DetailApi", "syncfusion": "X-Syncfusion"}[fmt]] = "true" + + if o.get("xfiles"): + h["X-Files"] = encode_header_value(json.dumps(o["xfiles"], separators=(",", ":"))) + + if o.get("fetch_row_number"): + h["X-Fetch-RowNumber"] = o["fetch_row_number"] + + for cc in o.get("computedColumns") or []: + h[f"X-CQL-SEL-{cc['name']}"] = cc["expression"] + + if o.get("customOperators"): + h["X-Custom-SQL-W"] = " AND ".join(co["sql"] for co in o["customOperators"]) + + return h + + +def _int(s: Optional[str]) -> int: + try: + return int(s) # type: ignore[arg-type] + except (TypeError, ValueError): + return 0 + + +def _wrap(response: httpx.Response) -> APIResponse: + """Wrap a raw restheadspec body, deriving metadata from Content-Range / X-Limit.""" + data = parse_json(response) + if not response.is_success: + raise error_from(response, data) + cr = response.headers.get("content-range") + total = _int(cr.split("/")[-1]) if cr else 0 + offset = _int(cr.split("/")[0].split("-")[0].split(" ")[-1]) if cr else 0 + return { + "data": data, + "success": True, + "error": data.get("error") if isinstance(data, dict) else None, + "metadata": { + "count": total, + "total": total, + "filtered": total, + "offset": offset, + "limit": _int(response.headers.get("x-limit")), + }, + } + + +class _Base: + def __init__( + self, + base_url: str, + token: Optional[str] = None, + headers: Optional[Mapping[str, str]] = None, + timeout: Optional[float] = 30.0, + ): + self.base_url = base_url + self.token = token + self.headers = dict(headers or {}) # snapshot + self.timeout = timeout + + def _base_headers(self) -> Dict[str, str]: + return client_headers(self.token, self.headers) + + def _req(self, method, schema, entity, id, options=None, body=None): + opt = build_headers(options) if options else {} + return ( + method, + build_url(self.base_url, schema, entity, id), + merge_headers(self._base_headers(), opt), + body, + ) + + def _read_req(self, schema, entity, id, options): + return self._req("GET", schema, entity, id, options) + + def _create_req(self, schema, entity, data, options): + return self._req("POST", schema, entity, None, options, data) + + def _update_req(self, schema, entity, id, data, options): + return self._req("PUT", schema, entity, id, options, data) + + def _delete_req(self, schema, entity, id): + return self._req("DELETE", schema, entity, id) + + +class HeaderSpecClient(_Base): + """Synchronous client. Use as a context manager or call close().""" + + def __init__(self, *args: Any, transport: Optional[httpx.BaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.Client(timeout=self.timeout, transport=transport) + + def close(self) -> None: + self._http.close() + + def __enter__(self) -> "HeaderSpecClient": + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + def _send(self, req) -> APIResponse: + method, url, headers, body = req + return _wrap(self._http.request(method, url, headers=headers, json=body)) + + def read(self, schema: str, entity: str, id: Optional[str] = None, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return self._send(self._read_req(schema, entity, id, options)) + + def create(self, schema: str, entity: str, data: Any, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return self._send(self._create_req(schema, entity, data, options)) + + def update(self, schema: str, entity: str, id: str, data: Any, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return self._send(self._update_req(schema, entity, id, data, options)) + + def delete(self, schema: str, entity: str, id: str) -> APIResponse: + return self._send(self._delete_req(schema, entity, id)) + + +class AsyncHeaderSpecClient(_Base): + """Asyncio client. Use as an async context manager or await aclose().""" + + def __init__(self, *args: Any, transport: Optional[httpx.AsyncBaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.AsyncClient(timeout=self.timeout, transport=transport) + + async def aclose(self) -> None: + await self._http.aclose() + + async def __aenter__(self) -> "AsyncHeaderSpecClient": + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.aclose() + + async def _send(self, req) -> APIResponse: + method, url, headers, body = req + return _wrap(await self._http.request(method, url, headers=headers, json=body)) + + async def read(self, schema: str, entity: str, id: Optional[str] = None, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return await self._send(self._read_req(schema, entity, id, options)) + + async def create(self, schema: str, entity: str, data: Any, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return await self._send(self._create_req(schema, entity, data, options)) + + async def update(self, schema: str, entity: str, id: str, data: Any, options: Optional[HeaderSpecOptions] = None) -> APIResponse: + return await self._send(self._update_req(schema, entity, id, data, options)) + + async def delete(self, schema: str, entity: str, id: str) -> APIResponse: + return await self._send(self._delete_req(schema, entity, id)) diff --git a/resolvespec-python/src/resolvespec/http.py b/resolvespec-python/src/resolvespec/http.py new file mode 100644 index 0000000..8029bed --- /dev/null +++ b/resolvespec-python/src/resolvespec/http.py @@ -0,0 +1,76 @@ +"""Shared HTTP helpers for the REST clients.""" +from __future__ import annotations + +from typing import Any, Dict, Mapping, Optional +from urllib.parse import quote + + +class ResolveSpecError(Exception): + """Raised on a non-2xx response or an unsuccessful API result.""" + + def __init__( + self, + message: str, + status_code: Optional[int] = None, + code: Optional[str] = None, + details: Any = None, + detail: Optional[str] = None, + ): + super().__init__(message) + self.message = message + self.status_code = status_code + self.code = code + self.details = details + self.detail = detail # server-side reason (funcspec / restheadspec errors) + + +def merge_headers(*sources: Mapping[str, str]) -> Dict[str, str]: + """Merge HTTP headers case-insensitively; the last source wins and keeps its spelling.""" + result: Dict[str, str] = {} + for source in sources: + for name, value in source.items(): + for existing in [k for k in result if k.lower() == name.lower()]: + del result[existing] + result[name] = value + return result + + +def client_headers(token: Optional[str], headers: Optional[Mapping[str, str]]) -> Dict[str, str]: + """Content-Type < custom headers < bearer token.""" + return merge_headers( + {"Content-Type": "application/json"}, + headers or {}, + {"Authorization": f"Bearer {token}"} if token else {}, + ) + + +def build_url(base_url: str, schema: str, entity: str, id: Optional[Any] = None) -> str: + url = f"{base_url.rstrip('/')}/{quote(schema, safe='')}/{quote(entity, safe='')}" + if id is not None and id != "": + url += f"/{quote(str(id), safe='')}" + return url + + +def drop_none(d: Mapping[str, Any]) -> Dict[str, Any]: + return {k: v for k, v in d.items() if v is not None} + + +def parse_json(response: Any) -> Any: + try: + return response.json() + except ValueError: + return None + + +def error_from(response: Any, data: Any) -> ResolveSpecError: + err = data.get("error") if isinstance(data, dict) else None + err = err if isinstance(err, dict) else {} + text = (response.text or "").strip() if data is None else "" + fallback = text[:200] or f"{response.reason_phrase} ({response.status_code})" + return ResolveSpecError( + err.get("message") or fallback, + status_code=response.status_code, + code=err.get("code"), + details=err.get("details"), + detail=err.get("detail"), + ) diff --git a/resolvespec-python/src/resolvespec/resolvespec.py b/resolvespec-python/src/resolvespec/resolvespec.py new file mode 100644 index 0000000..070721b --- /dev/null +++ b/resolvespec-python/src/resolvespec/resolvespec.py @@ -0,0 +1,137 @@ +"""ResolveSpec client: JSON body protocol (POST {operation, data, options}).""" +from __future__ import annotations + +from typing import Any, Dict, List, Mapping, Optional, Tuple + +import httpx + +from .http import build_url, client_headers, drop_none, error_from, parse_json +from .types import APIResponse, Options, RecordId + + +def _url_id(id: Optional[RecordId]) -> Optional[str]: + return str(id) if isinstance(id, (int, str)) else None + + +def _body_id(id: Optional[RecordId]) -> Optional[List[str]]: + return id if isinstance(id, list) else None + + +class _Base: + def __init__( + self, + base_url: str, + token: Optional[str] = None, + headers: Optional[Mapping[str, str]] = None, + timeout: Optional[float] = 30.0, + ): + self.base_url = base_url + self.token = token + self.headers = dict(headers or {}) # snapshot + self.timeout = timeout + + def _headers(self) -> Dict[str, str]: + return client_headers(self.token, self.headers) + + def _request( + self, method: str, schema: str, entity: str, id: Optional[str], body: Optional[Dict[str, Any]] + ) -> Tuple[str, str, Dict[str, str], Optional[Dict[str, Any]]]: + return method, build_url(self.base_url, schema, entity, id), self._headers(), body + + @staticmethod + def _result(response: httpx.Response) -> APIResponse: + data = parse_json(response) + if not response.is_success: + raise error_from(response, data) + return data + + # request builders (shared by sync and async) + def _metadata_req(self, schema, entity): + return self._request("GET", schema, entity, None, None) + + def _read_req(self, schema, entity, id, options): + body = drop_none({"operation": "read", "id": _body_id(id), "options": options}) + return self._request("POST", schema, entity, _url_id(id), body) + + def _create_req(self, schema, entity, data, options): + body = drop_none({"operation": "create", "data": data, "options": options}) + return self._request("POST", schema, entity, None, body) + + def _update_req(self, schema, entity, data, id, options): + body = drop_none({"operation": "update", "id": _body_id(id), "data": data, "options": options}) + return self._request("POST", schema, entity, _url_id(id), body) + + def _delete_req(self, schema, entity, id): + return self._request("POST", schema, entity, str(id), {"operation": "delete"}) + + +class ResolveSpecClient(_Base): + """Synchronous client. Use as a context manager or call close().""" + + def __init__(self, *args: Any, transport: Optional[httpx.BaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.Client(timeout=self.timeout, transport=transport) + + def close(self) -> None: + self._http.close() + + def __enter__(self) -> "ResolveSpecClient": + return self + + def __exit__(self, *exc: Any) -> None: + self.close() + + def _send(self, req) -> APIResponse: + method, url, headers, body = req + return self._result(self._http.request(method, url, headers=headers, json=body)) + + def get_metadata(self, schema: str, entity: str) -> APIResponse: + return self._send(self._metadata_req(schema, entity)) + + def read(self, schema: str, entity: str, id: Optional[RecordId] = None, options: Optional[Options] = None) -> APIResponse: + return self._send(self._read_req(schema, entity, id, options)) + + def create(self, schema: str, entity: str, data: Any, options: Optional[Options] = None) -> APIResponse: + return self._send(self._create_req(schema, entity, data, options)) + + def update(self, schema: str, entity: str, data: Any, id: Optional[RecordId] = None, options: Optional[Options] = None) -> APIResponse: + return self._send(self._update_req(schema, entity, data, id, options)) + + def delete(self, schema: str, entity: str, id: Any) -> APIResponse: + return self._send(self._delete_req(schema, entity, id)) + + +class AsyncResolveSpecClient(_Base): + """Asyncio client. Use as an async context manager or await aclose().""" + + def __init__(self, *args: Any, transport: Optional[httpx.AsyncBaseTransport] = None, **kwargs: Any): + super().__init__(*args, **kwargs) + self._http = httpx.AsyncClient(timeout=self.timeout, transport=transport) + + async def aclose(self) -> None: + await self._http.aclose() + + async def __aenter__(self) -> "AsyncResolveSpecClient": + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.aclose() + + async def _send(self, req) -> APIResponse: + method, url, headers, body = req + return self._result(await self._http.request(method, url, headers=headers, json=body)) + + async def get_metadata(self, schema: str, entity: str) -> APIResponse: + return await self._send(self._metadata_req(schema, entity)) + + async def read(self, schema: str, entity: str, id: Optional[RecordId] = None, options: Optional[Options] = None) -> APIResponse: + return await self._send(self._read_req(schema, entity, id, options)) + + async def create(self, schema: str, entity: str, data: Any, options: Optional[Options] = None) -> APIResponse: + return await self._send(self._create_req(schema, entity, data, options)) + + async def update(self, schema: str, entity: str, data: Any, id: Optional[RecordId] = None, options: Optional[Options] = None) -> APIResponse: + return await self._send(self._update_req(schema, entity, data, id, options)) + + async def delete(self, schema: str, entity: str, id: Any) -> APIResponse: + return await self._send(self._delete_req(schema, entity, id)) diff --git a/resolvespec-python/src/resolvespec/types.py b/resolvespec-python/src/resolvespec/types.py new file mode 100644 index 0000000..085d5b0 --- /dev/null +++ b/resolvespec-python/src/resolvespec/types.py @@ -0,0 +1,166 @@ +"""Types aligned with Go pkg/common/types.go. Dict keys are the wire names.""" +from __future__ import annotations + +from typing import Any, Dict, List, NotRequired, TypedDict, Union + +Operator = str # eq neq gt gte lt lte like ilike in contains startswith endswith +# between between_inclusive is_null is_not_null +# st_dwithin bbox (spatial) | l2_within cosine_within ip_within (vector) +Operation = str # read | create | update | delete +SortDirection = str # asc | desc | ASC | DESC +VectorMetric = str # l2 | cosine | ip +ResponseFormat = str # simple | detail | syncfusion + +RecordId = Union[int, str, List[str]] + + +class Parameter(TypedDict): + name: str + value: str + sequence: NotRequired[int] + + +class FilterOption(TypedDict): + column: str + operator: str + value: Any + logic_operator: NotRequired[str] # "AND" | "OR" + + +class SortOption(TypedDict): + column: str + direction: str + + +class CustomOperator(TypedDict): + name: str + sql: str + + +class ComputedColumn(TypedDict): + name: str + expression: str + + +class PreloadOption(TypedDict, total=False): + relation: str + table_name: str + columns: List[str] + omit_columns: List[str] + sort: List[SortOption] + filters: List[FilterOption] + where: str + limit: int + offset: int + updateable: bool + computed_ql: Dict[str, str] + recursive: bool + primary_key: str + related_key: str + foreign_key: str + recursive_child_key: str + sql_joins: List[str] + join_aliases: List[str] + + +# `as` is a keyword, so the functional syntax is required. +VectorSearchOption = TypedDict( + "VectorSearchOption", + { + "column": str, + "vector": List[float], + "metric": str, # l2 (default) | cosine | ip + "as": str, # distance column alias, default _distance + "direction": str, # asc (default) | desc + }, + total=False, +) + + +class ExpandOption(TypedDict, total=False): + relation: str + columns: List[str] + + +class XFiles(TypedDict, total=False): + tablename: str + schema: str + primarykey: str + foreignkey: str + relatedkey: str + sort: List[str] + prefix: str + editable: bool + recursive: bool + expand: bool + rownumber: bool + skipcount: bool + offset: int + limit: int + columns: List[str] + omit_columns: List[str] + cql_columns: List[str] + sql_joins: List[str] + sql_or: List[str] + sql_and: List[str] + parenttables: List["XFiles"] + childtables: List["XFiles"] + filter_fields: List[Dict[str, str]] + cursor_forward: str + cursor_backward: str + + +class Options(TypedDict, total=False): + preload: List[PreloadOption] + columns: List[str] + omit_columns: List[str] + filters: List[FilterOption] + sort: List[SortOption] + limit: int + offset: int + customOperators: List[CustomOperator] + computedColumns: List[ComputedColumn] + parameters: List[Parameter] + cursor_forward: str + cursor_backward: str + fetch_row_number: str + vector_search: VectorSearchOption + + +class HeaderSpecOptions(Options, total=False): + """Options only available to the header-based (restheadspec) protocol.""" + + expand: List[ExpandOption] # X-Expand + custom_sql_joins: List[str] # X-Custom-SQL-Join + custom_sql_or: List[str] # X-Custom-SQL-Or + search_columns: List[str] # X-SearchCols + advanced_sql: Dict[str, str] # X-AdvSQL-{col} + clean_json: bool # X-Clean-JSON + distinct: bool # X-Distinct + skip_count: bool # X-SkipCount + skip_cache: bool # X-SkipCache + pk_row: str # X-PKRow + response_format: str # X-SimpleApi / X-DetailApi / X-Syncfusion + single_record_as_object: bool # X-Single-Record-As-Object + atomic_transaction: bool # X-Transaction-Atomic + xfiles: XFiles # X-Files + + +class FuncSpecOptions(TypedDict, total=False): + """Options understood by funcspec endpoints (sent as X-* headers).""" + + filters: List[FilterOption] # eq+AND -> X-FieldFilter; others X-SearchOp / X-SearchOr (one per column) + search_filters: Dict[str, str] # X-SearchFilter-{col}: text ILIKE + custom_sql_where: str # X-Custom-SQL-W + custom_sql_or: str # X-Custom-SQL-Or + sort: List[SortOption] # sent as SQL ORDER BY terms ("col DESC") + limit: int + offset: int + distinct: bool + skip_count: bool + skip_cache: bool + response_format: str # simple | detail | syncfusion + + +# Responses are plain dicts: {"success", "data", "metadata"?, "error"?} +APIResponse = Dict[str, Any] diff --git a/resolvespec-python/src/resolvespec/websocket.py b/resolvespec-python/src/resolvespec/websocket.py new file mode 100644 index 0000000..b14bdbc --- /dev/null +++ b/resolvespec-python/src/resolvespec/websocket.py @@ -0,0 +1,335 @@ +"""WebSocketSpec client (asyncio). Mirrors the Go websocketspec message protocol.""" +from __future__ import annotations + +import asyncio +import json +import logging +import uuid +from dataclasses import dataclass, field +from typing import Any, Awaitable, Callable, Dict, List, Optional, Union + +from websockets.asyncio.client import ClientConnection, connect + +from .http import ResolveSpecError +from .types import FilterOption, PreloadOption, SortOption + +log = logging.getLogger("resolvespec.websocket") + +# Connection states +DISCONNECTED = "disconnected" +CONNECTING = "connecting" +CONNECTED = "connected" +DISCONNECTING = "disconnecting" +RECONNECTING = "reconnecting" + +Notification = Dict[str, Any] +Callback = Callable[[Any], Union[None, Awaitable[None]]] +EVENTS = ("connect", "disconnect", "error", "message", "state_change") + + +@dataclass +class Subscription: + id: str + entity: str + schema: Optional[str] = None + options: Optional[Dict[str, Any]] = None + callback: Optional[Callback] = field(default=None, repr=False) + + +def _drop_none(d: Dict[str, Any]) -> Dict[str, Any]: + return {k: v for k, v in d.items() if v is not None} + + +class WebSocketClient: + """ + Usage: + async with WebSocketClient("ws://localhost:8080/ws") as ws: + rows = await ws.read("users", schema="public", limit=10) + + Events (`on(event, callback)`): connect, disconnect, error, message, state_change. + Callbacks may be sync or async. + """ + + def __init__( + self, + url: str, + *, + reconnect: bool = True, + reconnect_interval: float = 3.0, + max_reconnect_attempts: int = 10, + heartbeat_interval: float = 30.0, + request_timeout: float = 30.0, + subscribe_timeout: float = 10.0, + headers: Optional[Dict[str, str]] = None, + ): + self.url = url + self.reconnect = reconnect + self.reconnect_interval = reconnect_interval + self.max_reconnect_attempts = max_reconnect_attempts + self.heartbeat_interval = heartbeat_interval + self.request_timeout = request_timeout + self.subscribe_timeout = subscribe_timeout + self.headers = dict(headers or {}) + + self._ws: Optional[ClientConnection] = None + self._state = DISCONNECTED + self._pending: Dict[str, "asyncio.Future[Dict[str, Any]]"] = {} + self._subscriptions: Dict[str, Subscription] = {} + self._listeners: Dict[str, Callback] = {} + self._tasks: List["asyncio.Task[Any]"] = [] + self._reader: Optional["asyncio.Task[Any]"] = None + self._manual_close = False + + # ---- lifecycle ------------------------------------------------------- + + async def __aenter__(self) -> "WebSocketClient": + await self.connect() + return self + + async def __aexit__(self, *exc: Any) -> None: + await self.close() + + async def connect(self) -> None: + if self.is_connected(): + return + self._manual_close = False + self._set_state(CONNECTING) + try: + self._ws = await connect(self.url, additional_headers=self.headers or None) + except Exception as e: + self._set_state(DISCONNECTED) + await self._emit("error", e) + raise + self._set_state(CONNECTED) + self._reader = asyncio.create_task(self._read_loop(self._ws)) + self._heartbeat = asyncio.create_task(self._heartbeat_loop()) + await self._emit("connect") + + async def close(self) -> None: + self._manual_close = True + self._set_state(DISCONNECTING) + for t in (self._reader, getattr(self, "_heartbeat", None), getattr(self, "_reconnect_task", None)): + if t and t is not asyncio.current_task(): + t.cancel() + if self._ws: + await self._ws.close() + self._ws = None + self._fail_pending(ResolveSpecError("WebSocket closed")) + self._set_state(DISCONNECTED) + + def is_connected(self) -> bool: + return self._ws is not None and self._state == CONNECTED + + @property + def state(self) -> str: + return self._state + + def on(self, event: str, callback: Callback) -> None: + if event not in EVENTS: + raise ValueError(f"unknown event {event!r}; expected one of {EVENTS}") + self._listeners[event] = callback + + def off(self, event: str) -> None: + self._listeners.pop(event, None) + + def get_subscriptions(self) -> List[Subscription]: + return list(self._subscriptions.values()) + + # ---- operations ------------------------------------------------------ + + async def request( + self, + operation: str, + entity: str, + *, + schema: Optional[str] = None, + record_id: Optional[str] = None, + data: Any = None, + options: Optional[Dict[str, Any]] = None, + ) -> Any: + message = _drop_none({ + "type": "request", + "operation": operation, + "entity": entity, + "schema": schema, + "record_id": record_id, + "data": data, + "options": options, + }) + response = await self._call(message, self.request_timeout, "Request") + return response.get("data") + + async def read( + self, + entity: str, + *, + schema: Optional[str] = None, + record_id: Optional[str] = None, + filters: Optional[List[FilterOption]] = None, + columns: Optional[List[str]] = None, + sort: Optional[List[SortOption]] = None, + preload: Optional[List[PreloadOption]] = None, + limit: Optional[int] = None, + offset: Optional[int] = None, + ) -> Any: + options = _drop_none({ + "filters": filters, "columns": columns, "sort": sort, + "preload": preload, "limit": limit, "offset": offset, + }) + return await self.request("read", entity, schema=schema, record_id=record_id, options=options) + + async def create(self, entity: str, data: Any, *, schema: Optional[str] = None) -> Any: + return await self.request("create", entity, schema=schema, data=data) + + async def update(self, entity: str, id: str, data: Any, *, schema: Optional[str] = None) -> Any: + return await self.request("update", entity, schema=schema, record_id=id, data=data) + + async def delete(self, entity: str, id: str, *, schema: Optional[str] = None) -> None: + await self.request("delete", entity, schema=schema, record_id=id) + + async def meta(self, entity: str, *, schema: Optional[str] = None) -> Any: + return await self.request("meta", entity, schema=schema) + + async def subscribe( + self, + entity: str, + callback: Callback, + *, + schema: Optional[str] = None, + filters: Optional[List[FilterOption]] = None, + ) -> str: + message = _drop_none({ + "type": "subscription", + "operation": "subscribe", + "entity": entity, + "schema": schema, + "options": _drop_none({"filters": filters}), + }) + response = await self._call(message, self.subscribe_timeout, "Subscription") + sub_id = (response.get("data") or {}).get("subscription_id") + if not sub_id: + raise ResolveSpecError("Subscription failed") + self._subscriptions[sub_id] = Subscription( + sub_id, entity, schema, _drop_none({"filters": filters}) or None, callback + ) + return sub_id + + async def unsubscribe(self, subscription_id: str) -> None: + message = {"type": "subscription", "operation": "unsubscribe", "subscription_id": subscription_id} + await self._call(message, self.subscribe_timeout, "Unsubscribe") + self._subscriptions.pop(subscription_id, None) + + # ---- internals ------------------------------------------------------- + + async def _call(self, message: Dict[str, Any], timeout: float, what: str) -> Dict[str, Any]: + self._ensure_connected() + mid = str(uuid.uuid4()) + message["id"] = mid + fut: "asyncio.Future[Dict[str, Any]]" = asyncio.get_running_loop().create_future() + self._pending[mid] = fut + try: + await self._ws.send(json.dumps(message)) # type: ignore[union-attr] + response = await asyncio.wait_for(fut, timeout) + except asyncio.TimeoutError: + raise ResolveSpecError(f"{what} timeout") from None + finally: + self._pending.pop(mid, None) + if not response.get("success"): + err = response.get("error") or {} + raise ResolveSpecError( + err.get("message") or f"{what} failed", code=err.get("code"), details=err.get("details") + ) + return response + + def _ensure_connected(self) -> None: + if not self.is_connected(): + raise ResolveSpecError("WebSocket is not connected. Call connect() first.") + + def _fail_pending(self, exc: Exception) -> None: + for fut in self._pending.values(): + if not fut.done(): + fut.set_exception(exc) + self._pending.clear() + + async def _read_loop(self, ws: ClientConnection) -> None: + try: + async for raw in ws: + await self._handle_message(raw) + except asyncio.CancelledError: + raise + except Exception as e: # connection error + await self._emit("error", e) + # connection ended + if ws is not self._ws: + return + self._ws = None + if hb := getattr(self, "_heartbeat", None): + hb.cancel() + self._fail_pending(ResolveSpecError("WebSocket disconnected")) + self._set_state(DISCONNECTED) + await self._emit("disconnect", ws.close_code, ws.close_reason) + if self.reconnect and not self._manual_close: + self._reconnect_task = asyncio.create_task(self._reconnect()) + + async def _reconnect(self) -> None: + for attempt in range(1, self.max_reconnect_attempts + 1): + if self._manual_close: + return + log.debug("Reconnection attempt %d/%d", attempt, self.max_reconnect_attempts) + self._set_state(RECONNECTING) + await asyncio.sleep(self.reconnect_interval) + try: + await self.connect() + return + except Exception as e: + log.debug("Reconnection failed: %s", e) + self._set_state(DISCONNECTED) + + async def _handle_message(self, raw: Union[str, bytes]) -> None: + try: + message = json.loads(raw) + except ValueError as e: + log.debug("Error parsing message: %s", e) + return + await self._emit("message", message) + kind = message.get("type") + if kind == "response": + fut = self._pending.get(message.get("id")) + if fut and not fut.done(): + fut.set_result(message) + elif kind == "notification": + sub = self._subscriptions.get(message.get("subscription_id")) + if sub and sub.callback: + await _maybe_await(sub.callback(message)) + elif kind != "pong": + log.debug("Unknown message type: %s", kind) + + async def _heartbeat_loop(self) -> None: + try: + while True: + await asyncio.sleep(self.heartbeat_interval) + if self.is_connected(): + await self._ws.send(json.dumps({"id": str(uuid.uuid4()), "type": "ping"})) # type: ignore[union-attr] + except asyncio.CancelledError: + raise + except Exception as e: + log.debug("Heartbeat failed: %s", e) + + def _set_state(self, state: str) -> None: + if self._state != state: + self._state = state + cb = self._listeners.get("state_change") + if cb: + res = cb(state) + if asyncio.iscoroutine(res): + asyncio.ensure_future(res) + + async def _emit(self, event: str, *args: Any) -> None: + cb = self._listeners.get(event) + if cb: + await _maybe_await(cb(*args)) + + +async def _maybe_await(result: Any) -> None: + if asyncio.iscoroutine(result) or isinstance(result, asyncio.Future): + await result diff --git a/resolvespec-python/tests/test_funcspec.py b/resolvespec-python/tests/test_funcspec.py new file mode 100644 index 0000000..3d39625 --- /dev/null +++ b/resolvespec-python/tests/test_funcspec.py @@ -0,0 +1,132 @@ +import httpx +import pytest + +from resolvespec import AsyncFuncSpecClient, FuncSpecClient, ResolveSpecError +from resolvespec.funcspec import build_headers, build_query +from resolvespec.headerspec import decode_header_value + + +def make(handler, **kw): + return FuncSpecClient("http://localhost:3000", "tok", transport=httpx.MockTransport(handler), **kw) + + +def capture(status=200, body=None, headers=None): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(status, json=body if body is not None else [], headers=headers) + + return seen, handler + + +def test_filters(): + h = build_headers({"filters": [ + {"column": "status", "operator": "eq", "value": "active"}, + {"column": "age", "operator": "gte", "value": 18}, + {"column": "name", "operator": "contains", "value": "x", "logic_operator": "OR"}, + {"column": "deleted", "operator": "is_null", "value": None}, + {"column": "id", "operator": "in", "value": [1, 2]}, + {"column": "p", "operator": "between_inclusive", "value": [1, 5]}, + ]}) + assert h == { + "X-FieldFilter-status": "active", + "X-SearchOp-greaterthanorequal-age": "18", + "X-SearchOr-contains-name": "x", + "X-SearchOp-empty-deleted": "", + "X-SearchOp-in-id": "1,2", + "X-SearchOp-betweeninclusive-p": "1,5", + } + + +def test_sort_is_sql_not_prefixed(): + # server inserts sort verbatim into ORDER BY; "-col" would negate the column + h = build_headers({"sort": [{"column": "name", "direction": "asc"}, {"column": "created_at", "direction": "DESC"}]}) + assert h["X-Sort"] == "name ASC,created_at DESC" + + +def test_misc_options(): + h = build_headers({ + "search_filters": {"name": "bob"}, "custom_sql_where": "a = 1", "custom_sql_or": "b = 2", + "limit": 5, "offset": 10, "distinct": True, "skip_count": True, "skip_cache": False, + "response_format": "syncfusion", + }) + assert h == { + "X-SearchFilter-name": "bob", "X-Custom-SQL-W": "a = 1", "X-Custom-SQL-Or": "b = 2", + "X-Limit": "5", "X-Offset": "10", "X-Distinct": "true", "X-SkipCount": "true", + "X-SkipCache": "false", "X-Syncfusion": "true", + } + + +def test_ambiguous_values_are_encoded(): + h = build_headers({"custom_sql_where": "name = 'café'", "filters": [{"column": "c", "operator": "eq", "value": " pad "}]}) + assert h["X-Custom-SQL-W"].startswith("ZIP_") + assert decode_header_value(h["X-Custom-SQL-W"]) == "name = 'café'" + assert decode_header_value(h["X-FieldFilter-c"]) == " pad " + + +def test_build_query(): + q = build_query({"p-id": 5, "flag": True, "ids": [1, 2], "skip": None, "m": "match=ab"}) + assert q == {"p-id": "5", "flag": "true", "ids": ["1", "2"], "m": "match=ab"} + + +def test_query_list_request_and_metadata(): + seen, h = capture(206, [{"id": 1}, {"id": 2}], {"content-range": "items 10-12/50"}) + with make(h) as c: + res = c.query_list("/api/orders", {"p-status": "open", "id": [1, 2]}, {"limit": 2, "offset": 10}) + r = seen[0] + assert r.method == "GET" + assert r.url.path == "/api/orders" + assert r.url.params.multi_items() == [("p-status", "open"), ("id", "1"), ("id", "2")] + assert r.headers["x-limit"] == "2" and r.headers["authorization"] == "Bearer tok" + assert res == { + "success": True, + "data": [{"id": 1}, {"id": 2}], + "metadata": {"total": 50, "count": 2, "filtered": 50, "offset": 10, "limit": 2}, + } + + +def test_query_list_empty_result(): + seen, h = capture(200, [], {"content-range": "items 0-0/0"}) + with make(h) as c: + assert c.query_list("orders")["metadata"]["total"] == 0 + assert seen[0].url.path == "/orders" + + +def test_query_single_has_no_metadata_and_method(): + seen, h = capture(200, {"id": 1}) + with make(h) as c: + res = c.query("api/order", method="post") + assert seen[0].method == "POST" + assert res == {"success": True, "data": {"id": 1}} + + +def test_detail_format_data_passthrough(): + body = {"items": [{"a": 1}], "count": "1", "total": "1", "tablename": "/x", "tableprefix": "gsql"} + _, h = capture(200, body, {"content-range": "items 0-1/1"}) + with make(h) as c: + assert c.query_list("x", options={"response_format": "detail"})["data"] == body + + +def test_server_error_shape(): + err = {"success": False, "error": {"code": "query_failed", "message": "Failed to retrieve records", "detail": "no such column", "sql": "SELECT"}} + _, h = capture(400, err) + with make(h) as c: + with pytest.raises(ResolveSpecError, match="Failed to retrieve") as ei: + c.query_list("x") + assert ei.value.code == "query_failed" and ei.value.detail == "no such column" and ei.value.status_code == 400 + + +def test_plain_text_panic_error(): + with make(lambda r: httpx.Response(500, text="Internal server error: boom")) as c: + with pytest.raises(ResolveSpecError, match="boom"): + c.query("x") + + +async def test_async(): + async def handler(req): + return httpx.Response(200, json=[{"id": 1}], headers={"content-range": "items 0-1/1"}) + + async with AsyncFuncSpecClient("http://localhost:3000", transport=httpx.MockTransport(handler)) as c: + assert (await c.query_list("x"))["metadata"]["total"] == 1 + assert (await c.query("x"))["data"] == [{"id": 1}] diff --git a/resolvespec-python/tests/test_headerspec.py b/resolvespec-python/tests/test_headerspec.py new file mode 100644 index 0000000..276e4e6 --- /dev/null +++ b/resolvespec-python/tests/test_headerspec.py @@ -0,0 +1,236 @@ +import json + +import httpx +import pytest + +from resolvespec import ( + AsyncHeaderSpecClient, + HeaderSpecClient, + ResolveSpecError, + build_headers, + decode_header_value, + encode_header_value, + get_headerspec_client, +) +import base64 + +CFG = dict(base_url="http://localhost:3000", token="tok") + + +# ---- build_headers (ported from headerspec.test.ts) ---- + +def test_preload_shared_where(): + h = build_headers({"preload": [ + {"relation": "Items", "columns": ["id"], "where": "active = true"}, + {"relation": "Tags", "where": "active = true"}, + ]}) + assert h["X-Preload"] == "Items:id|Tags" + assert h["X-Preload-Where"] == "active = true" + + +def test_preload_mixed_where_numbered(): + h = build_headers({"preload": [ + {"relation": "Items", "where": "a = 1"}, + {"relation": "Category"}, + {"relation": "Tags", "where": "b = 2"}, + ]}) + assert h["X-Preload"] == "Category" + assert "X-Preload-Where" not in h + assert h["X-Preload-1"] == "Items" and h["X-Preload-1-Where"] == "a = 1" + assert h["X-Preload-2"] == "Tags" and h["X-Preload-2-Where"] == "b = 2" + + +def test_expand_joins_or_searchcols_advsql(): + h = build_headers({ + "expand": [{"relation": "Dept", "columns": ["id", "name"]}, {"relation": "Role"}], + "custom_sql_joins": ["LEFT JOIN a ON a.id = b.id", "INNER JOIN c ON c.id = b.cid"], + "custom_sql_or": ["x = 1", "y = 2"], + "search_columns": ["name", "email"], + "advanced_sql": {"total": "a + b"}, + }) + assert h["X-Expand"] == "Dept:id,name|Role" + assert h["X-Custom-SQL-Join"] == "LEFT JOIN a ON a.id = b.id|INNER JOIN c ON c.id = b.cid" + assert h["X-Custom-SQL-Or"] == "x = 1 OR y = 2" + assert h["X-SearchCols"] == "name,email" + assert h["X-AdvSQL-total"] == "a + b" + + +def test_flags_pkrow_format(): + h = build_headers({ + "clean_json": True, "distinct": True, "skip_count": True, "skip_cache": False, + "atomic_transaction": True, "single_record_as_object": False, + "pk_row": "42", "response_format": "detail", + }) + assert h["X-Clean-JSON"] == "true" + assert h["X-Distinct"] == "true" + assert h["X-SkipCount"] == "true" + assert h["X-SkipCache"] == "false" + assert h["X-Transaction-Atomic"] == "true" + assert h["X-Single-Record-As-Object"] == "false" + assert h["X-PKRow"] == "42" + assert h["X-DetailApi"] == "true" + + +def test_spatial_and_vector_filters(): + h = build_headers({"filters": [ + {"column": "geom", "operator": "st_dwithin", "value": {"geom": "POINT(0 0)", "distance": 5}, "logic_operator": "OR"}, + {"column": "emb", "operator": "cosine_within", "value": {"vector": [1, 2], "distance": 0.3}}, + ]}) + assert json.loads(h["X-SpatialFilter-geom"]) == { + "op": "st_dwithin", "value": {"geom": "POINT(0 0)", "distance": 5}, "logic": "or"} + assert json.loads(h["X-VectorFilter-emb"])["op"] == "cosine_within" + + +def test_vector_search(): + h = build_headers({"vector_search": {"column": "emb", "vector": [0.1, 0.2], "metric": "cosine", "as": "dist", "direction": "desc"}}) + assert h["X-Vector-Search-emb"] == "cosine" + assert h["X-Vector-Search-Vector"] == "[0.1,0.2]" + assert h["X-Vector-Search-As"] == "dist" + assert h["X-Vector-Search-Dir"] == "desc" + + +def test_xfiles_zip(): + xf = {"tablename": "users", "prefix": "USR", "limit": 10} + h = build_headers({"xfiles": xf}) + assert h["X-Files"].startswith("ZIP_") + assert json.loads(decode_header_value(h["X-Files"])) == xf + + +def test_columns_and_omit(): + assert build_headers({"columns": ["id", "name", "email"]})["X-Select-Fields"] == "id,name,email" + assert build_headers({"omit_columns": ["secret", "internal"]})["X-Not-Select-Fields"] == "secret,internal" + + +def test_filters(): + assert build_headers({"filters": [{"column": "status", "operator": "eq", "value": "active"}]})["X-FieldFilter-status"] == "active" + assert build_headers({"filters": [{"column": "age", "operator": "gte", "value": 18}]})["X-SearchOp-greaterthanorequal-age"] == "18" + assert build_headers({"filters": [{"column": "name", "operator": "contains", "value": "test", "logic_operator": "OR"}]})["X-SearchOr-contains-name"] == "test" + assert build_headers({"filters": [{"column": "price", "operator": "between", "value": [10, 100]}]})["X-SearchOp-between-price"] == "10,100" + assert build_headers({"filters": [{"column": "deleted_at", "operator": "is_null", "value": None}]})["X-SearchOp-empty-deleted_at"] == "" + assert build_headers({"filters": [{"column": "id", "operator": "in", "value": [1, 2, 3]}]})["X-SearchOp-in-id"] == "1,2,3" + assert build_headers({"filters": [{"column": "a", "operator": "eq", "value": True}]})["X-FieldFilter-a"] == "true" + + +def test_sort_pagination_cursor(): + h = build_headers({ + "sort": [{"column": "name", "direction": "asc"}, {"column": "created_at", "direction": "DESC"}], + "limit": 25, "offset": 0, "cursor_forward": "abc", "cursor_backward": "xyz", + }) + assert h["X-Sort"] == "+name,-created_at" + assert h["X-Limit"] == "25" and h["X-Offset"] == "0" + assert h["X-Cursor-Forward"] == "abc" and h["X-Cursor-Backward"] == "xyz" + + +def test_preload_basic_rownumber_computed_custom(): + h = build_headers({ + "preload": [{"relation": "Items", "columns": ["id", "name"]}, {"relation": "Category"}], + "fetch_row_number": "42", + "computedColumns": [{"name": "total", "expression": "price * qty"}], + "customOperators": [{"name": "a", "sql": "status = 'active'"}, {"name": "v", "sql": "verified = true"}], + }) + assert h["X-Preload"] == "Items:id,name|Category" + assert h["X-Fetch-RowNumber"] == "42" + assert h["X-CQL-SEL-total"] == "price * qty" + assert h["X-Custom-SQL-W"] == "status = 'active' AND verified = true" + + +def test_empty_options(): + assert build_headers({}) == {} + + +# ---- encode / decode ---- + +def test_roundtrip(): + for s in ("some complex value with spaces & symbols!", "café ☕ 你好"): + enc = encode_header_value(s) + assert enc.startswith("ZIP_") + assert decode_header_value(enc) == s + + +def test_decode_double_underscore_and_plain(): + assert decode_header_value("__" + base64.b64encode(b"hello").decode()) == "hello" + assert decode_header_value("__" + base64.b64encode("café ☕".encode()).decode()) == "café ☕" + assert decode_header_value("plain") == "plain" + + +def test_decode_nested(): + assert decode_header_value(encode_header_value(encode_header_value("x"))) == "x" + + +# ---- client ---- + +def make(handler, cls=HeaderSpecClient, **kw): + return cls(**{**CFG, **kw}, transport=httpx.MockTransport(handler)) + + +def test_read_sends_get_with_headers(): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(200, json=[{"id": 1}], headers={"content-range": "0-9/100", "x-limit": "10"}) + + with make(handler) as c: + res = c.read("public", "users", options={"columns": ["id", "name"], "limit": 10}) + r = seen[0] + assert str(r.url) == "http://localhost:3000/public/users" + assert r.method == "GET" + assert r.headers["x-select-fields"] == "id,name" + assert r.headers["x-limit"] == "10" + assert r.headers["authorization"] == "Bearer tok" + assert res["success"] is True + assert res["data"] == [{"id": 1}] + assert res["metadata"] == {"count": 100, "total": 100, "filtered": 100, "offset": 0, "limit": 10} + + +def test_metadata_defaults_without_content_range(): + with make(lambda r: httpx.Response(200, json=[])) as c: + assert c.read("public", "users")["metadata"]["total"] == 0 + + +def test_read_with_id_create_update_delete(): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(200, json={}) + + with make(handler) as c: + c.read("public", "users", "42") + c.create("public", "users", {"name": "Test"}) + c.update("public", "users", "1", {"name": "Updated"}, {"filters": [{"column": "active", "operator": "eq", "value": True}]}) + c.delete("public", "users", "1") + assert str(seen[0].url) == "http://localhost:3000/public/users/42" + assert seen[1].method == "POST" and json.loads(seen[1].content) == {"name": "Test"} + assert seen[2].method == "PUT" and str(seen[2].url).endswith("/public/users/1") + assert seen[2].headers["x-fieldfilter-active"] == "true" + assert seen[3].method == "DELETE" + + +def test_error_response(): + with make(lambda r: httpx.Response(400, json={"error": {"code": "err", "message": "fail"}})) as c: + with pytest.raises(ResolveSpecError, match="fail") as ei: + c.read("public", "users") + assert ei.value.status_code == 400 and ei.value.code == "err" + + +def test_error_non_json(): + with make(lambda r: httpx.Response(502, text="bad gateway")) as c: + with pytest.raises(ResolveSpecError, match="bad gateway") as ei: + c.read("public", "users") + assert ei.value.status_code == 502 + + +async def test_async_client(): + async def handler(req): + return httpx.Response(200, json=[{"id": 1}]) + + async with AsyncHeaderSpecClient(**CFG, transport=httpx.MockTransport(handler)) as c: + res = await c.read("public", "users", options={"limit": 1}) + assert res["data"] == [{"id": 1}] + + +def test_singleton(): + a = get_headerspec_client("http://hs-singleton:3000") + assert a is get_headerspec_client("http://hs-singleton:3000") + assert a is not get_headerspec_client("http://hs-singleton-b:3000") diff --git a/resolvespec-python/tests/test_resolvespec.py b/resolvespec-python/tests/test_resolvespec.py new file mode 100644 index 0000000..90686ab --- /dev/null +++ b/resolvespec-python/tests/test_resolvespec.py @@ -0,0 +1,210 @@ +import json + +import httpx +import pytest + +from resolvespec import ( + AsyncHeaderSpecClient, + AsyncResolveSpecClient, + HeaderSpecClient, + ResolveSpecClient, + ResolveSpecError, + get_headerspec_client, + get_resolvespec_client, +) + +CFG = dict(base_url="http://localhost:3000", token="test-token") + + +def make(handler, **kw): + return ResolveSpecClient(**{**CFG, **kw}, transport=httpx.MockTransport(handler)) + + +def ok(_req): + return httpx.Response(200, json={"success": True, "data": [{"id": 1}]}) + + +def capture(): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(200, json={"success": True, "data": {"id": 1, "name": "Test"}}) + + return seen, handler + + +def body(req): + return json.loads(req.content) + + +def test_read_with_numeric_id(): + seen, h = capture() + with make(h) as c: + assert c.read("public", "users", 1)["success"] is True + r = seen[0] + assert str(r.url) == "http://localhost:3000/public/users/1" + assert r.method == "POST" + assert r.headers["authorization"] == "Bearer test-token" + assert r.headers["content-type"] == "application/json" + assert body(r) == {"operation": "read"} + + +def test_read_array_id_goes_in_body(): + seen, h = capture() + with make(h) as c: + c.read("public", "users", ["1", "2"]) + assert str(seen[0].url) == "http://localhost:3000/public/users" + assert body(seen[0])["id"] == ["1", "2"] + + +def test_read_options_passthrough(): + seen, h = capture() + opts = { + "columns": ["id", "name"], "omit_columns": ["secret"], + "filters": [{"column": "active", "operator": "eq", "value": True}], + "sort": [{"column": "name", "direction": "asc"}], + "limit": 10, "offset": 0, "cursor_forward": "cursor1", "fetch_row_number": "5", + "customOperators": [{"name": "x", "sql": "a = 1"}], + } + with make(h) as c: + c.read("public", "users", options=opts) + assert body(seen[0])["options"] == opts + + +def test_create(): + seen, h = capture() + with make(h) as c: + res = c.create("public", "users", {"name": "Test"}) + assert res["data"]["name"] == "Test" + assert body(seen[0]) == {"operation": "create", "data": {"name": "Test"}} + + +def test_create_batch(): + seen, h = capture() + with make(h) as c: + c.create("public", "users", [{"a": 1}, {"a": 2}]) + assert body(seen[0])["data"] == [{"a": 1}, {"a": 2}] + + +def test_update_with_id_in_url_and_array(): + seen, h = capture() + with make(h) as c: + c.update("public", "users", {"name": "X"}, 5) + c.update("public", "users", {"name": "X"}, ["1", "2"]) + assert str(seen[0].url).endswith("/public/users/5") + assert body(seen[0]) == {"operation": "update", "data": {"name": "X"}} + assert str(seen[1].url).endswith("/public/users") + assert body(seen[1])["id"] == ["1", "2"] + + +def test_update_preserves_empty_string_and_null(): + seen, h = capture() + with make(h) as c: + c.update("public", "users", {"a": "", "b": None}, 1) + assert body(seen[0])["data"] == {"a": "", "b": None} + + +def test_delete(): + seen, h = capture() + with make(h) as c: + c.delete("public", "users", 1) + assert str(seen[0].url).endswith("/public/users/1") + assert body(seen[0]) == {"operation": "delete"} + + +def test_get_metadata(): + seen, h = capture() + with make(h) as c: + c.get_metadata("public", "users") + assert seen[0].method == "GET" + assert str(seen[0].url) == "http://localhost:3000/public/users" + assert not seen[0].content + + +def test_error_uses_server_message(): + with make(lambda r: httpx.Response(404, json={"success": False, "error": {"code": "not_found", "message": "nope"}})) as c: + with pytest.raises(ResolveSpecError, match="nope") as ei: + c.read("public", "users", 1) + assert ei.value.status_code == 404 and ei.value.code == "not_found" + + +def test_id_is_url_quoted(): + seen, h = capture() + with make(h) as c: + c.read("public", "users", "a/b") + assert str(seen[0].url).endswith("/public/users/a%2Fb") + + +def test_trailing_slash_base_url(): + seen, h = capture() + with make(h, base_url="http://localhost:3000/") as c: + c.read("public", "users") + assert str(seen[0].url) == "http://localhost:3000/public/users" + + +async def test_async_client(): + async def handler(req): + return httpx.Response(200, json={"success": True, "data": [1]}) + + async with AsyncResolveSpecClient(**CFG, transport=httpx.MockTransport(handler)) as c: + assert (await c.read("public", "users"))["data"] == [1] + assert (await c.create("public", "users", {}))["success"] + assert (await c.update("public", "users", {}, 1))["success"] + assert (await c.delete("public", "users", 1))["success"] + assert (await c.get_metadata("public", "users"))["success"] + + +# ---- custom headers (ported from custom-headers.test.ts) ---- + +@pytest.mark.parametrize("cls", [ResolveSpecClient, HeaderSpecClient]) +def test_custom_headers_on_every_op_case_insensitive(cls): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(200, json={"success": True, "data": []}) + + headers = {"X-Tenant": "acme", "authorization": "Basic ignored", + "content-type": "application/custom+json", "x-limit": "99"} + with cls("http://localhost:3000", "tok", headers, transport=httpx.MockTransport(handler)) as c: + c.read("public", "users", options={"limit": 10}) + c.create("public", "users", {}) + if cls is ResolveSpecClient: + c.update("public", "users", {}, "1") + c.get_metadata("public", "users") + else: + c.update("public", "users", "1", {}) + c.delete("public", "users", "1") + for r in seen: + assert r.headers["x-tenant"] == "acme" + assert r.headers["authorization"] == "Bearer tok" + assert r.headers["content-type"] == "application/custom+json" + if cls is HeaderSpecClient: + assert seen[0].headers["x-limit"] == "10" + assert headers["authorization"] == "Basic ignored" + assert headers["x-limit"] == "99" + + +@pytest.mark.parametrize("cls", [ResolveSpecClient, HeaderSpecClient]) +def test_custom_auth_without_token(cls): + seen = [] + + def handler(req): + seen.append(req) + return httpx.Response(200, json={"success": True, "data": []}) + + with cls("http://localhost:3000", headers={"Authorization": "Basic custom"}, transport=httpx.MockTransport(handler)) as c: + c.read("public", "users") + assert seen[0].headers["authorization"] == "Basic custom" + + +@pytest.mark.parametrize("factory", [get_resolvespec_client, get_headerspec_client]) +def test_cache_isolation_and_snapshot(factory): + headers = {"X-Tenant": "acme", "X-App": "grid"} + first = factory("http://tenant-cache", "one", headers) + assert factory("http://tenant-cache", "one", {"x-app": "grid", "x-tenant": "acme"}) is first + assert factory("http://tenant-cache", "two", headers) is not first + headers["X-Tenant"] = "other" + assert factory("http://tenant-cache", "one", headers) is not first + assert first.headers["X-Tenant"] == "acme" diff --git a/resolvespec-python/tests/test_websocket.py b/resolvespec-python/tests/test_websocket.py new file mode 100644 index 0000000..0c4f350 --- /dev/null +++ b/resolvespec-python/tests/test_websocket.py @@ -0,0 +1,152 @@ +import asyncio +import json + +import pytest +from websockets.asyncio.server import serve + +from resolvespec import ResolveSpecError, WebSocketClient + + +class Server: + """Minimal in-process WebSocketSpec server.""" + + def __init__(self): + self.received = [] + self.conns = set() + self.respond = True + + async def handler(self, ws): + self.conns.add(ws) + try: + async for raw in ws: + msg = json.loads(raw) + self.received.append(msg) + if msg["type"] == "ping": + await ws.send(json.dumps({"type": "pong"})) + continue + if not self.respond: + continue + await ws.send(json.dumps(self.reply(msg))) + finally: + self.conns.discard(ws) + + def reply(self, msg): + base = {"id": msg["id"], "type": "response", "success": True, "timestamp": "t"} + if msg["type"] == "subscription" and msg["operation"] == "subscribe": + return {**base, "data": {"subscription_id": "sub-1"}} + if msg.get("entity") == "fail": + return {**base, "success": False, "error": {"code": "bad", "message": "boom"}} + return {**base, "data": {"echo": msg.get("operation"), "record_id": msg.get("record_id")}} + + +@pytest.fixture +async def server(): + s = Server() + async with serve(s.handler, "127.0.0.1", 0) as srv: + s.url = "ws://127.0.0.1:%d" % srv.sockets[0].getsockname()[1] + yield s + + +async def test_operations_and_message_shape(server): + async with WebSocketClient(server.url, reconnect=False) as c: + assert c.state == "connected" + assert await c.read("users", schema="public", record_id="1", limit=5, filters=[{"column": "a", "operator": "eq", "value": 1}]) == {"echo": "read", "record_id": "1"} + await c.create("users", {"n": 1}, schema="public") + await c.update("users", "2", {"n": 2}) + await c.delete("users", "3") + await c.meta("users") + m = server.received + assert m[0]["type"] == "request" and m[0]["operation"] == "read" + assert m[0]["schema"] == "public" and m[0]["record_id"] == "1" + assert m[0]["options"] == {"filters": [{"column": "a", "operator": "eq", "value": 1}], "limit": 5} + assert m[1]["data"] == {"n": 1} + assert m[2]["record_id"] == "2" + assert [x["operation"] for x in m] == ["read", "create", "update", "delete", "meta"] + assert "schema" not in m[2] + assert len({x["id"] for x in m}) == 5 + + +async def test_error_response_raises(server): + async with WebSocketClient(server.url, reconnect=False) as c: + with pytest.raises(ResolveSpecError, match="boom") as ei: + await c.read("fail") + assert ei.value.code == "bad" + + +async def test_request_timeout(server): + server.respond = False + async with WebSocketClient(server.url, reconnect=False, request_timeout=0.1) as c: + with pytest.raises(ResolveSpecError, match="timeout"): + await c.read("users") + assert not c._pending + + +async def test_not_connected_raises(): + c = WebSocketClient("ws://127.0.0.1:1") + with pytest.raises(ResolveSpecError, match="not connected"): + await c.read("users") + + +async def test_subscribe_notify_unsubscribe(server): + got = asyncio.Queue() + async with WebSocketClient(server.url, reconnect=False) as c: + sid = await c.subscribe("users", got.put, schema="public", filters=[{"column": "a", "operator": "eq", "value": 1}]) + assert sid == "sub-1" + assert [s.id for s in c.get_subscriptions()] == ["sub-1"] + assert server.received[0]["operation"] == "subscribe" + assert server.received[0]["options"] == {"filters": [{"column": "a", "operator": "eq", "value": 1}]} + for ws in server.conns: + await ws.send(json.dumps({"type": "notification", "operation": "create", "subscription_id": "sub-1", + "entity": "users", "data": {"id": 9}, "timestamp": "t"})) + n = await asyncio.wait_for(got.get(), 2) + assert n["data"] == {"id": 9} + await c.unsubscribe("sub-1") + assert c.get_subscriptions() == [] + assert server.received[-1] == {**server.received[-1], "operation": "unsubscribe", "subscription_id": "sub-1"} + + +async def test_events_and_heartbeat(server): + events = [] + c = WebSocketClient(server.url, reconnect=False, heartbeat_interval=0.05) + c.on("connect", lambda: events.append("connect")) + c.on("state_change", lambda s: events.append(s)) + c.on("message", lambda m: events.append(("msg", m["type"]))) + await c.connect() + await asyncio.sleep(0.2) + await c.close() + assert events[:3] == ["connecting", "connected", "connect"] + assert ("msg", "pong") in events + assert events[-1] == "disconnected" + assert any(m["type"] == "ping" for m in server.received) + with pytest.raises(ValueError): + c.on("bogus", lambda: None) + + +async def test_reconnect_after_server_drop(server): + states = [] + c = WebSocketClient(server.url, reconnect=True, reconnect_interval=0.05) + c.on("state_change", states.append) + await c.connect() + for ws in list(server.conns): + await ws.close() + for _ in range(100): + if states.count("connected") >= 2: + break + await asyncio.sleep(0.05) + assert "reconnecting" in states + assert c.is_connected() + assert (await c.read("users"))["echo"] == "read" + await c.close() + + +async def test_pending_requests_fail_on_disconnect(server): + server.respond = False + c = WebSocketClient(server.url, reconnect=False) + await c.connect() + task = asyncio.create_task(c.read("users")) + await asyncio.sleep(0.05) + for ws in list(server.conns): + await ws.close() + with pytest.raises(ResolveSpecError, match="disconnected"): + await asyncio.wait_for(task, 2) + await c.close() diff --git a/resolvespec-python/todo.md b/resolvespec-python/todo.md index f6f8d3e..e44b1de 100644 --- a/resolvespec-python/todo.md +++ b/resolvespec-python/todo.md @@ -4,34 +4,34 @@ ### 1. ResolveSpec Client API -- [ ] Core API implementation (read, create, update, delete, get_metadata) -- [ ] Unit tests for API functions +- [x] Core API implementation (read, create, update, delete, get_metadata) +- [x] Unit tests for API functions - [ ] Integration tests with server -- [ ] Error handling and edge cases +- [x] Error handling and edge cases ### 2. HeaderSpec Client API -- [ ] Client API implementation -- [ ] Unit tests +- [x] Client API implementation +- [x] Unit tests - [ ] Integration tests with server ### 3. FunctionSpec Client API -- [ ] Client API implementation -- [ ] Unit tests +- [x] Client API implementation +- [x] Unit tests - [ ] Integration tests with server ### 4. WebSocketSpec Client API -- [ ] WebSocketClient class implementation (read, create, update, delete, meta, subscribe, unsubscribe) -- [ ] Unit tests for WebSocketClient -- [ ] Connection handling tests -- [ ] Subscription tests +- [x] WebSocketClient class implementation (read, create, update, delete, meta, subscribe, unsubscribe) +- [x] Unit tests for WebSocketClient +- [x] Connection handling tests +- [x] Subscription tests - [ ] Integration tests with server ### 5. Testing Infrastructure -- [ ] Set up test framework (pytest) +- [x] Set up test framework (pytest) - [ ] Configure test coverage reporting (pytest-cov) - [ ] Add test utilities and fixtures - [ ] Create test documentation @@ -43,8 +43,8 @@ - [ ] Usage examples for each client API - [ ] Installation guide - [ ] Contributing guidelines -- [ ] README with quick start +- [x] README (cheatsheet) --- -**Last Updated:** 2026-02-07 \ No newline at end of file +**Last Updated:** 2026-09-30 \ No newline at end of file