"""NocoDB API-only insert adapter with bounded final duplicate checks.""" from __future__ import annotations import http.client import json import ssl import time import urllib.parse from dataclasses import dataclass, field from typing import Mapping, Protocol, Sequence, runtime_checkable from ..errors import ApplicationError, ContractViolation, InvalidConfiguration from ..writer.models import InsertCandidate _MAX_RESPONSE_BYTES = 2 * 1024 * 1024 class NocoApiError(ApplicationError): """Raised when the NocoDB API rejects or returns malformed data.""" def __init__(self, message: str, *, status: int | None = None) -> None: super().__init__(message) self.status = status @runtime_checkable class JsonTransport(Protocol): """Small injectable JSON transport used by the API repository.""" def request( self, *, method: str, url: str, headers: Mapping[str, str], body: Mapping[str, object] | None, timeout_seconds: float, ) -> object: """Issue one JSON request and return a detached decoded value.""" class HttpsJsonTransport: """Direct HTTPS JSON transport with no redirects or environment proxies.""" def request( self, *, method: str, url: str, headers: Mapping[str, str], body: Mapping[str, object] | None, timeout_seconds: float, ) -> object: parsed = urllib.parse.urlsplit(url) if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: raise InvalidConfiguration("Noco transport requires a credential-free HTTPS URL") if method not in {"GET", "POST"}: raise InvalidConfiguration("Noco transport supports only GET and POST") payload = None if body is None else json.dumps(body, separators=(",", ":")).encode() request_headers = dict(headers) request_headers["Accept"] = "application/json" if payload is not None: request_headers["Content-Type"] = "application/json" path = urllib.parse.urlunsplit(("", "", parsed.path or "/", parsed.query, "")) connection = http.client.HTTPSConnection( parsed.hostname, port=parsed.port, timeout=timeout_seconds, context=ssl.create_default_context(), ) try: connection.request(method, path, body=payload, headers=request_headers) response = connection.getresponse() raw = response.read(_MAX_RESPONSE_BYTES + 1) if len(raw) > _MAX_RESPONSE_BYTES: raise NocoApiError( "Noco response exceeded the bounded size", status=response.status ) if not 200 <= response.status < 300: raise NocoApiError( f"Noco API returned HTTP {response.status}", status=response.status, ) if not raw: return {} try: return json.loads(raw) except json.JSONDecodeError as exc: raise NocoApiError( "Noco API returned invalid JSON", status=response.status ) from exc finally: connection.close() @dataclass(frozen=True, slots=True) class NocoInsertConfig: """Secret-bearing API configuration; token is excluded from repr.""" base_url: str table_id: str api_token: str = field(repr=False) timeout_seconds: float = 30.0 get_attempts: int = 3 def __post_init__(self) -> None: parsed = urllib.parse.urlsplit(self.base_url) if parsed.scheme != "https" or not parsed.hostname or parsed.query or parsed.fragment: raise InvalidConfiguration("Noco base_url must be a clean HTTPS origin") if parsed.username or parsed.password: raise InvalidConfiguration("Noco base_url cannot contain credentials") if not self.table_id.strip() or len(self.table_id) > 128: raise InvalidConfiguration("Noco table_id must be nonblank and bounded") if not self.api_token.strip(): raise InvalidConfiguration("Noco api_token must be nonblank") if not 1 <= self.timeout_seconds <= 120: raise InvalidConfiguration("Noco timeout_seconds is outside the supported range") if isinstance(self.get_attempts, bool) or not 1 <= self.get_attempts <= 5: raise InvalidConfiguration("Noco get_attempts is outside the supported range") object.__setattr__(self, "base_url", self.base_url.rstrip("/")) object.__setattr__(self, "table_id", self.table_id.strip()) @dataclass(frozen=True, slots=True) class DuplicateRecord: """Bounded duplicate evidence returned by the final pre-insert lookup.""" record_id: int match_paths: tuple[str, ...] class NocoApiInsertRepository: """Perform final duplicate reads and one-record API inserts only.""" def __init__(self, *, config: NocoInsertConfig, transport: JsonTransport) -> None: if not isinstance(config, NocoInsertConfig): raise ContractViolation("config must be NocoInsertConfig") if not isinstance(transport, JsonTransport): raise ContractViolation("transport must implement JsonTransport") self._config = config self._transport = transport @property def table_id(self) -> str: return self._config.table_id def _headers(self) -> Mapping[str, str]: return {"xc-token": self._config.api_token} def _records_url(self, query: Mapping[str, str] | None = None) -> str: path = f"/api/v2/tables/{urllib.parse.quote(self.table_id, safe='')}/records" if query: return f"{self._config.base_url}{path}?{urllib.parse.urlencode(query)}" return f"{self._config.base_url}{path}" @staticmethod def _where_value(value: str) -> str: cleaned = value.strip() if not cleaned: raise ContractViolation("duplicate lookup value must be nonblank") if len(cleaned) > 2048: raise ContractViolation("duplicate lookup value exceeds the bounded length") return '"' + cleaned.replace("\\", "\\\\").replace('"', '\\"') + '"' def _get(self, where: str) -> Sequence[Mapping[str, object]]: last_error: NocoApiError | None = None for attempt in range(self._config.get_attempts): try: response = self._transport.request( method="GET", url=self._records_url( { "fields": "Id,source,osm_type,osm_id,osm_url,name,website,phone,address", "limit": "20", "where": where, } ), headers=self._headers(), body=None, timeout_seconds=self._config.timeout_seconds, ) return self._extract_records(response) except NocoApiError as exc: last_error = exc retryable = exc.status == 429 or (exc.status is not None and exc.status >= 500) if not retryable or attempt + 1 >= self._config.get_attempts: raise time.sleep(0.25 * (2**attempt)) if last_error is None: raise NocoApiError("Noco GET failed without an error") raise last_error @staticmethod def _extract_records(response: object) -> Sequence[Mapping[str, object]]: raw_records: object if isinstance(response, list): raw_records = response elif isinstance(response, dict): raw_records = response.get("list", response.get("records")) else: raise NocoApiError("Noco lookup returned an unsupported JSON shape") if not isinstance(raw_records, list): raise NocoApiError("Noco lookup did not return a record list") if len(raw_records) > 20: raise NocoApiError("Noco lookup exceeded the bounded result count") records: list[Mapping[str, object]] = [] for item in raw_records: if not isinstance(item, dict) or any(not isinstance(key, str) for key in item): raise NocoApiError("Noco lookup returned a malformed record") records.append(item) return records @staticmethod def _record_id(record: Mapping[str, object]) -> int: value = record.get("Id", record.get("id")) if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise NocoApiError("Noco record is missing a positive Id") return value def find_duplicates( self, candidate: InsertCandidate, *, include_website: bool = True, ) -> tuple[DuplicateRecord, ...]: if not isinstance(candidate, InsertCandidate): raise ContractViolation("candidate must be InsertCandidate") source = self._where_value("OSM") osm_type = self._where_value(candidate.osm_type) osm_id = self._where_value(str(candidate.osm_id)) lookups: list[tuple[str, str]] = [ ( "osm_identity", f"(source,eq,{source})~and(osm_type,eq,{osm_type})~and(osm_id,eq,{osm_id})", ), ("osm_url", f"(osm_url,eq,{self._where_value(candidate.osm_url)})"), ] name = self._where_value(candidate.name) if include_website and candidate.website is not None: website = self._where_value(candidate.website) lookups.append(("website_name", f"(website,eq,{website})~and(name,eq,{name})")) if candidate.phone is not None: phone = self._where_value(candidate.phone) lookups.append(("phone_name", f"(phone,eq,{phone})~and(name,eq,{name})")) if candidate.address is not None: address = self._where_value(candidate.address) lookups.append(("address_name", f"(address,eq,{address})~and(name,eq,{name})")) matches: dict[int, set[str]] = {} for match_path, where in lookups: for record in self._get(where): matches.setdefault(self._record_id(record), set()).add(match_path) if len(matches) > 20: raise NocoApiError("combined duplicate evidence exceeded the bounded result count") return tuple( DuplicateRecord(record_id, tuple(sorted(paths))) for record_id, paths in sorted(matches.items()) ) def create(self, payload: Mapping[str, object]) -> int: if not isinstance(payload, Mapping) or not payload: raise ContractViolation("insert payload must be a nonempty mapping") response = self._transport.request( method="POST", url=self._records_url(), headers=self._headers(), body=dict(payload), timeout_seconds=self._config.timeout_seconds, ) if isinstance(response, list): if len(response) != 1 or not isinstance(response[0], dict): raise NocoApiError("Noco insert returned an unsupported list response") record = response[0] elif isinstance(response, dict): record = response else: raise NocoApiError("Noco insert returned an unsupported JSON shape") return self._record_id(record) __all__ = [ "DuplicateRecord", "HttpsJsonTransport", "JsonTransport", "NocoApiError", "NocoApiInsertRepository", "NocoInsertConfig", ]