Added a simple AI generated Python web client to run tools and scripts

src/brecal_api_client/README.md

@@ -0,0 +1,51 @@
+# BreCal API Client
+
+Minimal Python helper for `misc/BreCalApi.yaml`. It focuses on the login, shipcall, and times endpoints needed by CLI tools, but the helper method `BreCalClient.raw_request` makes it straightforward to call any other endpoint defined in the OpenAPI specification.
+
+Dependencies: only the `requests` package in addition to the standard library.
+
+## Endpoint selection
+
+`BreCalClient` reads its default `base_url` from `~/.config/brecal/client.json`. The file lets you define multiple deployments and switch between them without modifying code:
+
+```json
+{
+  "environment": "devel",
+  "endpoints": {
+    "local": "http://localhost:5000",
+    "devel": "https://brecaldevel.bsmd-emswe.eu",
+    "test": "https://brecaltest.example.net",
+    "prod": "https://brecal.example.com"
+  }
+}
+```
+
+Override the selection at runtime via `BreCalClient(base_url="...")` or the environment variable `BRECAL_BASE_URL`. If no config is present the client falls back to the development server URL.
+
+## Credentials
+
+Store credentials in `~/.config/brecal/credentials.json`:
+
+```json
+{
+  "username": "alfred",
+  "password": "123456"
+}
+```
+
+You can override the location when calling `Credentials.load("/path/to/file.json")` or provide credentials from environment variables via `Credentials.from_env()`.
+
+## Example
+
+```python
+from brecal_api_client import BreCalClient, Credentials
+
+creds = Credentials.load()
+with BreCalClient(credentials=creds) as client:
+    # list ship calls from the last week
+    shipcalls = client.get_shipcalls(past_days=7)
+
+    # create/update ship calls or times
+    shipcall_id = client.create_shipcall({...})
+    times = client.get_times(shipcall_id=shipcall_id)
+```

src/brecal_api_client/__init__.py

@@ -0,0 +1,25 @@
+"""Simple Python client for the BreCal REST API."""
+
+from .client import BreCalClient, DEFAULT_BASE_URL
+from .config import ClientConfig, get_default_base_url
+from .credentials import Credentials
+from .exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    BreCalApiError,
+    ClientConfigurationError,
+)
+from .types import LoginResult
+
+__all__ = [
+    "BreCalClient",
+    "Credentials",
+    "ClientConfig",
+    "get_default_base_url",
+    "LoginResult",
+    "DEFAULT_BASE_URL",
+    "BreCalApiError",
+    "AuthenticationError",
+    "AuthorizationError",
+    "ClientConfigurationError",
+]

src/brecal_api_client/client.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Iterable, Mapping, Optional, Sequence
+from urllib.parse import urljoin
+
+import requests
+import time
+
+from .config import get_default_base_url
+from .credentials import Credentials
+from .exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    BreCalApiError,
+    ClientConfigurationError,
+)
+from .types import JsonDict, LoginResult, MutableJsonDict
+
+DEFAULT_BASE_URL = get_default_base_url()
+
+
+@dataclass
+class _RequestContext:
+    method: str
+    path: str
+    expected: Sequence[int]
+    auth: bool
+
+
+class BreCalClient:
+    """Thin convenience wrapper around the BreCal REST API."""
+
+    def __init__(
+        self,
+        base_url: Optional[str] = None,
+        *,
+        credentials: Optional[Credentials] = None,
+        timeout: float = 30.0,
+        session: Optional[requests.Session] = None,
+        auto_login: bool = True,
+    ) -> None:
+        resolved_base_url = base_url or get_default_base_url()
+        if not resolved_base_url:
+            raise ClientConfigurationError("base_url must be provided.")
+        self.base_url = resolved_base_url.rstrip("/")
+        self._timeout = timeout
+        self._session = session or requests.Session()
+        self._credentials = credentials
+        self._login: Optional[LoginResult] = None
+        if auto_login and credentials is not None:
+            self.login(credentials)
+
+    # -----------------------------------------------------
+    # lifecycle helpers
+    # -----------------------------------------------------
+    def close(self) -> None:
+        self._session.close()
+
+    def __enter__(self) -> "BreCalClient":
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> None:
+        self.close()
+
+    # -----------------------------------------------------
+    # authentication
+    # -----------------------------------------------------
+    @property
+    def token(self) -> Optional[str]:
+        return self._login.token if self._login else None
+
+    @property
+    def login_info(self) -> Optional[LoginResult]:
+        return self._login
+
+    def ensure_authenticated(self) -> None:
+        if self._login and self._login.expires_at.timestamp() > _epoch_seconds() + 30:
+            return
+        if not self._credentials:
+            raise AuthenticationError(
+                "Client has no stored credentials. Call login() with credentials first."
+            )
+        self.login(self._credentials)
+
+    def login(self, credentials: Credentials) -> LoginResult:
+        payload = {"username": credentials.username, "password": credentials.password}
+        data = self._request_json(
+            _RequestContext("POST", "/login", expected=(200,), auth=False),
+            json=payload,
+        )
+        if not isinstance(data, Mapping):
+            raise AuthenticationError("Login returned unexpected payload.")
+        result = LoginResult.from_api(data)
+        if not result.token:
+            raise AuthenticationError("Login response did not include a token.")
+        self._login = result
+        self._credentials = credentials
+        return result
+
+    # -----------------------------------------------------
+    # shipcalls
+    # -----------------------------------------------------
+    def get_shipcalls(self, *, past_days: Optional[int] = None) -> Sequence[JsonDict]:
+        params: Dict[str, Any] = {}
+        if past_days is not None:
+            params["past_days"] = int(past_days)
+        data = self._request_json(
+            _RequestContext("GET", "/shipcalls", expected=(200,), auth=True),
+            params=params or None,
+        )
+        return _as_sequence_of_dicts(data)
+
+    def create_shipcall(self, shipcall: Mapping[str, Any]) -> int:
+        payload = _copy_without_keys(shipcall, drop_keys=("id",))
+        data = self._request_json(
+            _RequestContext("POST", "/shipcalls", expected=(201,), auth=True),
+            json=payload,
+        )
+        return _extract_id(data)
+
+    def update_shipcall(self, shipcall: Mapping[str, Any]) -> int:
+        if "id" not in shipcall:
+            raise ValueError("Shipcall update requires an 'id' field.")
+        data = self._request_json(
+            _RequestContext("PUT", "/shipcalls", expected=(200,), auth=True),
+            json=dict(shipcall),
+        )
+        return _extract_id(data)
+
+    # -----------------------------------------------------
+    # times
+    # -----------------------------------------------------
+    def get_times(self, *, shipcall_id: Optional[int] = None) -> Sequence[JsonDict]:
+        params = {"shipcall_id": int(shipcall_id)} if shipcall_id is not None else None
+        data = self._request_json(
+            _RequestContext("GET", "/times", expected=(200,), auth=True),
+            params=params,
+        )
+        return _as_sequence_of_dicts(data)
+
+    def create_times(self, entry: Mapping[str, Any]) -> int:
+        payload = _copy_without_keys(entry, drop_keys=("id",))
+        data = self._request_json(
+            _RequestContext("POST", "/times", expected=(201,), auth=True),
+            json=payload,
+        )
+        return _extract_id(data)
+
+    def update_times(self, entry: Mapping[str, Any]) -> int:
+        if "id" not in entry:
+            raise ValueError("Times update requires an 'id' field.")
+        data = self._request_json(
+            _RequestContext("PUT", "/times", expected=(200,), auth=True),
+            json=dict(entry),
+        )
+        return _extract_id(data)
+
+    def delete_times(self, times_id: int) -> int:
+        data = self._request_json(
+            _RequestContext("DELETE", "/times", expected=(200,), auth=True),
+            params={"id": int(times_id)},
+        )
+        return _extract_id(data)
+
+    # -----------------------------------------------------
+    # generic helpers
+    # -----------------------------------------------------
+    def raw_request(
+        self,
+        method: str,
+        path: str,
+        *,
+        expected: Sequence[int] = (200,),
+        auth: bool = True,
+        **kwargs: Any,
+    ) -> Any:
+        """Expose the low-level request helper for endpoints not wrapped yet."""
+        ctx = _RequestContext(method.upper(), path, expected, auth)
+        return self._request_json(ctx, **kwargs)
+
+    def _request_json(self, ctx: _RequestContext, **kwargs: Any) -> Any:
+        url = urljoin(f"{self.base_url}/", ctx.path.lstrip("/"))
+        headers: Dict[str, str] = kwargs.pop("headers", {})
+        headers.setdefault("Accept", "application/json")
+        if "json" in kwargs:
+            headers.setdefault("Content-Type", "application/json")
+        if ctx.auth:
+            self.ensure_authenticated()
+            headers.setdefault("Authorization", f"Bearer {self.token}")
+
+        response = self._session.request(
+            ctx.method,
+            url,
+            timeout=self._timeout,
+            headers=headers,
+            **kwargs,
+        )
+        if response.status_code == 401 or response.status_code == 403:
+            raise AuthorizationError(
+                f"{ctx.method} {ctx.path} returned {response.status_code}",
+                status_code=response.status_code,
+                payload=_safe_json(response),
+            )
+        if response.status_code not in ctx.expected:
+            raise BreCalApiError(
+                f"{ctx.method} {ctx.path} returned {response.status_code}",
+                status_code=response.status_code,
+                payload=_safe_json(response),
+            )
+        if response.content:
+            return _safe_json(response)
+        return None
+
+
+def _copy_without_keys(
+    data: Mapping[str, Any], *, drop_keys: Iterable[str]
+) -> MutableJsonDict:
+    payload: MutableJsonDict = dict(data)
+    for key in drop_keys:
+        payload.pop(key, None)
+    return payload
+
+
+def _extract_id(payload: Any) -> int:
+    if isinstance(payload, Mapping) and "id" in payload:
+        return int(payload["id"])
+    raise BreCalApiError("API response did not include an 'id' field.", payload=payload)
+
+
+def _as_sequence_of_dicts(data: Any) -> Sequence[JsonDict]:
+    if isinstance(data, list):
+        return data
+    raise BreCalApiError("Expected list response from API.", payload=data)
+
+
+def _safe_json(response: requests.Response) -> Any:
+    content_type = response.headers.get("Content-Type", "")
+    if "application/json" in content_type:
+        try:
+            return response.json()
+        except ValueError:
+            pass
+    return response.text
+
+
+def _epoch_seconds() -> int:
+    return int(time.time())

src/brecal_api_client/config.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping, Optional, Union
+
+from .exceptions import ClientConfigurationError
+
+ConfigPath = Union[str, Path]
+
+DEFAULT_BASE_URL_FALLBACK = "https://brecaldevel.bsmd-emswe.eu"
+CONFIG_FILENAME = "client.json"
+
+
+def _default_config_path() -> Path:
+    xdg = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
+    return (xdg / "brecal" / CONFIG_FILENAME).expanduser()
+
+
+@dataclass(frozen=True)
+class ClientConfig:
+    base_url: str
+    environment: Optional[str] = None
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any]) -> "ClientConfig":
+        environment = data.get("environment")
+        base_url = data.get("base_url")
+        endpoints = data.get("endpoints")
+
+        if isinstance(endpoints, Mapping):
+            if environment and environment in endpoints:
+                base_url = endpoints[environment]
+            elif not base_url and endpoints:
+                # Pick the first entry as a last resort
+                _, base_url = next(iter(endpoints.items()))
+
+        if not base_url:
+            raise ClientConfigurationError(
+                "Client configuration requires either 'base_url' or an "
+                "'endpoints' mapping."
+            )
+
+        return cls(
+            base_url=str(base_url).rstrip("/"),
+            environment=str(environment) if environment else None,
+        )
+
+    @classmethod
+    def load(cls, path: Optional[ConfigPath] = None) -> "ClientConfig":
+        file_path = Path(path) if path else _default_config_path()
+        data = json.loads(file_path.read_text(encoding="utf-8"))
+        return cls.from_mapping(data)
+
+
+def get_default_base_url(path: Optional[ConfigPath] = None) -> str:
+    """Resolve the default base URL using env vars or ~/.config/brecal/client.json."""
+    env_override = os.getenv("BRECAL_BASE_URL")
+    if env_override:
+        return env_override.rstrip("/")
+
+    try:
+        config = ClientConfig.load(path=path)
+        return config.base_url
+    except FileNotFoundError:
+        return DEFAULT_BASE_URL_FALLBACK
+    except ClientConfigurationError:
+        raise
+    except Exception as exc:
+        raise ClientConfigurationError(
+            f"Failed to load BreCal client configuration: {exc}"
+        ) from exc

src/brecal_api_client/credentials.py

@@ -0,0 +1,68 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping, Optional, Union
+
+ConfigPath = Union[str, Path]
+
+
+def _default_credentials_path() -> Path:
+    """Return the default path for the credential file."""
+    xdg = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
+    return (xdg / "brecal" / "credentials.json").expanduser()
+
+
+@dataclass(frozen=True)
+class Credentials:
+    """Holds username/password pairs for the BreCal API."""
+
+    username: str
+    password: str
+
+    @classmethod
+    def from_mapping(cls, data: Mapping[str, Any]) -> "Credentials":
+        """Create credentials from a mapping (dict, TOML config, etc.)."""
+        username = _coalesce_key(
+            data, ("username", "user_name", "user"), required="username"
+        )
+        password = _coalesce_key(
+            data, ("password", "pass", "secret"), required="password"
+        )
+        if not isinstance(username, str) or not username.strip():
+            raise ValueError("BreCal credentials require a non-empty username.")
+        if not isinstance(password, str) or not password:
+            raise ValueError("BreCal credentials require a non-empty password.")
+        return cls(username=username.strip(), password=password)
+
+    @classmethod
+    def load(cls, path: Optional[ConfigPath] = None) -> "Credentials":
+        """Load credentials from a JSON file."""
+        file_path = Path(path) if path else _default_credentials_path()
+        text = file_path.read_text(encoding="utf-8")
+        data = json.loads(text)
+        return cls.from_mapping(data)
+
+    @classmethod
+    def from_env(
+        cls, username_var: str = "BRECAL_USERNAME", password_var: str = "BRECAL_PASSWORD"
+    ) -> "Credentials":
+        """Load credentials from environment variables."""
+        username = os.getenv(username_var)
+        password = os.getenv(password_var)
+        if not username or not password:
+            raise EnvironmentError(
+                f"Missing credentials in env vars {username_var}/{password_var}"
+            )
+        return cls(username=username, password=password)
+
+
+def _coalesce_key(
+    data: Mapping[str, Any], keys: tuple[str, ...], *, required: str
+) -> Any:
+    for key in keys:
+        if key in data:
+            return data[key]
+    raise KeyError(f"Missing '{required}' in credentials mapping.")

src/brecal_api_client/exceptions.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from typing import Any, Optional
+
+
+class BreCalApiError(RuntimeError):
+    """Base exception for API client failures."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        status_code: Optional[int] = None,
+        payload: Optional[Any] = None,
+    ) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.payload = payload
+
+
+class AuthenticationError(BreCalApiError):
+    """Raised when login fails."""
+
+
+class AuthorizationError(BreCalApiError):
+    """Raised for 401/403 responses after authentication."""
+
+
+class ClientConfigurationError(ValueError):
+    """Raised for invalid client configuration or missing dependencies."""

src/brecal_api_client/types.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Dict, Mapping, MutableMapping, Optional
+
+JsonDict = Dict[str, Any]
+MutableJsonDict = MutableMapping[str, Any]
+
+
+@dataclass
+class LoginResult:
+    """Represents the payload returned by /login."""
+
+    id: int
+    participant_id: Optional[int]
+    first_name: str
+    last_name: str
+    user_name: str
+    user_email: Optional[str]
+    user_phone: Optional[str]
+    token: str
+    exp: int
+
+    @classmethod
+    def from_api(cls, data: Mapping[str, Any]) -> "LoginResult":
+        return cls(
+            id=_coerce_int(data.get("id")),
+            participant_id=_coerce_optional_int(data.get("participant_id")),
+            first_name=str(data.get("first_name") or ""),
+            last_name=str(data.get("last_name") or ""),
+            user_name=str(data.get("user_name") or ""),
+            user_email=_coerce_optional_str(data.get("user_email")),
+            user_phone=_coerce_optional_str(data.get("user_phone")),
+            token=str(data.get("token") or ""),
+            exp=_coerce_int(data.get("exp")),
+        )
+
+    @property
+    def expires_at(self) -> datetime:
+        return datetime.fromtimestamp(self.exp, tz=timezone.utc)
+
+
+def _coerce_int(value: Any) -> int:
+    if value is None:
+        raise ValueError("Expected integer value, got None")
+    return int(value)
+
+
+def _coerce_optional_int(value: Any) -> Optional[int]:
+    return None if value is None else int(value)
+
+
+def _coerce_optional_str(value: Any) -> Optional[str]:
+    if value is None:
+        return None
+    text = str(value)
+    return text if text else None