100 lines
4.3 KiB
Python
100 lines
4.3 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
|
||
|
|
class NotFoundError(RuntimeError):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class WebsiteContentAPIError(RuntimeError):
|
||
|
|
pass
|
||
|
|
|
||
|
|
|
||
|
|
class WebsiteContentService:
|
||
|
|
"""Authenticated HTTPS client for the website administration API."""
|
||
|
|
|
||
|
|
def __init__(self, api_url=None, api_token=None, client: httpx.Client | None = None):
|
||
|
|
self.api_url = (api_url or settings.WEBSITE_CONTENT_API_URL).rstrip("/")
|
||
|
|
explicit_token = api_token is not None
|
||
|
|
self.api_token = api_token if explicit_token else settings.WEBSITE_CONTENT_API_TOKEN
|
||
|
|
if not explicit_token and not self.api_token and settings.WEBSITE_CONTENT_MYSQL_PASSWORD:
|
||
|
|
self.api_token = settings.WEBSITE_CONTENT_MYSQL_PASSWORD
|
||
|
|
self.client = client or httpx.Client(timeout=settings.WEBSITE_CONTENT_API_TIMEOUT_SECONDS)
|
||
|
|
|
||
|
|
def _request(self, method: str, resource: str, *, item_id=None, action=None,
|
||
|
|
data=None, files=None, expect_bytes=False, extra_params=None):
|
||
|
|
if not self.api_url or not self.api_token:
|
||
|
|
raise WebsiteContentAPIError("Website admin-API er ikke konfigureret")
|
||
|
|
params = {"resource": resource, **(extra_params or {})}
|
||
|
|
if item_id is not None:
|
||
|
|
params["id"] = item_id
|
||
|
|
if action:
|
||
|
|
params["action"] = action
|
||
|
|
headers = {"X-Website-Admin-Token": self.api_token, "Accept": "application/json"}
|
||
|
|
if settings.WEBSITE_CONTENT_API_HOST_HEADER:
|
||
|
|
headers["Host"] = settings.WEBSITE_CONTENT_API_HOST_HEADER
|
||
|
|
try:
|
||
|
|
response = self.client.request(
|
||
|
|
method, self.api_url, params=params, headers=headers,
|
||
|
|
json=self._json_values(data) if data is not None and files is None else None,
|
||
|
|
files=files,
|
||
|
|
)
|
||
|
|
except httpx.HTTPError as exc:
|
||
|
|
raise WebsiteContentAPIError("Kunne ikke kontakte website admin-API") from exc
|
||
|
|
if response.status_code == 404:
|
||
|
|
raise NotFoundError()
|
||
|
|
if not response.is_success:
|
||
|
|
try:
|
||
|
|
payload = response.json()
|
||
|
|
detail = payload.get("message") or payload.get("error")
|
||
|
|
except (ValueError, AttributeError):
|
||
|
|
detail = None
|
||
|
|
raise WebsiteContentAPIError(detail or f"Website admin-API svarede {response.status_code}")
|
||
|
|
if expect_bytes:
|
||
|
|
return response.content, response.headers.get("content-type", "application/octet-stream")
|
||
|
|
return response.json()
|
||
|
|
|
||
|
|
def list(self, kind: str, include_hidden: bool = True) -> list[dict[str, Any]]:
|
||
|
|
result = self._request("GET", kind, extra_params={"include_hidden": int(include_hidden)})
|
||
|
|
return result.get("items", result) if isinstance(result, dict) else result
|
||
|
|
|
||
|
|
def get(self, kind: str, item_id: int) -> dict[str, Any]:
|
||
|
|
return self._request("GET", kind, item_id=item_id)
|
||
|
|
|
||
|
|
def create(self, kind: str, data: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("POST", kind, data=data)
|
||
|
|
|
||
|
|
def update(self, kind: str, item_id: int, data: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
return self._request("PATCH", kind, item_id=item_id, data=data)
|
||
|
|
|
||
|
|
def upload_logo(self, item_id: int, content: bytes, mime_type: str) -> dict[str, Any]:
|
||
|
|
return self._request("POST", "customers", item_id=item_id, action="logo",
|
||
|
|
files={"logo": ("logo", content, mime_type)})
|
||
|
|
|
||
|
|
def logo(self, item_id: int) -> tuple[bytes, str]:
|
||
|
|
return self._request("GET", "customers", item_id=item_id, action="logo", expect_bytes=True)
|
||
|
|
|
||
|
|
def complete_operation(self, item_id: int, ends_at: datetime | None, is_public: bool):
|
||
|
|
return self._request("POST", "operations", item_id=item_id, action="complete",
|
||
|
|
data={"ends_at": ends_at, "is_public": is_public})
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def _json_values(cls, data):
|
||
|
|
if data is None:
|
||
|
|
return None
|
||
|
|
return {key: value.isoformat() if isinstance(value, datetime)
|
||
|
|
else str(value) if cls._is_url(value) else value for key, value in data.items()}
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def _is_url(value: Any) -> bool:
|
||
|
|
return value is not None and value.__class__.__module__.startswith("pydantic.networks")
|
||
|
|
|
||
|
|
|
||
|
|
website_content_service = WebsiteContentService()
|