"""Zero-dependency, server-side Python client for TotemClock Company API v1. Keep the ``tc_live_...`` secret in an environment variable or secret manager. The API key selects exactly one company, so this client never accepts an ``organizationId`` tenant selector. Example: import os from totemclock_company_api import TotemClockCompanyClient client = TotemClockCompanyClient(os.environ["TOTEMCLOCK_API_KEY"]) print(client.organization()) """ from __future__ import annotations import json import re from typing import Any, Callable, Mapping, Optional from urllib.error import HTTPError from urllib.parse import quote, urlencode, urlsplit, urlunsplit from urllib.request import Request, urlopen DEFAULT_BASE_URL = "https://totemclock.com" API_KEY_PATTERN = re.compile(r"^tc_live_[A-Za-z0-9_-]{8,}_[A-Za-z0-9_-]{32,}$") Opener = Callable[..., Any] class TotemClockCompanyApiError(RuntimeError): """An HTTP error returned by the TotemClock Company API.""" def __init__(self, status: int, body: Any) -> None: message = body.get("message") if isinstance(body, Mapping) else None super().__init__( message if isinstance(message, str) else f"TotemClock Company API request failed with HTTP {status}" ) self.status = status self.body = body def _normalize_base_url(value: str) -> str: parsed = urlsplit(str(value or DEFAULT_BASE_URL)) local_http = parsed.scheme == "http" and parsed.hostname in { "localhost", "127.0.0.1", "::1", } if parsed.scheme != "https" and not local_http: raise ValueError( "base_url must use HTTPS (HTTP is allowed only for local development)" ) if parsed.username or parsed.password or parsed.query or parsed.fragment: raise ValueError( "base_url cannot contain credentials, query parameters or a fragment" ) return urlunsplit((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "")) def _assert_no_organization_id(value: Any, seen: Optional[set[int]] = None) -> None: if not isinstance(value, (Mapping, list, tuple)): return visited = seen if seen is not None else set() identity = id(value) if identity in visited: return visited.add(identity) if isinstance(value, Mapping): if "organizationId" in value: raise ValueError( "organizationId is not accepted; the API key selects the company" ) for child in value.values(): _assert_no_organization_id(child, visited) else: for child in value: _assert_no_organization_id(child, visited) class TotemClockCompanyClient: """Synchronous Company API v1 client using only the Python standard library.""" def __init__( self, api_key: str, base_url: str = DEFAULT_BASE_URL, *, timeout: float = 10, opener: Optional[Opener] = None, ) -> None: normalized_key = str(api_key or "").strip() if not API_KEY_PATTERN.fullmatch(normalized_key): raise ValueError("A valid tc_live_... Company API key is required") if timeout <= 0: raise ValueError("timeout must be greater than zero") self._api_key = normalized_key self.base_url = _normalize_base_url(base_url) self.timeout = timeout self._opener = opener or urlopen @staticmethod def _decode_body(response: Any) -> Any: raw = response.read() if not raw: return None try: return json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return raw.decode("utf-8", errors="replace") def _request( self, method: str, path: str, *, parameters: Optional[Mapping[str, Any]] = None, body: Optional[Mapping[str, Any]] = None, accept: str = "application/json", ) -> Any: _assert_no_organization_id(parameters) _assert_no_organization_id(body) query = urlencode({ key: value for key, value in (parameters or {}).items() if value is not None and value != "" }) url = f"{self.base_url}{path}" if query: url = f"{url}?{query}" data = None if body is None else json.dumps( body, separators=(",", ":") ).encode("utf-8") headers = {"Accept": accept, "X-API-Key": self._api_key} if data is not None: headers["Content-Type"] = "application/json" request = Request(url, data=data, headers=headers, method=method) try: with self._opener(request, timeout=self.timeout) as response: return self._decode_body(response) except HTTPError as error: response_body = self._decode_body(error) raise TotemClockCompanyApiError(error.code, response_body) from error def organization(self) -> Any: """Return the company identity selected by the API key.""" return self._request("GET", "/api/v1/organization") def employees(self) -> Any: """List active operational employee records.""" return self._request("GET", "/api/v1/employees") def sites(self) -> Any: """List active company sites.""" return self._request("GET", "/api/v1/sites") def schedules(self) -> Any: """List active schedules and their employee assignments.""" return self._request("GET", "/api/v1/schedules") def punches( self, *, from_: str, to: str, employee_id: Optional[str] = None, limit: Optional[int] = None, ) -> Any: """List punches in a required bounded ISO 8601 date range.""" return self._request("GET", "/api/v1/punches", parameters={ "from": from_, "to": to, "employeeId": employee_id, "limit": limit, }) def create_punch( self, *, external_id: str, employee_id: str, type: str, occurred_at: str, site_id: Optional[str] = None, note: Optional[str] = None, ) -> Any: """Create or safely retry a company-scoped external punch.""" body = { "externalId": external_id, "employeeId": employee_id, "type": type, "occurredAt": occurred_at, } if site_id is not None: body["siteId"] = site_id if note is not None: body["note"] = note return self._request("POST", "/api/v1/punches", body=body) def employee_day(self, *, employee_id: str, date: str) -> Any: """Return punches and calculated attendance for one employee day.""" employee = quote(str(employee_id), safe="") day = quote(str(date), safe="") return self._request("GET", f"/api/v1/employees/{employee}/days/{day}") def timecards(self, *, date: Optional[str] = None) -> Any: """Return calculated daily timecard rows.""" return self._request("GET", "/api/v1/timecards", parameters={"date": date}) def monthly_timecard(self, *, month: str, employee_id: str) -> Any: """Return one employee's calculated monthly timecard.""" return self._request("GET", "/api/v1/monthly-timecards", parameters={ "month": month, "employeeId": employee_id, }) def timesheet(self, *, month: str, employee_id: str) -> Any: """Return one employee's monthly project/activity timesheet.""" return self._request("GET", "/api/v1/timesheets", parameters={ "month": month, "employeeId": employee_id, }) def export_timecards_csv( self, *, month: str, employee_id: Optional[str] = None, language: Optional[str] = None, ) -> str: """Download calculated timecards as localized CSV text.""" return self._request("GET", "/api/v1/exports/timecards.csv", parameters={ "month": month, "employeeId": employee_id, "lang": language, }, accept="text/csv") def export_timesheets_csv( self, *, month: str, employee_id: Optional[str] = None, language: Optional[str] = None, ) -> str: """Download timesheets as localized CSV text.""" return self._request("GET", "/api/v1/exports/timesheets.csv", parameters={ "month": month, "employeeId": employee_id, "lang": language, }, accept="text/csv")