database: transition to get_device_shared_secret

doc/APIClientRegen.md

@@ -1,11 +1,4 @@
 # Regenerating the API Client
 
 1. Download the latest API specification from https://api.infuse-iot.com/docs
-2. Delete the previous API client: `rm -r ./src/infuse-iot/api_client`
-3. Generate API client into the root directory: `openapi-python-client generate --path ./infuse-api.yaml`
-4. Move API client to desired directory: `mv infuse-api-client/infuse_api_client/ ./src/infuse_iot/api_client/`
-5. Move README: `mv infuse-api-client/README.md ./src/infuse_iot/api_client/`
-6. Remove extraneous files: `rm -r infuse-api-client`
-7. Manually fixup `README.md` for naming
-
-Some of these steps can possibly be automated with the `openapi-python-client` `--config` parameter in the future.
+2. Run the regeneration script `./scripts/regenrate_api_client.py /path/to/infuse-api.yaml`

scripts/regenerate_api_client.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""Regenerate the Infuse-IoT OpenAPI client."""
+
+from __future__ import annotations
+
+import argparse
+import shutil
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+TARGET_CLIENT = Path("src/infuse_iot/api_client")
+GENERATED_PACKAGE = "infuse_api_client"
+
+
+def repo_root() -> Path:
+    return Path(__file__).resolve().parents[1]
+
+
+def validate_spec_path(spec_path: Path) -> None:
+    if not spec_path.exists():
+        raise RuntimeError(f"API specification does not exist: {spec_path}")
+    if not spec_path.is_file():
+        raise RuntimeError(f"API specification is not a file: {spec_path}")
+
+
+def run_generator(spec_path: Path, staging_dir: Path) -> Path:
+    command = ["openapi-python-client", "generate", "--path", str(spec_path)]
+    print(f"Generating API client in {staging_dir}")
+    try:
+        subprocess.run(command, cwd=staging_dir, check=True)
+    except FileNotFoundError as exc:
+        raise RuntimeError("openapi-python-client was not found. Install it and rerun this script.") from exc
+    except subprocess.CalledProcessError as exc:
+        raise RuntimeError(f"openapi-python-client failed with exit code {exc.returncode}") from exc
+
+    matches = sorted(staging_dir.glob(f"*/{GENERATED_PACKAGE}"))
+    if not matches:
+        raise RuntimeError(f"generated package {GENERATED_PACKAGE!r} was not found under {staging_dir}")
+    if len(matches) > 1:
+        raise RuntimeError(
+            "multiple generated client packages were found: " + ", ".join(str(match) for match in matches)
+        )
+    return matches[0]
+
+
+def replace_client(generated_client: Path, target_client: Path) -> None:
+    readme = None
+    readme_path = target_client / "README.md"
+    if readme_path.exists():
+        readme = readme_path.read_bytes()
+
+    target_parent = target_client.parent
+    with tempfile.TemporaryDirectory(prefix="api-client-backup-", dir=target_parent) as backup:
+        backup_client = Path(backup) / target_client.name
+        if target_client.exists():
+            shutil.move(str(target_client), backup_client)
+
+        try:
+            shutil.move(str(generated_client), target_client)
+            if readme is not None:
+                (target_client / "README.md").write_bytes(readme)
+        except Exception:
+            if target_client.exists():
+                shutil.rmtree(target_client)
+            if backup_client.exists():
+                shutil.move(str(backup_client), target_client)
+            raise
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Regenerate src/infuse_iot/api_client from an OpenAPI YAML file.")
+    parser.add_argument(
+        "spec_path",
+        help="Path to the downloaded OpenAPI YAML file.",
+    )
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv if argv is not None else sys.argv[1:])
+    root = repo_root()
+    spec_path = Path(args.spec_path).expanduser()
+    if not spec_path.is_absolute():
+        spec_path = root / spec_path
+    spec_path = spec_path.resolve()
+    target_client = root / TARGET_CLIENT
+
+    try:
+        validate_spec_path(spec_path)
+
+        with tempfile.TemporaryDirectory(prefix="api-client-gen-", dir=root) as staging:
+            generated_client = run_generator(spec_path, Path(staging))
+            replace_client(generated_client, target_client)
+    except RuntimeError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+
+    print(f"Regenerated {target_client.relative_to(root)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/infuse_iot/api_client/api/key/get_device_shared_secret.py

@@ -0,0 +1,156 @@
+from http import HTTPStatus
+from typing import Any
+
+import httpx
+
+from ... import errors
+from ...client import AuthenticatedClient, Client
+from ...models.get_device_shared_secret_body import GetDeviceSharedSecretBody
+from ...models.key import Key
+from ...types import Response
+
+
+def _get_kwargs(
+    *,
+    body: GetDeviceSharedSecretBody,
+) -> dict[str, Any]:
+    headers: dict[str, Any] = {}
+
+    _kwargs: dict[str, Any] = {
+        "method": "post",
+        "url": "/key/sharedSecret/device",
+    }
+
+    _kwargs["json"] = body.to_dict()
+
+    headers["Content-Type"] = "application/json"
+
+    _kwargs["headers"] = headers
+    return _kwargs
+
+
+def _parse_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Key | None:
+    if response.status_code == 200:
+        response_200 = Key.from_dict(response.json())
+
+        return response_200
+
+    if client.raise_on_unexpected_status:
+        raise errors.UnexpectedStatus(response.status_code, response.content)
+    else:
+        return None
+
+
+def _build_response(*, client: AuthenticatedClient | Client, response: httpx.Response) -> Response[Key]:
+    return Response(
+        status_code=HTTPStatus(response.status_code),
+        content=response.content,
+        headers=response.headers,
+        parsed=_parse_response(client=client, response=response),
+    )
+
+
+def sync_detailed(
+    *,
+    client: AuthenticatedClient | Client,
+    body: GetDeviceSharedSecretBody,
+) -> Response[Key]:
+    """Get a device's shared secret key
+
+    Args:
+        body (GetDeviceSharedSecretBody):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Key]
+    """
+
+    kwargs = _get_kwargs(
+        body=body,
+    )
+
+    response = client.get_httpx_client().request(
+        **kwargs,
+    )
+
+    return _build_response(client=client, response=response)
+
+
+def sync(
+    *,
+    client: AuthenticatedClient | Client,
+    body: GetDeviceSharedSecretBody,
+) -> Key | None:
+    """Get a device's shared secret key
+
+    Args:
+        body (GetDeviceSharedSecretBody):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Key
+    """
+
+    return sync_detailed(
+        client=client,
+        body=body,
+    ).parsed
+
+
+async def asyncio_detailed(
+    *,
+    client: AuthenticatedClient | Client,
+    body: GetDeviceSharedSecretBody,
+) -> Response[Key]:
+    """Get a device's shared secret key
+
+    Args:
+        body (GetDeviceSharedSecretBody):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Response[Key]
+    """
+
+    kwargs = _get_kwargs(
+        body=body,
+    )
+
+    response = await client.get_async_httpx_client().request(**kwargs)
+
+    return _build_response(client=client, response=response)
+
+
+async def asyncio(
+    *,
+    client: AuthenticatedClient | Client,
+    body: GetDeviceSharedSecretBody,
+) -> Key | None:
+    """Get a device's shared secret key
+
+    Args:
+        body (GetDeviceSharedSecretBody):
+
+    Raises:
+        errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
+        httpx.TimeoutException: If the request takes longer than Client.timeout.
+
+    Returns:
+        Key
+    """
+
+    return (
+        await asyncio_detailed(
+            client=client,
+            body=body,
+        )
+    ).parsed

src/infuse_iot/api_client/models/__init__.py

@@ -82,6 +82,7 @@
 from .generate_mqtt_token_body import GenerateMQTTTokenBody
 from .generated_api_key import GeneratedAPIKey
 from .generated_mqtt_token import GeneratedMQTTToken
+from .get_device_shared_secret_body import GetDeviceSharedSecretBody
 from .get_last_routes_for_devices_body import GetLastRoutesForDevicesBody
 from .health_check import HealthCheck
 from .interface_data import InterfaceData
@@ -208,6 +209,7 @@
     "GeneratedAPIKey",
     "GeneratedMQTTToken",
     "GenerateMQTTTokenBody",
+    "GetDeviceSharedSecretBody",
     "GetLastRoutesForDevicesBody",
     "HealthCheck",
     "InterfaceData",

src/infuse_iot/api_client/models/get_device_shared_secret_body.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import TYPE_CHECKING, Any, TypeVar
+
+from attrs import define as _attrs_define
+from attrs import field as _attrs_field
+
+from ..types import UNSET, Unset
+
+if TYPE_CHECKING:
+    from ..models.security_state import SecurityState
+
+
+T = TypeVar("T", bound="GetDeviceSharedSecretBody")
+
+
+@_attrs_define
+class GetDeviceSharedSecretBody:
+    """
+    Attributes:
+        device_id (str): The ID of the device as a hex string Example: d291d4d66bf0a955.
+        security_state (SecurityState | Unset):
+    """
+
+    device_id: str
+    security_state: SecurityState | Unset = UNSET
+    additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        device_id = self.device_id
+
+        security_state: dict[str, Any] | Unset = UNSET
+        if not isinstance(self.security_state, Unset):
+            security_state = self.security_state.to_dict()
+
+        field_dict: dict[str, Any] = {}
+        field_dict.update(self.additional_properties)
+        field_dict.update(
+            {
+                "deviceId": device_id,
+            }
+        )
+        if security_state is not UNSET:
+            field_dict["securityState"] = security_state
+
+        return field_dict
+
+    @classmethod
+    def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T:
+        from ..models.security_state import SecurityState
+
+        d = dict(src_dict)
+        device_id = d.pop("deviceId")
+
+        _security_state = d.pop("securityState", UNSET)
+        security_state: SecurityState | Unset
+        if isinstance(_security_state, Unset):
+            security_state = UNSET
+        else:
+            security_state = SecurityState.from_dict(_security_state)
+
+        get_device_shared_secret_body = cls(
+            device_id=device_id,
+            security_state=security_state,
+        )
+
+        get_device_shared_secret_body.additional_properties = d
+        return get_device_shared_secret_body
+
+    @property
+    def additional_keys(self) -> list[str]:
+        return list(self.additional_properties.keys())
+
+    def __getitem__(self, key: str) -> Any:
+        return self.additional_properties[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        self.additional_properties[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        del self.additional_properties[key]
+
+    def __contains__(self, key: str) -> bool:
+        return key in self.additional_properties