remove src folder
This commit is contained in:
21
pypretix_device/__init__.py
Normal file
21
pypretix_device/__init__.py
Normal file
@@ -0,0 +1,21 @@
|
||||
"""pypretix-device: A minimal API client for pretix device authentication and ticket scanning."""
|
||||
|
||||
from pypretix_device import auth
|
||||
from pypretix_device.client import PretixDeviceClient
|
||||
from pypretix_device.models import (
|
||||
CheckinList,
|
||||
CheckinResult,
|
||||
DeviceInfo,
|
||||
Event,
|
||||
SearchResult,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"auth",
|
||||
"CheckinList",
|
||||
"CheckinResult",
|
||||
"DeviceInfo",
|
||||
"Event",
|
||||
"PretixDeviceClient",
|
||||
"SearchResult",
|
||||
]
|
||||
252
pypretix_device/auth.py
Normal file
252
pypretix_device/auth.py
Normal file
@@ -0,0 +1,252 @@
|
||||
"""Device registration and token persistence for pypretix-device."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import platform
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
import requests
|
||||
|
||||
from pypretix_device.models import DeviceInfo
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
CONFIG_FILENAME = "config.json"
|
||||
SOFTWARE_BRAND = "pypretix"
|
||||
SOFTWARE_VERSION = "0.1.0"
|
||||
|
||||
|
||||
def _get_config_path() -> Path:
|
||||
"""Get the config file path."""
|
||||
from pypretix_device.models import CONFIG_DIR
|
||||
if CONFIG_DIR:
|
||||
return Path(CONFIG_DIR) / CONFIG_FILENAME
|
||||
return Path.home() / ".config" / "pypretix" / CONFIG_FILENAME
|
||||
|
||||
|
||||
def _ensure_config_dir() -> Path:
|
||||
"""Create config directory if it doesn't exist."""
|
||||
config_path = _get_config_path().parent
|
||||
config_path.mkdir(parents=True, exist_ok=True)
|
||||
return config_path
|
||||
|
||||
|
||||
def load_config() -> Optional[Dict[str, Any]]:
|
||||
"""Load config from disk. Returns None if not found or invalid."""
|
||||
config_path = _get_config_path()
|
||||
if not config_path.exists():
|
||||
return None
|
||||
try:
|
||||
with open(config_path, "r") as f:
|
||||
data = json.load(f)
|
||||
# Validate required fields
|
||||
required = ["base_url", "api_token"]
|
||||
if not all(k in data for k in required):
|
||||
logger.warning("Config file missing required fields")
|
||||
return None
|
||||
return data
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
logger.error(f"Failed to load config: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def save_config(config: Dict[str, Any]) -> None:
|
||||
"""Save config to disk."""
|
||||
_ensure_config_dir()
|
||||
config_path = _get_config_path()
|
||||
with open(config_path, "w") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
logger.info(f"Config saved to {config_path}")
|
||||
|
||||
|
||||
class RegistrationFailedError(Exception):
|
||||
"""Raised when device registration fails."""
|
||||
pass
|
||||
|
||||
|
||||
def register_device(
|
||||
base_url: str,
|
||||
init_token: str,
|
||||
device_name: str = "example-scanner",
|
||||
) -> DeviceInfo:
|
||||
"""Register a new device with pretix using an initialization token.
|
||||
|
||||
Args:
|
||||
base_url: Base URL of the pretix instance (e.g. "http://localhost").
|
||||
init_token: The 24-character initialization token from the web UI.
|
||||
device_name: Human-readable name for this device.
|
||||
|
||||
Returns:
|
||||
DeviceInfo with the newly assigned API token.
|
||||
|
||||
Raises:
|
||||
RegistrationFailedError: If registration fails.
|
||||
"""
|
||||
# Normalize base_url - ensure no trailing slash, ensure no path
|
||||
base_url = base_url.rstrip("/")
|
||||
url = f"{base_url}/api/v1/device/initialize"
|
||||
|
||||
payload = {
|
||||
"token": init_token,
|
||||
"hardware_brand": platform.machine() or "unknown",
|
||||
"hardware_model": "pypretix-device",
|
||||
"os_name": platform.system() or "unknown",
|
||||
"os_version": platform.version() or "unknown",
|
||||
"software_brand": SOFTWARE_BRAND,
|
||||
"software_version": SOFTWARE_VERSION,
|
||||
}
|
||||
|
||||
try:
|
||||
resp = requests.post(url, json=payload, timeout=30)
|
||||
except requests.RequestException as e:
|
||||
raise RegistrationFailedError(f"Failed to connect to pretix at {base_url}: {e}")
|
||||
|
||||
if resp.status_code != 200:
|
||||
try:
|
||||
errors = resp.json()
|
||||
error_msg = json.dumps(errors)
|
||||
except (ValueError, KeyError):
|
||||
error_msg = resp.text[:500]
|
||||
raise RegistrationFailedError(
|
||||
f"Device registration failed (HTTP {resp.status_code}): {error_msg}"
|
||||
)
|
||||
|
||||
data = resp.json()
|
||||
device_info = DeviceInfo(
|
||||
organizer=data["organizer"],
|
||||
device_id=data["device_id"],
|
||||
name=data["name"],
|
||||
api_token=data["api_token"],
|
||||
unique_serial=data.get("unique_serial", ""),
|
||||
gate=data.get("gate"),
|
||||
server_version=data.get("server", {}).get("version", {}).get("pretix"),
|
||||
)
|
||||
|
||||
# Save config if directory is set
|
||||
from pypretix_device.models import CONFIG_DIR
|
||||
if CONFIG_DIR:
|
||||
config = {
|
||||
"base_url": base_url,
|
||||
"api_token": device_info.api_token,
|
||||
"device_id": device_info.device_id,
|
||||
"organizer": device_info.organizer,
|
||||
"device_name": device_name,
|
||||
"software_version": SOFTWARE_VERSION,
|
||||
}
|
||||
if data.get("gate"):
|
||||
config["checkin_list"] = data["gate"].get("id")
|
||||
save_config(config)
|
||||
logger.info(f"Device '{device_info.name}' registered successfully (id={device_info.device_id})")
|
||||
|
||||
return device_info
|
||||
|
||||
|
||||
def get_or_register_device(
|
||||
base_url: str,
|
||||
init_token: Optional[str] = None,
|
||||
device_name: str = "helferscanner",
|
||||
) -> DeviceInfo:
|
||||
"""Load existing device config or register a new one.
|
||||
|
||||
Args:
|
||||
base_url: Base URL of the pretix instance.
|
||||
init_token: Init token for registration. Required if no config exists.
|
||||
device_name: Device name for registration.
|
||||
|
||||
Returns:
|
||||
DeviceInfo for the current device.
|
||||
|
||||
Raises:
|
||||
RegistrationFailedError: If no config exists and no init_token provided.
|
||||
"""
|
||||
config = load_config()
|
||||
if config and "api_token" in config:
|
||||
logger.info(f"Loaded config from {_get_config_path()}")
|
||||
return DeviceInfo(
|
||||
organizer=config.get("organizer", ""),
|
||||
device_id=config.get("device_id", 0),
|
||||
name=config.get("device_name", "Unknown"),
|
||||
api_token=config["api_token"],
|
||||
unique_serial=config.get("unique_serial", ""),
|
||||
)
|
||||
|
||||
if not init_token:
|
||||
raise RegistrationFailedError(
|
||||
"No existing config found and no init_token provided.\n"
|
||||
"Create a new device in the pretix web UI (Organizer Settings => Devices) "
|
||||
"to obtain an initialization token."
|
||||
)
|
||||
|
||||
return register_device(base_url, init_token, device_name)
|
||||
|
||||
|
||||
def revoke_device() -> None:
|
||||
"""Revoke (disable) the current device. This invalidates the API token permanently."""
|
||||
config = load_config()
|
||||
if not config:
|
||||
logger.warning("No config found, nothing to revoke")
|
||||
return
|
||||
|
||||
base_url = config.get("base_url", "").rstrip("/")
|
||||
api_token = config.get("api_token", "")
|
||||
|
||||
if not base_url or not api_token:
|
||||
logger.warning("Config missing base_url or api_token")
|
||||
return
|
||||
|
||||
url = f"{base_url}/api/v1/device/revoke/"
|
||||
try:
|
||||
resp = requests.post(
|
||||
url,
|
||||
headers={"Authorization": f"Device {api_token}"},
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code in (200, 204):
|
||||
logger.info("Device revoked successfully")
|
||||
else:
|
||||
logger.error(f"Revocation failed (HTTP {resp.status_code})")
|
||||
except requests.RequestException as e:
|
||||
logger.error(f"Failed to revoke device: {e}")
|
||||
|
||||
# Remove config file
|
||||
config_path = _get_config_path()
|
||||
if config_path.exists():
|
||||
config_path.unlink()
|
||||
logger.info(f"Config file removed: {config_path}")
|
||||
|
||||
|
||||
def update_device_info(
|
||||
base_url: str,
|
||||
api_token: str,
|
||||
software_version: str = SOFTWARE_VERSION,
|
||||
) -> Dict[str, Any]:
|
||||
"""Notify the server about a software version update."""
|
||||
base_url = base_url.rstrip("/")
|
||||
url = f"{base_url}/api/v1/device/update/"
|
||||
|
||||
payload = {
|
||||
"hardware_brand": platform.machine() or "unknown",
|
||||
"hardware_model": "pypretix-device",
|
||||
"os_name": platform.system() or "unknown",
|
||||
"os_version": platform.version() or "unknown",
|
||||
"software_brand": SOFTWARE_BRAND,
|
||||
"software_version": software_version,
|
||||
}
|
||||
|
||||
try:
|
||||
resp = requests.post(
|
||||
url,
|
||||
headers={"Authorization": f"Device {api_token}"},
|
||||
json=payload,
|
||||
timeout=10,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
return resp.json()
|
||||
logger.debug(f"Update failed (HTTP {resp.status_code})")
|
||||
return {}
|
||||
except requests.RequestException as e:
|
||||
logger.debug(f"Update request failed: {e}")
|
||||
return {}
|
||||
497
pypretix_device/client.py
Normal file
497
pypretix_device/client.py
Normal file
@@ -0,0 +1,497 @@
|
||||
"""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
|
||||
83
pypretix_device/models.py
Normal file
83
pypretix_device/models.py
Normal file
@@ -0,0 +1,83 @@
|
||||
"""Data models for pypretix-device API responses and operations."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckinResult:
|
||||
"""Result of a check-in or check-out operation via checkinrpc/redeem."""
|
||||
|
||||
status: str # "ok", "incomplete", "exchange", "error"
|
||||
reason: Optional[str] = None # "invalid", "already_redeemed", "unpaid", etc.
|
||||
reason_explanation: Optional[str] = None
|
||||
position: Optional[Dict[str, Any]] = None
|
||||
require_attention: bool = False
|
||||
checkin_texts: List[str] = field(default_factory=list)
|
||||
list_info: Optional[Dict[str, Any]] = None
|
||||
questions: Optional[List[Dict[str, Any]]] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class SearchResult:
|
||||
"""Result of a ticket search via checkinrpc/search."""
|
||||
|
||||
position: Optional[Dict[str, Any]] = None
|
||||
require_attention: bool = False
|
||||
checkin_texts: List[str] = field(default_factory=list)
|
||||
list_info: Optional[Dict[str, Any]] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class Event:
|
||||
"""Pretix event."""
|
||||
|
||||
slug: str
|
||||
name: str
|
||||
testmode: bool = False
|
||||
subevent_id: Optional[int] = None
|
||||
currency: str = "EUR"
|
||||
date_from: Optional[str] = None
|
||||
date_to: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CheckinList:
|
||||
"""A check-in list within an event."""
|
||||
|
||||
id: int
|
||||
name: str
|
||||
all_products: bool = True
|
||||
checkin_count: int = 0
|
||||
position_count: int = 0
|
||||
allow_entry_after_exit: bool = False
|
||||
subevent: Optional[int] = None
|
||||
include_pending: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class AttendeeInfo:
|
||||
"""Attendee/personalized ticket information."""
|
||||
|
||||
name: str
|
||||
email: Optional[str] = None
|
||||
position_id: Optional[int] = None
|
||||
order_code: Optional[str] = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class DeviceInfo:
|
||||
"""Information about the registered device."""
|
||||
|
||||
organizer: str
|
||||
device_id: int
|
||||
name: str
|
||||
api_token: str
|
||||
unique_serial: str
|
||||
gate: Optional[Dict[str, Any]] = None
|
||||
server_version: Optional[str] = None
|
||||
|
||||
|
||||
CONFIG_DIR = None # Set by Client before using
|
||||
Reference in New Issue
Block a user