# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping
from functools import wraps
from types import FunctionType, MethodType
from typing import Generic, TypeVar
from kelvin.api.base.api_service_model import ApiServiceModel, AsyncApiServiceModel
from kelvin.api.base.http_client.base_client import AsyncBaseClient, SyncBaseClient
from .api.app_parameters import AppParameters
from .api.app_workloads import AppWorkloads
from .api.apps import Apps
from .api.asset import Asset
from .api.asset_insights import AssetInsights
from .api.control_change import ControlChange
from .api.custom_actions import CustomActions
from .api.data_quality import DataQuality
from .api.data_tag import DataTag
from .api.datastreams import Datastreams
from .api.deprecated_app_registry import DeprecatedAppRegistry
from .api.deprecated_bridge import DeprecatedBridge
from .api.deprecated_parameters import DeprecatedParameters
from .api.deprecated_workload import DeprecatedWorkload
from .api.events import Events
from .api.filestorage import Filestorage
from .api.guardrails import Guardrails
from .api.instance import Instance
from .api.orchestration import Orchestration
from .api.properties import Properties
from .api.recommendation import Recommendation
from .api.secret import Secret
from .api.tasks import Tasks
from .api.thread import Thread
from .api.timeseries import Timeseries
from .api.user import User
from .api.user_authorization import UserAuthorization
from .async_api.app_parameters import AppParameters as AsyncAppParameters
from .async_api.app_workloads import AppWorkloads as AsyncAppWorkloads
from .async_api.apps import Apps as AsyncApps
from .async_api.asset import Asset as AsyncAsset
from .async_api.asset_insights import AssetInsights as AsyncAssetInsights
from .async_api.control_change import ControlChange as AsyncControlChange
from .async_api.custom_actions import CustomActions as AsyncCustomActions
from .async_api.data_quality import DataQuality as AsyncDataQuality
from .async_api.data_tag import DataTag as AsyncDataTag
from .async_api.datastreams import Datastreams as AsyncDatastreams
from .async_api.deprecated_app_registry import DeprecatedAppRegistry as AsyncDeprecatedAppRegistry
from .async_api.deprecated_bridge import DeprecatedBridge as AsyncDeprecatedBridge
from .async_api.deprecated_parameters import DeprecatedParameters as AsyncDeprecatedParameters
from .async_api.deprecated_workload import DeprecatedWorkload as AsyncDeprecatedWorkload
from .async_api.events import Events as AsyncEvents
from .async_api.filestorage import Filestorage as AsyncFilestorage
from .async_api.guardrails import Guardrails as AsyncGuardrails
from .async_api.instance import Instance as AsyncInstance
from .async_api.orchestration import Orchestration as AsyncOrchestration
from .async_api.properties import Properties as AsyncProperties
from .async_api.recommendation import Recommendation as AsyncRecommendation
from .async_api.secret import Secret as AsyncSecret
from .async_api.tasks import Tasks as AsyncTasks
from .async_api.thread import Thread as AsyncThread
from .async_api.timeseries import Timeseries as AsyncTimeseries
from .async_api.user import User as AsyncUser
from .async_api.user_authorization import UserAuthorization as AsyncUserAuthorization
MODELS: Mapping[str, type[ApiServiceModel]] = {
"deprecated_app_registry": DeprecatedAppRegistry, # type: ignore
"deprecated_bridge": DeprecatedBridge, # type: ignore
"deprecated_parameters": DeprecatedParameters, # type: ignore
"deprecated_workload": DeprecatedWorkload, # type: ignore
"app_parameters": AppParameters, # type: ignore
"app_workloads": AppWorkloads, # type: ignore
"apps": Apps, # type: ignore
"asset": Asset, # type: ignore
"asset_insights": AssetInsights, # type: ignore
"control_change": ControlChange, # type: ignore
"custom_actions": CustomActions, # type: ignore
"data_quality": DataQuality, # type: ignore
"data_tag": DataTag, # type: ignore
"datastreams": Datastreams, # type: ignore
"events": Events, # type: ignore
"filestorage": Filestorage, # type: ignore
"guardrails": Guardrails, # type: ignore
"instance": Instance, # type: ignore
"orchestration": Orchestration, # type: ignore
"properties": Properties, # type: ignore
"recommendation": Recommendation, # type: ignore
"secret": Secret, # type: ignore
"tasks": Tasks, # type: ignore
"thread": Thread, # type: ignore
"timeseries": Timeseries, # type: ignore
"user": User, # type: ignore
"user_authorization": UserAuthorization, # type: ignore
}
ASYNC_MODELS: Mapping[str, type[AsyncApiServiceModel]] = {
"deprecated_app_registry": AsyncDeprecatedAppRegistry, # type: ignore
"deprecated_bridge": AsyncDeprecatedBridge, # type: ignore
"deprecated_parameters": AsyncDeprecatedParameters, # type: ignore
"deprecated_workload": AsyncDeprecatedWorkload, # type: ignore
"app_parameters": AsyncAppParameters, # type: ignore
"app_workloads": AsyncAppWorkloads, # type: ignore
"apps": AsyncApps, # type: ignore
"asset": AsyncAsset, # type: ignore
"asset_insights": AsyncAssetInsights, # type: ignore
"control_change": AsyncControlChange, # type: ignore
"custom_actions": AsyncCustomActions, # type: ignore
"data_quality": AsyncDataQuality, # type: ignore
"data_tag": AsyncDataTag, # type: ignore
"datastreams": AsyncDatastreams, # type: ignore
"events": AsyncEvents, # type: ignore
"filestorage": AsyncFilestorage, # type: ignore
"guardrails": AsyncGuardrails, # type: ignore
"instance": AsyncInstance, # type: ignore
"orchestration": AsyncOrchestration, # type: ignore
"properties": AsyncProperties, # type: ignore
"recommendation": AsyncRecommendation, # type: ignore
"secret": AsyncSecret, # type: ignore
"tasks": AsyncTasks, # type: ignore
"thread": AsyncThread, # type: ignore
"timeseries": AsyncTimeseries, # type: ignore
"user": AsyncUser, # type: ignore
"user_authorization": AsyncUserAuthorization, # type: ignore
}
T = TypeVar("T", bound=ApiServiceModel)
class DataModelProxy(Generic[T]):
"""Proxy client to data models.
Provides attribute-based access to API model methods, automatically
injecting the client instance into method calls.
Attributes:
_model: The API service model class.
_client: The sync client instance.
"""
def __init__(self, model: type[T], client: Client) -> None:
"""Initialize the resource adaptor.
Args:
model: The API service model class to proxy.
client: The sync client instance to use for requests.
"""
self._model = model
self._client = client
def new(self, **kwargs: dict[str, object]) -> T:
"""Create a new instance of the model.
Args:
**kwargs: Field values for the new model instance.
Returns:
A new instance of the model.
"""
return self._model(self._client, **kwargs)
def __getattr__(self, name: str) -> object:
"""Map attribute access to model methods.
Args:
name: The attribute name to access.
Returns:
A wrapped method that injects the client, or the attribute value.
"""
if name.startswith("_"):
return super().__getattribute__(name)
try:
f = getattr(self._model, name)
except AttributeError:
return super().__getattribute__(name)
if isinstance(f, (FunctionType, MethodType)):
@wraps(f)
def wrapper(*args: tuple[object], **kwargs: dict[str, object]) -> object:
return f(*args, **kwargs, _client=self._client)
return wrapper
return super().__getattribute__(name)
def __dir__(self) -> list[str]:
"""List available methods for the model.
Returns:
A sorted list of public method names.
"""
return sorted(
k
for k in vars(self._model)
if not k.startswith("_") and isinstance(getattr(self._model, k), (FunctionType, MethodType))
)
def __str__(self) -> str:
"""Return a string representation.
Returns:
The string representation of the proxied model.
"""
return str(self._model)
def __repr__(self) -> str:
"""Return a detailed representation.
Returns:
The repr of the proxied model.
"""
return repr(self._model)
[docs]
class Client(SyncBaseClient):
"""Kelvin API Client.
Synchronous client for interacting with the Kelvin API.
Args:
password: Password for obtaining access token.
totp: Time-based one-time password for 2FA.
verbose: If True, log requests and responses.
"""
deprecated_app_registry: type[DeprecatedAppRegistry]
deprecated_bridge: type[DeprecatedBridge]
deprecated_parameters: type[DeprecatedParameters]
deprecated_workload: type[DeprecatedWorkload]
app_parameters: type[AppParameters]
app_workloads: type[AppWorkloads]
apps: type[Apps]
asset: type[Asset]
asset_insights: type[AssetInsights]
control_change: type[ControlChange]
custom_actions: type[CustomActions]
data_quality: type[DataQuality]
data_tag: type[DataTag]
datastreams: type[Datastreams]
events: type[Events]
filestorage: type[Filestorage]
guardrails: type[Guardrails]
instance: type[Instance]
orchestration: type[Orchestration]
properties: type[Properties]
recommendation: type[Recommendation]
secret: type[Secret]
tasks: type[Tasks]
thread: type[Thread]
timeseries: type[Timeseries]
user: type[User]
user_authorization: type[UserAuthorization]
[docs]
def __dir__(self) -> list[str]:
"""Return list of available attributes.
Returns:
A list of attribute names including API model names.
"""
return [*super().__dir__(), *MODELS]
[docs]
def __getattr__(self, name: str) -> object:
"""Get an attribute or API model proxy.
Args:
name: The attribute name to access.
Returns:
A DataModelProxy for API models, or the attribute value.
"""
if name.startswith("_") or name in super().__dir__():
return super().__getattribute__(name) # pragma: no cover
try:
model = MODELS[name]
except KeyError:
return super().__getattribute__(name)
return DataModelProxy(model, self)
A = TypeVar("A", bound=AsyncApiServiceModel)
class AsyncDataModelProxy(Generic[A]):
"""Proxy client to async data models.
Provides attribute-based access to async API model methods, automatically
injecting the client instance into method calls.
Attributes:
_model: The async API service model class.
_client: The async client instance.
"""
def __init__(self, model: type[A], client: AsyncClient) -> None:
"""Initialize the resource adaptor.
Args:
model: The async API service model class to proxy.
client: The async client instance to use for requests.
"""
self._model = model
self._client = client
def new(self, **kwargs: dict[str, object]) -> A:
"""Create a new instance of the model.
Args:
**kwargs: Field values for the new model instance.
Returns:
A new instance of the model.
"""
return self._model(self._client, **kwargs)
def __getattr__(self, name: str) -> object:
"""Map attribute access to model methods.
Args:
name: The attribute name to access.
Returns:
A wrapped method that injects the client, or the attribute value.
"""
if name.startswith("_"):
return super().__getattribute__(name)
try:
f = getattr(self._model, name)
except AttributeError:
return super().__getattribute__(name)
if isinstance(f, (FunctionType, MethodType)):
@wraps(f)
def wrapper(*args: tuple[object], **kwargs: dict[str, object]) -> object:
return f(*args, **kwargs, _client=self._client)
return wrapper
return super().__getattribute__(name)
def __dir__(self) -> list[str]:
"""List available methods for the model.
Returns:
A sorted list of public method names.
"""
return sorted(
k
for k in vars(self._model)
if not k.startswith("_") and isinstance(getattr(self._model, k), (FunctionType, MethodType))
)
def __str__(self) -> str:
"""Return a string representation.
Returns:
The string representation of the proxied model.
"""
return str(self._model)
def __repr__(self) -> str:
"""Return a detailed representation.
Returns:
The repr of the proxied model.
"""
return repr(self._model)
[docs]
class AsyncClient(AsyncBaseClient):
"""Kelvin API Async Client.
Asynchronous client for interacting with the Kelvin API.
Args:
password: Password for obtaining access token.
totp: Time-based one-time password for 2FA.
verbose: If True, log requests and responses.
"""
deprecated_app_registry: type[AsyncDeprecatedAppRegistry]
deprecated_bridge: type[AsyncDeprecatedBridge]
deprecated_parameters: type[AsyncDeprecatedParameters]
deprecated_workload: type[AsyncDeprecatedWorkload]
app_parameters: type[AsyncAppParameters]
app_workloads: type[AsyncAppWorkloads]
apps: type[AsyncApps]
asset: type[AsyncAsset]
asset_insights: type[AsyncAssetInsights]
control_change: type[AsyncControlChange]
custom_actions: type[AsyncCustomActions]
data_quality: type[AsyncDataQuality]
data_tag: type[AsyncDataTag]
datastreams: type[AsyncDatastreams]
events: type[AsyncEvents]
filestorage: type[AsyncFilestorage]
guardrails: type[AsyncGuardrails]
instance: type[AsyncInstance]
orchestration: type[AsyncOrchestration]
properties: type[AsyncProperties]
recommendation: type[AsyncRecommendation]
secret: type[AsyncSecret]
tasks: type[AsyncTasks]
thread: type[AsyncThread]
timeseries: type[AsyncTimeseries]
user: type[AsyncUser]
user_authorization: type[AsyncUserAuthorization]
[docs]
def __dir__(self) -> list[str]:
"""Return list of available attributes.
Returns:
A list of attribute names including async API model names.
"""
return [*super().__dir__(), *ASYNC_MODELS]
[docs]
def __getattr__(self, name: str) -> object:
"""Get an attribute or async API model proxy.
Args:
name: The attribute name to access.
Returns:
An AsyncDataModelProxy for API models, or the attribute value.
"""
if name.startswith("_") or name in super().__dir__():
return super().__getattribute__(name) # pragma: no cover
try:
model = ASYNC_MODELS[name]
except KeyError:
return super().__getattribute__(name)
return AsyncDataModelProxy(model, self)