498 lines
18 KiB
Python
498 lines
18 KiB
Python
"""HTTP client for pretix device authentication endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
import requests
|
|
|
|
from pypretix_device.models import CheckinList, CheckinResult, DeviceInfo, Event, SearchResult
|
|
|
|
# API endpoint bases
|
|
_CHECKINRPC_REDEEM = "checkinrpc/redeem/"
|
|
_CHECKINRPC_SEARCH = "checkinrpc/search/"
|
|
_DEVICE_INFO = "api/v1/device/info"
|
|
_DEVICE_UPDATE = "api/v1/device/update"
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PretixDeviceClient:
|
|
"""Client for pretix device authentication and ticket scanning.
|
|
|
|
Args:
|
|
base_url: Base URL of the pretix instance (e.g. "http://localhost").
|
|
api_token: Device API token.
|
|
organizer_slug: Slug of the organizer (e.g. "TEST").
|
|
event_slug: Slug of the event (e.g. "PB26").
|
|
checkin_list_id: ID of the default check-in list to use.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
api_token: str,
|
|
organizer_slug: str,
|
|
event_slug: str,
|
|
checkin_list_id: Optional[int] = None,
|
|
):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_token = api_token
|
|
self.organizer_slug = organizer_slug
|
|
self.event_slug = event_slug
|
|
self.checkin_list_id = checkin_list_id
|
|
self._session = requests.Session()
|
|
self._session.headers.update({
|
|
"Authorization": f"Device {api_token}",
|
|
"Accept": "application/json",
|
|
})
|
|
self._last_error: Optional[str] = None
|
|
self._last_response_time: float = 0
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def _request(self, method: str, endpoint: str, **kwargs) -> Optional[requests.Response]:
|
|
"""Make an authenticated request with retry on rate limits."""
|
|
# Default timeout - retries will add more
|
|
kwargs.setdefault("timeout", 15)
|
|
|
|
url = f"{self.base_url}/{endpoint.lstrip('/')}"
|
|
max_retries = 3
|
|
base_delay = 2
|
|
|
|
for attempt in range(max_retries):
|
|
try:
|
|
if method == "GET":
|
|
resp = self._session.get(url, **kwargs)
|
|
elif method == "POST":
|
|
resp = self._session.post(url, **kwargs)
|
|
else:
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
|
|
self._last_response_time = time.time()
|
|
self._last_error = None
|
|
|
|
if resp.status_code == 429:
|
|
retry_after = int(resp.headers.get("Retry-After", base_delay * (2 ** attempt)))
|
|
logger.warning(f"Rate limited (429). Waiting {retry_after}s")
|
|
time.sleep(retry_after)
|
|
continue
|
|
|
|
if resp.status_code >= 500:
|
|
if attempt < max_retries - 1:
|
|
delay = base_delay * (2 ** attempt)
|
|
logger.warning(f"Server error {resp.status_code}. Retrying in {delay}s")
|
|
time.sleep(delay)
|
|
continue
|
|
return resp
|
|
|
|
return resp
|
|
|
|
except requests.RequestException as e:
|
|
if attempt < max_retries - 1:
|
|
delay = base_delay * (2 ** attempt)
|
|
logger.warning(f"Request failed ({e}). Retrying in {delay}s")
|
|
time.sleep(delay)
|
|
continue
|
|
raise
|
|
|
|
return None
|
|
|
|
def _api_url(self, endpoint: str) -> str:
|
|
"""Build full API URL."""
|
|
return f"/api/v1/organizers/{self.organizer_slug}/{endpoint.lstrip('/')}"
|
|
|
|
# ------------------------------------------------------------------
|
|
# Check-in operations
|
|
# ------------------------------------------------------------------
|
|
|
|
def checkin(self, barcode: str, lists: Optional[List[int]] = None) -> CheckinResult:
|
|
"""Check a ticket in (entry).
|
|
|
|
Args:
|
|
barcode: The ticket secret/barcode string.
|
|
lists: Explicit list of check-in list IDs. Uses client default if not provided.
|
|
|
|
Returns:
|
|
CheckinResult with status, position info, and any error details.
|
|
"""
|
|
checkin_list_id = lists or [self.checkin_list_id] if self.checkin_list_id else None
|
|
if checkin_list_id is None:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="invalid",
|
|
reason_explanation="No check-in list configured.",
|
|
)
|
|
|
|
endpoint = _CHECKINRPC_REDEEM
|
|
payload = {
|
|
"secret": barcode,
|
|
"source_type": "barcode",
|
|
"lists": checkin_list_id,
|
|
"type": "entry",
|
|
"force": True,
|
|
"ignore_unpaid": False,
|
|
"questions_supported": False,
|
|
}
|
|
|
|
try:
|
|
resp = self._request("POST", self._api_url(endpoint), json=payload)
|
|
if resp is None:
|
|
raise Exception("Request failed after retries")
|
|
|
|
data = resp.json()
|
|
|
|
# Handle HTTP 404 for unknown tickets
|
|
if resp.status_code == 404:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="invalid",
|
|
)
|
|
|
|
return self._parse_checkin_result(data, resp.status_code)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Check-in request failed: {e}")
|
|
raise
|
|
except requests.exceptions.JSONDecodeError:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="error",
|
|
reason_explanation=f"Unexpected response: {resp.text[:200] if resp else 'No response'}",
|
|
)
|
|
|
|
def checkout(self, barcode: str, lists: Optional[List[int]] = None) -> CheckinResult:
|
|
"""Check a ticket out (exit).
|
|
|
|
Args:
|
|
barcode: The ticket secret/barcode string.
|
|
lists: Explicit list of check-in list IDs. Uses client default if not provided.
|
|
|
|
Returns:
|
|
CheckinResult with status, position info, and any error details.
|
|
"""
|
|
checkin_list_id = lists or [self.checkin_list_id] if self.checkin_list_id else None
|
|
if checkin_list_id is None:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="invalid",
|
|
reason_explanation="No check-in list configured.",
|
|
)
|
|
|
|
endpoint = _CHECKINRPC_REDEEM
|
|
payload = {
|
|
"secret": barcode,
|
|
"source_type": "barcode",
|
|
"lists": checkin_list_id,
|
|
"type": "exit",
|
|
"force": True,
|
|
"ignore_unpaid": False,
|
|
"questions_supported": False,
|
|
}
|
|
|
|
try:
|
|
resp = self._request("POST", self._api_url(endpoint), json=payload)
|
|
if resp is None:
|
|
raise Exception("Request failed after retries")
|
|
|
|
data = resp.json()
|
|
|
|
if resp.status_code == 404:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="invalid",
|
|
)
|
|
|
|
return self._parse_checkin_result(data, resp.status_code)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Check-out request failed: {e}")
|
|
raise
|
|
except requests.exceptions.JSONDecodeError:
|
|
return CheckinResult(
|
|
status="error",
|
|
reason="error",
|
|
reason_explanation=f"Unexpected response: {resp.text[:200] if resp else 'No response'}",
|
|
)
|
|
|
|
def search(self, barcode: str, lists: Optional[List[int]] = None) -> SearchResult:
|
|
"""Search for a ticket by barcode/secret (read-only lookup).
|
|
|
|
Args:
|
|
barcode: The ticket secret/barcode string to search for.
|
|
lists: Explicit list of check-in list IDs to search on.
|
|
|
|
Returns:
|
|
SearchResult with ticket details and check-in history.
|
|
"""
|
|
checkin_list_id = lists or [self.checkin_list_id] if self.checkin_list_id else None
|
|
|
|
query_params = {"secret": barcode}
|
|
if checkin_list_id:
|
|
query_params["list"] = checkin_list_id[0]
|
|
|
|
endpoint = _CHECKINRPC_SEARCH
|
|
|
|
try:
|
|
resp = self._request("GET", self._api_url(endpoint), params=query_params)
|
|
if resp is None:
|
|
raise Exception("Request failed after retries")
|
|
|
|
data = resp.json()
|
|
|
|
# Handle edge case: API returns an error string/list instead of a dict
|
|
# (e.g. "No check-in list passed.") when lists is not configured
|
|
if not isinstance(data, dict):
|
|
logger.warning(f"Unexpected search response type: {type(data)} - {data}")
|
|
return SearchResult(position=None, require_attention=False)
|
|
|
|
# Check for empty results
|
|
results = data.get("results", [])
|
|
if not results:
|
|
return SearchResult(
|
|
position=None,
|
|
require_attention=False,
|
|
)
|
|
|
|
position = results[0]
|
|
# Extract list info if available
|
|
list_info = None
|
|
if position.get("list"):
|
|
list_info = position["list"]
|
|
|
|
return SearchResult(
|
|
position=position,
|
|
require_attention=position.get("require_attention", False),
|
|
checkin_texts=position.get("checkin_texts", []),
|
|
list_info=list_info,
|
|
)
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Search request failed: {e}")
|
|
raise
|
|
except requests.exceptions.JSONDecodeError:
|
|
return SearchResult(
|
|
position=None,
|
|
require_attention=False,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Event and Check-in List management
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_events(self) -> List[Event]:
|
|
"""List all events for the configured organizer.
|
|
|
|
Returns:
|
|
List of Event objects.
|
|
"""
|
|
endpoint = "events/"
|
|
try:
|
|
resp = self._request("GET", self._api_url(endpoint))
|
|
if resp is None:
|
|
logger.error("Failed to list events")
|
|
return []
|
|
|
|
data = resp.json()
|
|
events = []
|
|
for item in data.get("results", []):
|
|
# Handle both string and dict names
|
|
name_raw = item.get("name", "")
|
|
if isinstance(name_raw, dict):
|
|
name = name_raw.get("de") or name_raw.get("en") or "Unnamed"
|
|
else:
|
|
name = str(name_raw)
|
|
|
|
events.append(Event(
|
|
slug=item["slug"],
|
|
name=name,
|
|
testmode=item.get("testmode", False),
|
|
currency=item.get("currency", "EUR"),
|
|
date_from=item.get("date_from"),
|
|
date_to=item.get("date_to"),
|
|
))
|
|
return events
|
|
|
|
except (requests.exceptions.RequestException, ValueError, KeyError) as e:
|
|
logger.error(f"Failed to list events: {e}")
|
|
return []
|
|
|
|
def list_checkin_lists(self, event_slug: Optional[str] = None) -> List[CheckinList]:
|
|
"""List check-in lists for the current (or specified) event.
|
|
|
|
Args:
|
|
event_slug: Event slug to fetch lists for. Uses client default if not provided.
|
|
|
|
Returns:
|
|
List of CheckinList objects.
|
|
"""
|
|
event = event_slug or self.event_slug
|
|
checkin = "checkinlists"
|
|
endpoint = f"events/{event}/{checkin}/"
|
|
try:
|
|
resp = self._request("GET", self._api_url(endpoint))
|
|
if resp is None:
|
|
logger.error("Failed to list check-in lists")
|
|
return []
|
|
|
|
data = resp.json()
|
|
result = []
|
|
for item in data.get("results", []):
|
|
result.append(CheckinList(
|
|
id=item["id"],
|
|
name=item["name"],
|
|
all_products=item.get("all_products", True),
|
|
checkin_count=item.get("checkin_count", 0),
|
|
position_count=item.get("position_count", 0),
|
|
allow_entry_after_exit=item.get("allow_entry_after_exit", False),
|
|
subevent=item.get("subevent"),
|
|
))
|
|
return result
|
|
|
|
except (requests.exceptions.RequestException, ValueError, KeyError) as e:
|
|
logger.error(f"Failed to list check-in lists: {e}")
|
|
return []
|
|
|
|
def select_event(self, event_slug: str) -> bool:
|
|
"""Switch the client to use a different event.
|
|
|
|
Args:
|
|
event_slug: The slug of the new event.
|
|
|
|
Returns:
|
|
True if switch was successful, False otherwise.
|
|
"""
|
|
try:
|
|
self._request("GET", self._api_url(f"events/{event_slug}/"))
|
|
self.event_slug = event_slug
|
|
self._last_error = None
|
|
logger.info(f"Selected event: {event_slug}")
|
|
return True
|
|
except (requests.exceptions.RequestException, ValueError):
|
|
self._last_error = f"Event not found: {event_slug}"
|
|
logger.error(self._last_error)
|
|
return False
|
|
|
|
# ------------------------------------------------------------------
|
|
# Device management
|
|
# ------------------------------------------------------------------
|
|
|
|
def device_info(self) -> Optional[DeviceInfo]:
|
|
"""Fetch device information from the pretix server.
|
|
|
|
Returns:
|
|
DeviceInfo if successful, None on failure.
|
|
"""
|
|
endpoint = _DEVICE_INFO
|
|
try:
|
|
resp = self._request("GET", endpoint)
|
|
if resp is None:
|
|
logger.error("Failed to fetch device info")
|
|
return None
|
|
|
|
data = resp.json()
|
|
dev_data = data.get("device", {})
|
|
server_data = data.get("server", {})
|
|
version_data = server_data.get("version", {})
|
|
|
|
return DeviceInfo(
|
|
organizer=dev_data.get("organizer", ""),
|
|
device_id=dev_data.get("device_id", 0),
|
|
name=dev_data.get("name", ""),
|
|
api_token=dev_data.get("api_token", ""),
|
|
unique_serial=dev_data.get("unique_serial", ""),
|
|
gate=dev_data.get("gate"),
|
|
server_version=version_data.get("pretix"),
|
|
)
|
|
|
|
except (requests.exceptions.RequestException, ValueError, KeyError) as e:
|
|
logger.error(f"Failed to fetch device info: {e}")
|
|
return None
|
|
|
|
def device_update(self, software_version: str) -> Optional[DeviceInfo]:
|
|
"""Update the device's software version on the server.
|
|
|
|
Args:
|
|
software_version: Current software version string.
|
|
|
|
Returns:
|
|
DeviceInfo if successful, None on failure.
|
|
"""
|
|
endpoint = DEVICE_UPDATE_ENDPOINT
|
|
payload = {
|
|
"software_brand": "pypretix",
|
|
"software_version": software_version,
|
|
}
|
|
try:
|
|
resp = self._request("POST", endpoint, json=payload)
|
|
if resp is None:
|
|
return None
|
|
|
|
data = resp.json()
|
|
dev_data = data.get("device", {})
|
|
server_data = data.get("server", {})
|
|
version_data = server_data.get("version", {})
|
|
|
|
if not dev_data:
|
|
return None
|
|
|
|
return DeviceInfo(
|
|
organizer=dev_data.get("organizer", ""),
|
|
device_id=dev_data.get("device_id", 0),
|
|
name=dev_data.get("name", ""),
|
|
api_token=dev_data.get("api_token", ""),
|
|
unique_serial=dev_data.get("unique_serial", ""),
|
|
gate=dev_data.get("gate"),
|
|
server_version=version_data.get("pretix"),
|
|
)
|
|
|
|
except (requests.exceptions.RequestException, ValueError, KeyError) as e:
|
|
logger.error(f"Device update failed: {e}")
|
|
return None
|
|
|
|
# ------------------------------------------------------------------
|
|
# Connection status
|
|
# ------------------------------------------------------------------
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
"""Check if the client can reach the server.
|
|
|
|
Returns:
|
|
True if the last request succeeded, False otherwise.
|
|
"""
|
|
return self._last_error is None
|
|
|
|
@property
|
|
def last_error(self) -> Optional[str]:
|
|
"""Get the last error message (if any)."""
|
|
return self._last_error
|
|
|
|
@property
|
|
def last_response_time(self) -> float:
|
|
"""Get the timestamp of the last successful API response."""
|
|
return self._last_response_time
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal: parse checkin result from API response
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _parse_checkin_result(data: Dict[str, Any], status_code: int) -> CheckinResult:
|
|
"""Parse a checkinrpc/redeem response into a CheckinResult."""
|
|
result = CheckinResult(
|
|
status=data.get("status", "error"),
|
|
reason=data.get("reason"),
|
|
reason_explanation=data.get("reason_explanation"),
|
|
position=data.get("position"),
|
|
require_attention=data.get("require_attention", False),
|
|
checkin_texts=data.get("checkin_texts", []),
|
|
list_info=data.get("list"),
|
|
questions=data.get("questions"),
|
|
)
|
|
return result
|