diff --git a/src/infuse_iot/api_client/api/device/get_device_last_route_by_device_id.py b/src/infuse_iot/api_client/api/device/get_device_last_route_by_device_id.py new file mode 100644 index 0000000..0a58dcc --- /dev/null +++ b/src/infuse_iot/api_client/api/device/get_device_last_route_by_device_id.py @@ -0,0 +1,153 @@ +from http import HTTPStatus +from typing import Any, Optional, Union, cast + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.uplink_route import UplinkRoute +from ...types import Response + + +def _get_kwargs( + device_id: str, +) -> dict[str, Any]: + _kwargs: dict[str, Any] = { + "method": "get", + "url": f"/device/deviceId/{device_id}/lastRoute", + } + + return _kwargs + + +def _parse_response( + *, client: Union[AuthenticatedClient, Client], response: httpx.Response +) -> Optional[Union[Any, UplinkRoute]]: + if response.status_code == 200: + response_200 = UplinkRoute.from_dict(response.json()) + + return response_200 + if response.status_code == 404: + response_404 = cast(Any, None) + return response_404 + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: Union[AuthenticatedClient, Client], response: httpx.Response +) -> Response[Union[Any, UplinkRoute]]: + return Response( + status_code=HTTPStatus(response.status_code), + content=response.content, + headers=response.headers, + parsed=_parse_response(client=client, response=response), + ) + + +def sync_detailed( + device_id: str, + *, + client: Union[AuthenticatedClient, Client], +) -> Response[Union[Any, UplinkRoute]]: + """Get last route by DeviceID + + Args: + device_id (str): + + 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[Union[Any, UplinkRoute]] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + device_id: str, + *, + client: Union[AuthenticatedClient, Client], +) -> Optional[Union[Any, UplinkRoute]]: + """Get last route by DeviceID + + Args: + device_id (str): + + 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: + Union[Any, UplinkRoute] + """ + + return sync_detailed( + device_id=device_id, + client=client, + ).parsed + + +async def asyncio_detailed( + device_id: str, + *, + client: Union[AuthenticatedClient, Client], +) -> Response[Union[Any, UplinkRoute]]: + """Get last route by DeviceID + + Args: + device_id (str): + + 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[Union[Any, UplinkRoute]] + """ + + kwargs = _get_kwargs( + device_id=device_id, + ) + + response = await client.get_async_httpx_client().request(**kwargs) + + return _build_response(client=client, response=response) + + +async def asyncio( + device_id: str, + *, + client: Union[AuthenticatedClient, Client], +) -> Optional[Union[Any, UplinkRoute]]: + """Get last route by DeviceID + + Args: + device_id (str): + + 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: + Union[Any, UplinkRoute] + """ + + return ( + await asyncio_detailed( + device_id=device_id, + client=client, + ) + ).parsed diff --git a/src/infuse_iot/api_client/api/device/get_last_routes_for_devices.py b/src/infuse_iot/api_client/api/device/get_last_routes_for_devices.py new file mode 100644 index 0000000..2ee3a9c --- /dev/null +++ b/src/infuse_iot/api_client/api/device/get_last_routes_for_devices.py @@ -0,0 +1,165 @@ +from http import HTTPStatus +from typing import Any, Optional, Union + +import httpx + +from ... import errors +from ...client import AuthenticatedClient, Client +from ...models.get_last_routes_for_devices_body import GetLastRoutesForDevicesBody +from ...models.uplink_route_and_device_id import UplinkRouteAndDeviceId +from ...types import Response + + +def _get_kwargs( + *, + body: GetLastRoutesForDevicesBody, +) -> dict[str, Any]: + headers: dict[str, Any] = {} + + _kwargs: dict[str, Any] = { + "method": "post", + "url": "/device/lastRoute", + } + + _body = body.to_dict() + + _kwargs["json"] = _body + headers["Content-Type"] = "application/json" + + _kwargs["headers"] = headers + return _kwargs + + +def _parse_response( + *, client: Union[AuthenticatedClient, Client], response: httpx.Response +) -> Optional[list["UplinkRouteAndDeviceId"]]: + if response.status_code == 200: + response_200 = [] + _response_200 = response.json() + for response_200_item_data in _response_200: + response_200_item = UplinkRouteAndDeviceId.from_dict(response_200_item_data) + + response_200.append(response_200_item) + + return response_200 + if client.raise_on_unexpected_status: + raise errors.UnexpectedStatus(response.status_code, response.content) + else: + return None + + +def _build_response( + *, client: Union[AuthenticatedClient, Client], response: httpx.Response +) -> Response[list["UplinkRouteAndDeviceId"]]: + 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: Union[AuthenticatedClient, Client], + body: GetLastRoutesForDevicesBody, +) -> Response[list["UplinkRouteAndDeviceId"]]: + """Get last routes for a group of devices + + Args: + body (GetLastRoutesForDevicesBody): Body for getting last routes for devices + + 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[list['UplinkRouteAndDeviceId']] + """ + + kwargs = _get_kwargs( + body=body, + ) + + response = client.get_httpx_client().request( + **kwargs, + ) + + return _build_response(client=client, response=response) + + +def sync( + *, + client: Union[AuthenticatedClient, Client], + body: GetLastRoutesForDevicesBody, +) -> Optional[list["UplinkRouteAndDeviceId"]]: + """Get last routes for a group of devices + + Args: + body (GetLastRoutesForDevicesBody): Body for getting last routes for devices + + 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: + list['UplinkRouteAndDeviceId'] + """ + + return sync_detailed( + client=client, + body=body, + ).parsed + + +async def asyncio_detailed( + *, + client: Union[AuthenticatedClient, Client], + body: GetLastRoutesForDevicesBody, +) -> Response[list["UplinkRouteAndDeviceId"]]: + """Get last routes for a group of devices + + Args: + body (GetLastRoutesForDevicesBody): Body for getting last routes for devices + + 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[list['UplinkRouteAndDeviceId']] + """ + + 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: Union[AuthenticatedClient, Client], + body: GetLastRoutesForDevicesBody, +) -> Optional[list["UplinkRouteAndDeviceId"]]: + """Get last routes for a group of devices + + Args: + body (GetLastRoutesForDevicesBody): Body for getting last routes for devices + + 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: + list['UplinkRouteAndDeviceId'] + """ + + return ( + await asyncio_detailed( + client=client, + body=body, + ) + ).parsed diff --git a/src/infuse_iot/api_client/models/__init__.py b/src/infuse_iot/api_client/models/__init__.py index 85e0306..58f54e5 100644 --- a/src/infuse_iot/api_client/models/__init__.py +++ b/src/infuse_iot/api_client/models/__init__.py @@ -52,6 +52,7 @@ from .error import Error from .forwarded_downlink_route import ForwardedDownlinkRoute from .forwarded_uplink_route import ForwardedUplinkRoute +from .get_last_routes_for_devices_body import GetLastRoutesForDevicesBody from .health_check import HealthCheck from .interface_data import InterfaceData from .key import Key @@ -74,6 +75,7 @@ from .udp_downlink_route import UdpDownlinkRoute from .udp_uplink_route import UdpUplinkRoute from .uplink_route import UplinkRoute +from .uplink_route_and_device_id import UplinkRouteAndDeviceId __all__ = ( "Algorithm", @@ -128,6 +130,7 @@ "Error", "ForwardedDownlinkRoute", "ForwardedUplinkRoute", + "GetLastRoutesForDevicesBody", "HealthCheck", "InterfaceData", "Key", @@ -150,4 +153,5 @@ "UdpDownlinkRoute", "UdpUplinkRoute", "UplinkRoute", + "UplinkRouteAndDeviceId", ) diff --git a/src/infuse_iot/api_client/models/get_last_routes_for_devices_body.py b/src/infuse_iot/api_client/models/get_last_routes_for_devices_body.py new file mode 100644 index 0000000..483826c --- /dev/null +++ b/src/infuse_iot/api_client/models/get_last_routes_for_devices_body.py @@ -0,0 +1,60 @@ +from collections.abc import Mapping +from typing import Any, TypeVar, cast + +from attrs import define as _attrs_define +from attrs import field as _attrs_field + +T = TypeVar("T", bound="GetLastRoutesForDevicesBody") + + +@_attrs_define +class GetLastRoutesForDevicesBody: + """Body for getting last routes for devices + + Attributes: + device_ids (list[str]): + """ + + device_ids: list[str] + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + device_ids = self.device_ids + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "deviceIds": device_ids, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + d = dict(src_dict) + device_ids = cast(list[str], d.pop("deviceIds")) + + get_last_routes_for_devices_body = cls( + device_ids=device_ids, + ) + + get_last_routes_for_devices_body.additional_properties = d + return get_last_routes_for_devices_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 diff --git a/src/infuse_iot/api_client/models/uplink_route_and_device_id.py b/src/infuse_iot/api_client/models/uplink_route_and_device_id.py new file mode 100644 index 0000000..7cbe884 --- /dev/null +++ b/src/infuse_iot/api_client/models/uplink_route_and_device_id.py @@ -0,0 +1,74 @@ +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 + +if TYPE_CHECKING: + from ..models.uplink_route import UplinkRoute + + +T = TypeVar("T", bound="UplinkRouteAndDeviceId") + + +@_attrs_define +class UplinkRouteAndDeviceId: + """Uplink route with device ID + + Attributes: + device_id (str): 8 byte DeviceID as a hex string Example: d291d4d66bf0a955. + last_route (UplinkRoute): + """ + + device_id: str + last_route: "UplinkRoute" + additional_properties: dict[str, Any] = _attrs_field(init=False, factory=dict) + + def to_dict(self) -> dict[str, Any]: + device_id = self.device_id + + last_route = self.last_route.to_dict() + + field_dict: dict[str, Any] = {} + field_dict.update(self.additional_properties) + field_dict.update( + { + "deviceId": device_id, + "lastRoute": last_route, + } + ) + + return field_dict + + @classmethod + def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: + from ..models.uplink_route import UplinkRoute + + d = dict(src_dict) + device_id = d.pop("deviceId") + + last_route = UplinkRoute.from_dict(d.pop("lastRoute")) + + uplink_route_and_device_id = cls( + device_id=device_id, + last_route=last_route, + ) + + uplink_route_and_device_id.additional_properties = d + return uplink_route_and_device_id + + @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