unitelabs.sdk.connect

Attributes

Name
__all__
Type
list[str]
Value
= None
Description

Functions

__call__(uuid : str[], name : str[]) -> Service
Import a Service by its uuid or name
Parameters
- Name
  uuid
  Type
  str[]
  Default
  = None
  Description
- Name
  name
  Type
  str[]
  Default
  = None
  Description
Response
Type
Service
Description
__getattr__(name : str) -> type[Service]
Import a Service by its name
Parameters
- Name
  name
  Type
  str
  Default
  Description
Response
Type
type[Service]
Description
__getitem__(name : str) -> type[Service]
Access a Service by its name
Parameters
- Name
  name
  Type
  str
  Default
  Description
Response
Type
type[Service]
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
ClientError
Base class for all exceptions that may occur when communicating with a connector.
Bases
Exception
Service
Decorators
dataclasses.dataclass
Methods
- init(
  self,
  client : Client,
  id : str,
  name : str,
  modules : dict[str, Module]
  ) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    client
    Type
    Client
    Default
    Description
  - Name
    id
    Type
    str
    Default
    = ''
    Description
  - Name
    name
    Type
    str
    Default
    = ''
    Description
  - Name
    modules
    Type
    dict[str, Module]
    Default
    Description
- __getattr__(self, name : str) -> Module
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  Module
  Description
- __getitem__(self, name : str) -> Module
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  Module
  Description
Attributes
- Name
  client
  Type
  Client
  Value
  = None
  Description
- Name
  id
  Type
  str
  Value
  = ''
  Description
- Name
  name
  Type
  str
  Value
  = ''
  Description
- Name
  modules
  Type
  dict[str, Module]
  Value
  = dataclasses.field(repr=False, default_factory=dict)
  Description
ConnectModule
Connect to the devices in your lab.
Bases
types.ModuleType
Methods
- __init__(self, name : str) -> None
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
- init(self) -> None
- __call__(self, uuid : str[], name : str[]) -> Service
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    uuid
    Type
    str[]
    Default
    = None
    Description
  - Name
    name
    Type
    str[]
    Default
    = None
    Description
  Response
  Type
  Service
  Description
- __getattr__(self, name : str) -> type[Service]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  type[Service]
  Description
- __getitem__(self, name : str) -> type[Service]
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  type[Service]
  Description
- __dir__(self) -> typing.Iterable[str]
  Response
  Type
  typing.Iterable[str]
  Description
Attributes
- Name
  __cache
  Type
  dict[str, Service[]]
  Value
  = {}
  Description
- Name
  logger
  Type
  logging.Logger
  Value
  = None
  Description
- Name
  __all__
  Type
  typing.Iterable[str]
  Value
  = None
  Description
UniteLabsExample
A UniteLabs SiLA Python Example Server
MRO
Service
⎿⎽UniteLabsExample
Methods
- __init__(self) -> None
- __getattr__(self, name : str) -> Module
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  Module
  Description
- __getitem__(self, name : str) -> Module
  Parameters
  - Name
    self
    Type
    Default
    Description
  - Name
    name
    Type
    str
    Default
    Description
  Response
  Type
  Module
  Description
Attributes
- Name
  sila_service
  Type
  Module
  Value
  = None
  Description
- Name
  observable_command_test
  Type
  ObservableCommandTest
  Value
  = None
  Description
- Name
  observable_property_test
  Type
  ObservablePropertyTest
  Value
  = None
  Description
- Name
  unobservable_command_test
  Type
  UnobservableCommandTest
  Value
  = None
  Description
- Name
  unobservable_property_test
  Type
  UnobservablePropertyTest
  Value
  = None
  Description
- Name
  greeting_provider
  Type
  Module
  Value
  = None
  Description
- Name
  client
  Type
  Client
  Value
  = None
  Description
- Name
  id
  Type
  str
  Value
  = ''
  Description
- Name
  name
  Type
  str
  Value
  = ''
  Description
- Name
  modules
  Type
  dict[str, Module]
  Value
  = dataclasses.field(repr=False, default_factory=dict)
  Description

unitelabs.sdk.connect

Packages

Attributes

Functions

__call__(uuid : str[], name : str[]) -> Service

Parameters

Response

__getattr__(name : str) -> type[Service]

Parameters

Response

__getitem__(name : str) -> type[Service]

Parameters

Response

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

ClientError

Service

Methods

__init__(self, client : Client, id : str, name : str, modules : dict[str, Module]) -> None

Parameters

__getattr__(self, name : str) -> Module

Parameters

Response

__getitem__(self, name : str) -> Module

Parameters

Response

call(uuid : str[], name : str[]) -> Service

getattr(name : str) -> type[Service]

getitem(name : str) -> type[Service]

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

init(
self,
client : Client,
id : str,
name : str,
modules : dict[str, Module]
) -> None

getattr(self, name : str) -> Module

getitem(self, name : str) -> Module

init(self, name : str) -> None

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

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

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

dir(self) -> typing.Iterable[str]

init(self) -> None

getattr(self, name : str) -> Module

getitem(self, name : str) -> Module