pietrocal_client.pypython/pietrocal_client.py · 417 líneas
from __future__ import annotations
import json
import time
import uuid
from dataclasses import dataclass
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
IDEMPOTENT_POST_ENDPOINTS = {
"/calendars.php",
"/calendar_subscriptions.php",
"/events.php",
"/tasks.php",
"/contacts.php",
"/bookings.php",
"/availability_links.php",
}
class PietroCalApiError(RuntimeError):
def __init__(
self,
message: str,
status: int,
response_data: Any = None,
response_headers: dict[str, str] | None = None,
) -> None:
super().__init__(message)
self.status = status
self.response_data = response_data
self.response_headers = response_headers or {}
@dataclass(frozen=True)
class PietroCalResponse:
status: int
data: Any
headers: dict[str, str]
request_id: str | None
rate_limit: dict[str, int | None]
pagination: dict[str, int | None]
idempotency: dict[str, str | bool | None]
transport_attempts: int
class PietroCalClient:
def __init__(
self,
token: str,
base_url: str = "https://pietrocal.com/api/v1",
timeout_seconds: int = 30,
max_transport_retries: int = 1,
) -> None:
if not token.startswith("pc_live_"):
raise ValueError("El token debe comenzar con pc_live_.")
if timeout_seconds < 1:
raise ValueError("El timeout debe ser mayor que cero.")
if max_transport_retries < 0 or max_transport_retries > 5:
raise ValueError("max_transport_retries debe estar entre 0 y 5.")
self.token = token
self.base_url = base_url.rstrip("/")
self.timeout_seconds = timeout_seconds
self.max_transport_retries = max_transport_retries
def request(
self,
method: str,
endpoint: str,
query: dict[str, Any] | None = None,
body: Any = None,
*,
idempotency_key: str | None = None,
auto_idempotency: bool = True,
max_transport_retries: int | None = None,
) -> PietroCalResponse:
method = method.upper()
if method not in {"GET", "POST", "PATCH", "PUT", "DELETE"}:
raise ValueError("Método HTTP no permitido.")
endpoint = "/" + endpoint.lstrip("/")
if "\r" in endpoint or "\n" in endpoint:
raise ValueError("Endpoint inválido.")
url = f"{self.base_url}{endpoint}"
clean_query = {
key: value
for key, value in (query or {}).items()
if value is not None and value != ""
}
if clean_query:
url += "?" + urlencode(clean_query)
resolved_key = self._resolve_idempotency_key(
method,
endpoint,
idempotency_key,
auto_idempotency,
)
retry_limit = self._resolve_retry_limit(
method,
resolved_key,
max_transport_retries,
)
encoded_body = None
if body is not None:
encoded_body = json.dumps(
body,
ensure_ascii=False,
separators=(",", ":"),
).encode("utf-8")
transport_attempts = 0
while True:
transport_attempts += 1
headers = {
"Accept": "application/json",
"Authorization": f"Bearer {self.token}",
"User-Agent": "PietroCal-Python-SDK/1.1",
}
if encoded_body is not None:
headers["Content-Type"] = "application/json"
if resolved_key is not None:
headers["Idempotency-Key"] = resolved_key
request = Request(
url,
data=encoded_body,
headers=headers,
method=method,
)
try:
with urlopen(request, timeout=self.timeout_seconds) as response:
status = response.status
raw = response.read().decode("utf-8")
response_headers = {
key.lower(): value
for key, value in response.headers.items()
}
except HTTPError as error:
raw = error.read().decode("utf-8", errors="replace")
response_headers = {
key.lower(): value
for key, value in error.headers.items()
}
data = self._decode(raw)
message = self._error_message(data)
raise PietroCalApiError(
message,
error.code,
data,
response_headers,
) from error
except (URLError, TimeoutError, OSError) as error:
if transport_attempts <= retry_limit:
time.sleep(self._retry_delay_seconds(transport_attempts))
continue
reason = getattr(error, "reason", error)
raise RuntimeError(f"Error de red: {reason}") from error
data = self._decode(raw)
return PietroCalResponse(
status=status,
data=data,
headers=response_headers,
request_id=response_headers.get("x-request-id"),
rate_limit={
"limit": self._int_or_none(
response_headers.get("x-ratelimit-limit")
),
"remaining": self._int_or_none(
response_headers.get("x-ratelimit-remaining")
),
},
pagination={
"page": self._int_or_none(
response_headers.get("x-pagination-page")
),
"perPage": self._int_or_none(
response_headers.get("x-pagination-per-page")
),
"total": self._int_or_none(
response_headers.get("x-pagination-total")
),
"totalPages": self._int_or_none(
response_headers.get("x-pagination-total-pages")
),
},
idempotency={
"key": response_headers.get("idempotency-key")
or resolved_key,
"replayed": self._bool_or_none(
response_headers.get("idempotency-replayed")
),
},
transport_attempts=transport_attempts,
)
@staticmethod
def generate_idempotency_key() -> str:
return "sdk_" + uuid.uuid4().hex
def profile(self) -> PietroCalResponse:
return self.request("GET", "/me_app.php")
def calendars(self) -> PietroCalResponse:
return self.request("GET", "/calendars.php")
def events(
self,
page: int = 1,
per_page: int = 50,
**filters: Any,
) -> PietroCalResponse:
return self.request(
"GET",
"/events.php",
{"page": page, "perPage": per_page, **filters},
)
def tasks(self, **filters: Any) -> PietroCalResponse:
return self.request("GET", "/tasks.php", filters)
def contacts(
self,
page: int = 1,
per_page: int = 50,
**filters: Any,
) -> PietroCalResponse:
return self.request(
"GET",
"/contacts.php",
{"page": page, "perPage": per_page, **filters},
)
def bookings(
self,
page: int = 1,
per_page: int = 50,
**filters: Any,
) -> PietroCalResponse:
return self.request(
"GET",
"/bookings.php",
{"page": page, "perPage": per_page, **filters},
)
def create_calendar(
self,
calendar: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/calendars.php", calendar, idempotency_key)
def create_calendar_subscription(
self,
subscription: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create(
"/calendar_subscriptions.php",
subscription,
idempotency_key,
)
def create_event(
self,
event: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/events.php", event, idempotency_key)
def create_task(
self,
task: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/tasks.php", task, idempotency_key)
def create_contact(
self,
contact: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/contacts.php", contact, idempotency_key)
def create_booking(
self,
booking: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/bookings.php", booking, idempotency_key)
def create_availability_link(
self,
link: dict[str, Any],
idempotency_key: str | None = None,
) -> PietroCalResponse:
return self._create("/availability_links.php", link, idempotency_key)
def update_event(
self,
event_id: str,
event: dict[str, Any],
) -> PietroCalResponse:
return self.request("PATCH", "/events.php", {"id": event_id}, event)
def delete_event(self, event_id: str) -> PietroCalResponse:
return self.request("DELETE", "/events.php", {"id": event_id})
def _create(
self,
endpoint: str,
body: dict[str, Any],
idempotency_key: str | None,
) -> PietroCalResponse:
return self.request(
"POST",
endpoint,
body=body,
idempotency_key=idempotency_key,
)
def _resolve_idempotency_key(
self,
method: str,
endpoint: str,
explicit_key: str | None,
auto_idempotency: bool,
) -> str | None:
if explicit_key is not None:
if not self._valid_idempotency_key(explicit_key):
raise ValueError("Idempotency-Key inválida.")
return explicit_key
if (
auto_idempotency
and method == "POST"
and endpoint in IDEMPOTENT_POST_ENDPOINTS
):
return self.generate_idempotency_key()
return None
def _resolve_retry_limit(
self,
method: str,
idempotency_key: str | None,
override: int | None,
) -> int:
if method != "POST" or idempotency_key is None:
return 0
value = self.max_transport_retries if override is None else override
if value < 0 or value > 5:
raise ValueError("max_transport_retries debe estar entre 0 y 5.")
return value
@staticmethod
def _valid_idempotency_key(key: str) -> bool:
if len(key) < 8 or len(key) > 200:
return False
first, rest = key[0], key[1:]
allowed = set(
"abcdefghijklmnopqrstuvwxyz"
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"0123456789._:-"
)
return first.isalnum() and all(character in allowed for character in rest)
@staticmethod
def _retry_delay_seconds(attempt: int) -> float:
return min(2.0, 0.25 * (2 ** max(0, attempt - 1)))
@staticmethod
def _decode(raw: str) -> Any:
if not raw.strip():
return None
try:
return json.loads(raw)
except json.JSONDecodeError:
return raw
@staticmethod
def _error_message(data: Any) -> str:
if isinstance(data, dict):
error = data.get("error")
if isinstance(error, dict) and error.get("message"):
return str(error["message"])
if data.get("message"):
return str(data["message"])
return "La API devolvió un error."
@staticmethod
def _int_or_none(value: str | None) -> int | None:
try:
return int(value) if value is not None else None
except (TypeError, ValueError):
return None
@staticmethod
def _bool_or_none(value: str | None) -> bool | None:
if value is None:
return None
normalized = value.strip().lower()
if normalized in {"true", "1", "yes"}:
return True
if normalized in {"false", "0", "no"}:
return False
return None