"""SafetyDataSheetAPI Python client. A single file with no third-party dependencies (Python 3.9+). Copy it into your project or download it from https://safetydatasheetapi.com/sdk/python/safetydatasheetapi.py from safetydatasheetapi import Client client = Client(api_key="YOUR_API_KEY") result = client.extract("acetone-sds.pdf", language_hint="en") print(result["confidence_score"], result["outputs"]["json"]["product_identification"]["product_name"]) API reference: https://safetydatasheetapi.com/api-docs/ OpenAPI contract: https://safetydatasheetapi.com/openapi.yaml """ from __future__ import annotations import hashlib import hmac import json import mimetypes import os import time import urllib.error import urllib.request import uuid from typing import Any, Dict, Iterable, List, Optional, Tuple, Union __version__ = "1.0.0" DEFAULT_BASE_URL = "https://safetydatasheetapi.com" FileInput = Union[str, "os.PathLike[str]", bytes, Any] class ApiError(Exception): """Raised for any non-2xx response. Branch on ``code``; ``message`` is for humans.""" def __init__( self, status: int, code: str, message: str, request_id: Optional[str] = None, retry_after: Optional[int] = None, body: Optional[Dict[str, Any]] = None, ) -> None: super().__init__(f"{status} {code}: {message}") self.status = status self.code = code self.message = message self.request_id = request_id self.retry_after = retry_after self.body = body class RateLimitError(ApiError): """HTTP 429. ``retry_after`` is the number of seconds to wait, from the Retry-After header.""" def verify_webhook_signature(secret: str, body: bytes, signature_header: str) -> bool: """Check an ``X-Webhook-Signature`` header against the raw request body. ``body`` must be the exact bytes received, before any JSON parsing. """ if not signature_header: return False expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature_header.strip()) def _read_file(file: FileInput, filename: Optional[str]) -> Tuple[str, bytes]: if isinstance(file, (bytes, bytearray)): return filename or "document.pdf", bytes(file) if hasattr(file, "read"): name = filename or os.path.basename(getattr(file, "name", "") or "") or "document.pdf" return name, file.read() path = os.fspath(file) with open(path, "rb") as handle: return filename or os.path.basename(path), handle.read() def _encode_multipart( fields: Dict[str, str], files: List[Tuple[str, str, bytes]] ) -> Tuple[bytes, str]: boundary = "----sdsapi" + uuid.uuid4().hex chunks: List[bytes] = [] for name, value in fields.items(): chunks.append( ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="{name}"\r\n\r\n' f"{value}\r\n" ).encode("utf-8") ) for field, filename, content in files: content_type = mimetypes.guess_type(filename)[0] or "application/pdf" safe_name = filename.replace('"', "") chunks.append( ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="{field}"; filename="{safe_name}"\r\n' f"Content-Type: {content_type}\r\n\r\n" ).encode("utf-8") + content + b"\r\n" ) chunks.append(f"--{boundary}--\r\n".encode("utf-8")) return b"".join(chunks), f"multipart/form-data; boundary={boundary}" class Client: """Client for the SafetyDataSheetAPI extraction endpoints. :param api_key: your API key, sent as ``Authorization: Bearer ``. :param base_url: override only for testing. :param timeout: per-request timeout in seconds. Single-document extraction can take several seconds on large PDFs; the default leaves generous headroom. """ def __init__(self, api_key: str, base_url: str = DEFAULT_BASE_URL, timeout: float = 120.0) -> None: if not api_key: raise ValueError("api_key is required") self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout # -- endpoints ----------------------------------------------------------------- def extract( self, file: FileInput, *, filename: Optional[str] = None, language_hint: Optional[str] = None, schema_version: Optional[str] = None, ) -> Dict[str, Any]: """POST /extract-sds. Returns the parsed response (see ``outputs.json`` for the record).""" name, content = _read_file(file, filename) fields = _optional_fields(language_hint=language_hint, schema_version=schema_version) return self._request("POST", "/extract-sds", fields=fields, files=[("file", name, content)]) def submit_bulk( self, files: Iterable[FileInput], *, callback_url: Optional[str] = None, language_hint: Optional[str] = None, schema_version: Optional[str] = None, idempotency_key: Optional[str] = None, ) -> Dict[str, Any]: """POST /extract-sds/bulk. Returns ``job_id`` and ``status_url``. Pass the same ``idempotency_key`` when retrying a submission: the existing job is returned (``idempotent_replay: true``) instead of creating a duplicate. """ payload = [("file",) + _read_file(item, None) for item in files] fields = _optional_fields( callback_url=callback_url, language_hint=language_hint, schema_version=schema_version ) headers = {"Idempotency-Key": idempotency_key} if idempotency_key else {} return self._request("POST", "/extract-sds/bulk", fields=fields, files=payload, headers=headers) def get_bulk_job(self, job_id: str) -> Dict[str, Any]: """GET /extract-sds/bulk/{jobId}.""" return self._request("GET", f"/extract-sds/bulk/{job_id}") def wait_for_bulk_job( self, job_id: str, *, poll_interval: float = 3.0, timeout: float = 600.0 ) -> Dict[str, Any]: """Poll until the job is ``completed`` or ``failed``. Raises ``TimeoutError`` after ``timeout`` seconds.""" deadline = time.monotonic() + timeout while True: job = self.get_bulk_job(job_id) if job.get("status") in ("completed", "failed"): return job if time.monotonic() >= deadline: raise TimeoutError(f"bulk job {job_id} still {job.get('status')} after {timeout}s") time.sleep(poll_interval) # -- plumbing ------------------------------------------------------------------ def _request( self, method: str, path: str, *, fields: Optional[Dict[str, str]] = None, files: Optional[List[Tuple[str, str, bytes]]] = None, headers: Optional[Dict[str, str]] = None, ) -> Dict[str, Any]: request_headers = { "Authorization": f"Bearer {self.api_key}", "Accept": "application/json", "User-Agent": f"safetydatasheetapi-python/{__version__}", } request_headers.update(headers or {}) data = None if files is not None: data, content_type = _encode_multipart(fields or {}, files) request_headers["Content-Type"] = content_type request = urllib.request.Request( self.base_url + path, data=data, headers=request_headers, method=method ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as error: raise _to_api_error(error) from None def _optional_fields(**values: Optional[str]) -> Dict[str, str]: return {key: value for key, value in values.items() if value} def _to_api_error(error: "urllib.error.HTTPError") -> ApiError: raw = error.read().decode("utf-8", errors="replace") retry_header = error.headers.get("Retry-After") retry_after = int(retry_header) if retry_header and retry_header.isdigit() else None try: body = json.loads(raw) detail = body.get("error") or {} code = detail.get("code", "http_error") message = detail.get("message", f"HTTP {error.code}") request_id = body.get("request_id") except (ValueError, AttributeError): body, code, message, request_id = None, "http_error", f"HTTP {error.code}", None cls = RateLimitError if error.code == 429 else ApiError return cls(error.code, code, message, request_id=request_id, retry_after=retry_after, body=body)