unitelabs.sdk.client

Attributes

Name
__all__
Type
Value
= [ "Client" ]
Description

Classes

Client
MRO
BaseClient
⎿⎽SubscriptionsClient
⎿⎽⎽ActionsClient
⎿⎽⎽⎽ModulesClient
⎿⎽⎽⎽⎽ServicesClient
⎿⎽⎽⎽⎽⎽Client
Methods
- list_services(self) -> list[Service]
  Response
  Type
  list[Service]
  Description
- get_service(self, service_id : str) -> Service
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    service_id
    Type
    str
    Default
    Description
  Response
  Type
  Service
  Description
- get_service_by_name(self, name : str) -> Service[]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  Service[]
  Description
- list_modules(self, service_id : str) -> list[Module]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    service_id
    Type
    str
    Default
    Description
  Response
  Type
  list[Module]
  Description
- get_module(self, module_id : str) -> Module
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    module_id
    Type
    str
    Default
    Description
  Response
  Type
  Module
  Description
- list_actions(self, module_id : str) -> list[Action]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    module_id
    Type
    str
    Default
    Description
  Response
  Type
  list[Action]
  Description
- get_action(self, action_id : str) -> Action
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    action_id
    Type
    str
    Default
    Description
  Response
  Type
  Action
  Description
- get_property(self, property_id : str) -> Property
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    property_id
    Type
    str
    Default
    Description
  Response
  Type
  Property
  Description
- get_sensor(self, sensor_id : str) -> Sensor
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    sensor_id
    Type
    str
    Default
    Description
  Response
  Type
  Sensor
  Description
- get_control(self, control_id : str) -> Control
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    control_id
    Type
    str
    Default
    Description
  Response
  Type
  Control
  Description
- list_subscriptions(self) -> list[SubscriptionResponse]
  Response
  Type
  list[SubscriptionResponse]
  Description
- create_subscription(
  self,
  action_id : str,
  parameters : dict,
  retry : bool,
  retry_delay : float,
  timeouts : int[][],
  interval : int[][]
  ) -> Subscription
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    action_id
    Type
    str
    Default
    Description
  - Name
    parameters
    Type
    dict
    Default
    Description
  - Name
    retry
    Type
    bool
    Default
    = False
    Description
  - Name
    retry_delay
    Type
    float
    Default
    = 1.0
    Description
  - Name
    timeouts
    Type
    int[][]
    Default
    = None
    Description
  - Name
    interval
    Type
    int[][]
    Default
    = None
    Description
  Response
  Type
  Subscription
  Description
- delete_subscription(self, subscription_id : str) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    subscription_id
    Type
    str
    Default
    Description
- init(
  self,
  base_url : str[],
  auth_url : str[],
  client_id : str[],
  client_secret : str[],
  http_client : typing.Optional[httpx.AsyncClient],
  **kwargs
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    base_url
    Type
    str[]
    Default
    = None
    Description
  - Name
    auth_url
    Type
    str[]
    Default
    = None
    Description
  - Name
    client_id
    Type
    str[]
    Default
    = None
    Description
  - Name
    client_secret
    Type
    str[]
    Default
    = None
    Description
  - Name
    http_client
    Type
    typing.Optional[httpx.AsyncClient]
    Default
    = None
    Description
  - Name
    **kwargs
    Type
    Default
    = {}
    Description
- health(self) -> None
- get(
  self,
  url,
  *args,
  **kwargs
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    url
    Type
    Default
    Description
  - Name
    *args
    Type
    Default
    = ()
    Description
  - Name
    **kwargs
    Type
    Default
    = {}
    Description
- post(
  self,
  url,
  *args,
  **kwargs
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    url
    Type
    Default
    Description
  - Name
    *args
    Type
    Default
    = ()
    Description
  - Name
    **kwargs
    Type
    Default
    = {}
    Description
- delete(
  self,
  url,
  *args,
  **kwargs
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    url
    Type
    Default
    Description
  - Name
    *args
    Type
    Default
    = ()
    Description
  - Name
    **kwargs
    Type
    Default
    = {}
    Description
- stream(
  self,
  url,
  cancel : asyncio.Event,
  **kwargs
  ) -> collections.abc.AsyncIterator[tuple[str, typing.Any]]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    url
    Type
    Default
    Description
  - Name
    cancel
    Type
    asyncio.Event
    Default
    Description
  - Name
    **kwargs
    Type
    Default
    = {}
    Description
  Response
  Type
  collections.abc.AsyncIterator[tuple[str, typing.Any]]
  Description
- aclose(self) -> None
  Close underlying connection.
- __aenter__(self) -> None
- aexit(
  self,
  exc_type : typing.Optional[typing.Type[BaseException]],
  exc_value : typing.Optional[BaseException],
  traceback : typing.Optional[types.TracebackType]
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    exc_type
    Type
    typing.Optional[typing.Type[BaseException]]
    Default
    = None
    Description
  - Name
    exc_value
    Type
    typing.Optional[BaseException]
    Default
    = None
    Description
  - Name
    traceback
    Type
    typing.Optional[types.TracebackType]
    Default
    = None
    Description
Attributes
- Name
  base_url
  Type
  Value
  = base_url.rstrip('/') + '/v1'
  Description
- Name
  auth_url
  Type
  Value
  = auth_url.rstrip('/')
  Description
- Name
  client_id
  Type
  Value
  = client_id
  Description
- Name
  client_secret
  Type
  Value
  = client_secret
  Description
- Name
  _session
  Type
  httpx.AsyncClient
  Value
  = http_client or httpx.AsyncClient(base_url=self.base_url, auth=OAuth2(f'{self.auth_url}/token', client_id=self.client_id, client_secret=self.client_secret), **kwargs)
  Description
- Name
  kwargs
  Type
  Value
  = kwargs
  Description
- Name
  is_closed
  Type
  bool
  Value
  = None
  Description

unitelabs.sdk.client

Packages

Attributes

Classes

Client

Methods

list_services(self) -> list[Service]

Response

get_service(self, service_id : str) -> Service

Parameters

Response

get_service_by_name(self, name : str) -> Service[]

Parameters

Response

list_modules(self, service_id : str) -> list[Module]

Parameters

Response

get_module(self, module_id : str) -> Module

Parameters

Response

list_actions(self, module_id : str) -> list[Action]

Parameters

Response

get_action(self, action_id : str) -> Action

Parameters

Response

get_property(self, property_id : str) -> Property

Parameters

Response

get_sensor(self, sensor_id : str) -> Sensor

Parameters

Response

get_control(self, control_id : str) -> Control

Parameters

Response

list_subscriptions(self) -> list[SubscriptionResponse]

Response

create_subscription(self, action_id : str, parameters : dict, retry : bool, retry_delay : float, timeouts : int[][], interval : int[][]) -> Subscription

Parameters

Response

delete_subscription(self, subscription_id : str) -> None

Parameters

__init__(self, base_url : str[], auth_url : str[], client_id : str[], client_secret : str[], http_client : typing.Optional[httpx.AsyncClient], **kwargs) -> None

Parameters

health(self) -> None

get(self, url, *args, **kwargs) -> None

Parameters

post(self, url, *args, **kwargs) -> None

Parameters

delete(self, url, *args, **kwargs) -> None

Parameters

stream(self, url, cancel : asyncio.Event, **kwargs) -> collections.abc.AsyncIterator[tuple[str, typing.Any]]

Parameters

Response

aclose(self) -> None

__aenter__(self) -> None

__aexit__(self, exc_type : typing.Optional[typing.Type[BaseException]], exc_value : typing.Optional[BaseException], traceback : typing.Optional[types.TracebackType]) -> None

Parameters

Attributes

create_subscription(
self,
action_id : str,
parameters : dict,
retry : bool,
retry_delay : float,
timeouts : int[][],
interval : int[][]
) -> Subscription

init(
self,
base_url : str[],
auth_url : str[],
client_id : str[],
client_secret : str[],
http_client : typing.Optional[httpx.AsyncClient],
**kwargs
) -> None

get(
self,
url,
*args,
**kwargs
) -> None

post(
self,
url,
*args,
**kwargs
) -> None

delete(
self,
url,
*args,
**kwargs
) -> None

stream(
self,
url,
cancel : asyncio.Event,
**kwargs
) -> collections.abc.AsyncIterator[tuple[str, typing.Any]]

aenter(self) -> None

aexit(
self,
exc_type : typing.Optional[typing.Type[BaseException]],
exc_value : typing.Optional[BaseException],
traceback : typing.Optional[types.TracebackType]
) -> None