""" TradeFacts.io Python SDK Official client for the TradeFacts US HTS, Canadian Customs Tariff, and Mexico TIGIE API. https://tradefacts.io | https://docs.tradefacts.io """ import hmac import hashlib from typing import Optional try: import requests except ImportError: raise ImportError("TradeFacts SDK requires the 'requests' library: pip install requests") __version__ = "0.1.0" BASE_URL = "https://tradefacts.io" class TradeFacts: """ TradeFacts API client. Usage: from tradefacts import TradeFacts client = TradeFacts("your-api-key") # US HTS lookup record = client.hts("0901.11.00") # Canadian tariff lookup record = client.ca("0901.11.00") # Search (US or CA) results = client.search("green coffee beans") results = client.ca_search("green coffee beans") # Change logs changes = client.changes() ca_changes = client.ca_changes() """ def __init__(self, api_key: str, base_url: str = BASE_URL, timeout: int = 10): if not api_key: raise ValueError("api_key is required") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self._session = requests.Session() self._session.headers.update({ "X-API-Key": self.api_key, "User-Agent": f"tradefacts-python/{__version__}", }) def _get(self, path: str, params: Optional[dict] = None) -> dict: url = f"{self.base_url}{path}" resp = self._session.get(url, params=params, timeout=self.timeout) if resp.status_code == 401: raise AuthError("Invalid or expired API key") if resp.status_code == 403: raise PlanError(resp.json().get("detail", "Access denied — check your plan tier")) if resp.status_code == 404: raise NotFoundError(f"No record found at {path}") resp.raise_for_status() return resp.json() # ── US HTS ─────────────────────────────────────────────────────────────── def hts(self, code: str) -> dict: """Fetch a single US HTS record by tariff code.""" return self._get(f"/api/hts/{code}") def search(self, query: str) -> list: """Search US HTS records. Multi-word queries use AND logic.""" return self._get("/api/search", params={"q": query}) def chapter(self, number: int) -> list: """Fetch all US HTS records for a chapter (1–99).""" return self._get(f"/api/chapter/{number}") def changes(self) -> list: """Fetch the US HTS nightly change log. Available on all plans.""" return self._get("/api/changes") # ── Canada ─────────────────────────────────────────────────────────────── def ca(self, code: str) -> dict: """Fetch a single Canadian Customs Tariff record by tariff code. Requires Tier 2+.""" return self._get(f"/api/ca/{code}") def ca_search(self, query: str) -> list: """Search Canadian tariff records. Requires Tier 2+.""" return self._get("/api/ca/search", params={"q": query}) def ca_chapter(self, number: int) -> list: """Fetch all Canadian records for a chapter prefix. Requires Tier 2+.""" return self._get(f"/api/ca/chapter/{number}") def ca_ust(self) -> list: """Fetch all Canadian records with a UST preferential rate. Requires Tier 2+.""" return self._get("/api/ca/ust") def ca_changes(self) -> list: """Fetch the Canadian nightly change log. Requires Tier 2+.""" return self._get("/api/ca/changelog") # ── Mexico ─────────────────────────────────────────────────────────────── def mx(self, code: str) -> dict: """Fetch a single Mexico TIGIE fraccion. Accepts with or without dots. Requires Tier 2+.""" return self._get(f"/api/mx/{code}") def mx_search(self, query: str) -> list: """Search Mexico TIGIE records by keyword (Spanish descriptions). Requires Tier 2+.""" return self._get("/api/mx/search", params={"q": query}) def mx_chapter(self, number: int) -> list: """Fetch all Mexico TIGIE records for a chapter (1-98). Requires Tier 2+.""" return self._get(f"/api/mx/chapter/{number}") def mx_changes(self) -> list: """Fetch the Mexico TIGIE nightly change log. Requires Tier 2+.""" return self._get("/api/mx/changelog") # ── Webhooks ───────────────────────────────────────────────────────────── def register_webhook(self, url: str, secret: str) -> dict: """ Register or update a webhook. Requires Tier 3. URL must be HTTPS. Secret must be >= 16 characters. """ if not url.startswith("https://"): raise ValueError("Webhook URL must use HTTPS") if len(secret) < 16: raise ValueError("Webhook secret must be at least 16 characters") resp = self._session.post( f"{self.base_url}/api/webhook/register", json={"url": url, "secret": secret}, timeout=self.timeout, ) if resp.status_code == 403: raise PlanError("Webhooks require Tier 3 or above") resp.raise_for_status() return resp.json() def webhook_config(self) -> dict: """View current webhook configuration. Requires Tier 3.""" return self._get("/api/webhook/config") # ── IEEPA Overlay ──────────────────────────────────────────────────────── def ieepa(self, code: str) -> dict: """Fetch IEEPA provisional rate for a Chapter 99 code. Requires Tier 4.""" return self._get(f"/api/ieepa/{code}") def discrepancies(self) -> dict: """ Compare IEEPA overlay rates against the published HTS schedule. Returns rate_differs, footnote_gap, and hts_code_missing discrepancies. Requires Tier 4. """ return self._get("/api/discrepancies") # ── Webhook signature verification ─────────────────────────────────────────── def verify_webhook(payload_bytes: bytes, signature_header: str, secret: str) -> bool: """ Verify an incoming TradeFacts webhook signature. Args: payload_bytes: Raw request body bytes (do not decode) signature_header: Value of the X-TradeFacts-Signature header secret: The webhook secret you registered Returns: True if the signature is valid, False otherwise. Example (Flask): @app.route("/webhook", methods=["POST"]) def handle(): if not verify_webhook(request.data, request.headers["X-TradeFacts-Signature"], SECRET): abort(401) payload = request.get_json() ... """ if not signature_header.startswith("sha256="): return False expected = hmac.new(secret.encode(), payload_bytes, hashlib.sha256).hexdigest() received = signature_header[len("sha256="):] return hmac.compare_digest(expected, received) # ── Exceptions ──────────────────────────────────────────────────────────────── class TradeFacts_Error(Exception): """Base exception for TradeFacts SDK errors.""" class AuthError(TradeFacts_Error): """Raised on 401 — invalid or expired API key.""" class PlanError(TradeFacts_Error): """Raised on 403 — endpoint requires a higher plan tier.""" class NotFoundError(TradeFacts_Error): """Raised on 404 — tariff code not found."""