# Code generated by builder. DO NOT EDIT.
"""
Kelvin API Client.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from typing import Optional, Union, overload
from typing_extensions import Literal
from kelvin.api.base.api_service_model import ApiServiceModel
from kelvin.api.base.data_model import KIterator, KList
from kelvin.api.base.error import ResponseError
from kelvin.api.base.http_client.base_client import Response, SyncBaseClient
from ..model import requests, response, responses, type
[docs]
class Timeseries(ApiServiceModel):
@overload
@classmethod
def create_timeseries(
cls,
publish: Optional[bool] = None,
data: Optional[Union[requests.TimeseriesCreate, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def create_timeseries(
cls,
publish: Optional[bool] = None,
data: Optional[Union[requests.TimeseriesCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def create_timeseries(
cls,
publish: Optional[bool] = None,
data: Optional[Union[requests.TimeseriesCreate, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> str: ...
[docs]
@classmethod
def create_timeseries(
cls,
publish: Optional[bool] = None,
data: Optional[Union[requests.TimeseriesCreate, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Optional[Union[str, dict[str, object], Response]]:
"""Create Time Series Data individually or in bulk for one or more Asset / Data Stream resources.
<span style="color: #ff0000;font-weight: bold;">WARNING</span> : If a value already
exists at the defined time, then it will be overwritten with the new payload value.
The old value will be lost and is not recoverable !
**Permission Required:** `kelvin.permission.storage.create`.
``createTimeseries``: ``POST`` ``/api/v4/timeseries/create``
Args:
publish : :obj:`bool`
data: requests.TimeseriesCreate, optional
**kwargs:
Extra parameters for requests.TimeseriesCreate
- create_timeseries: dict
"""
result_types = {"201": str, "207": None, "400": response.Error, "401": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/timeseries/create",
values={},
params={"publish": publish},
files={},
headers={},
data=data,
body_type=requests.TimeseriesCreate,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_str_response(response=_response)
@overload
@classmethod
def get_timeseries_last(
cls,
data: Optional[Union[requests.TimeseriesLastGet, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def get_timeseries_last(
cls,
data: Optional[Union[requests.TimeseriesLastGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def get_timeseries_last(
cls,
data: Optional[Union[requests.TimeseriesLastGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> KList[responses.TimeseriesLastGet]: ...
[docs]
@classmethod
def get_timeseries_last(
cls,
data: Optional[Union[requests.TimeseriesLastGet, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[KList[responses.TimeseriesLastGet], dict[str, object], Response]:
"""Returns the latest stored value for each selector in the request.
For advanced field projection behavior, see the `fields` property
description.
This endpoint is not paginated.
**Permission Required:** `kelvin.permission.storage.read`.
``getTimeseriesLast``: ``POST`` ``/api/v4/timeseries/last/get``
Args:
data: requests.TimeseriesLastGet, optional
**kwargs:
Extra parameters for requests.TimeseriesLastGet
- get_timeseries_last: dict
"""
result_types = {
"200": list[responses.TimeseriesLastGet],
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/timeseries/last/get",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.TimeseriesLastGet,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
return cls._process_response_list(
response=_response,
result_types=result_types,
result_type=KList[responses.TimeseriesLastGet],
)
@overload
@classmethod
def list_timeseries(
cls,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.TimeseriesList, Mapping[str, object]]] = None,
fetch: bool = True,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def list_timeseries(
cls,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.TimeseriesList, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def list_timeseries(
cls,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.TimeseriesList, Mapping[str, object]]] = None,
*,
fetch: Literal[False],
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[responses.TimeseriesListPaginatedResponseCursor, responses.TimeseriesListPaginatedResponseLimits]: ...
@overload
@classmethod
def list_timeseries(
cls,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.TimeseriesList, Mapping[str, object]]] = None,
fetch: Literal[True] = True,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> KList[type.TimeseriesData]: ...
[docs]
@classmethod
def list_timeseries(
cls,
pagination_type: Optional[Literal["limits", "cursor", "stream"]] = None,
page_size: Optional[int] = 10000,
page: Optional[int] = None,
next: Optional[str] = None,
previous: Optional[str] = None,
direction: Optional[Literal["asc", "desc"]] = None,
nulls: Optional[Literal["first", "last"]] = None,
sort_by: Optional[Sequence[str]] = None,
data: Optional[Union[requests.TimeseriesList, Mapping[str, object]]] = None,
fetch: bool = True,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[
Union[
KList[type.TimeseriesData],
responses.TimeseriesListPaginatedResponseCursor,
responses.TimeseriesListPaginatedResponseLimits,
],
dict[str, object],
Response,
]:
"""Returns a list of Time Series objects and its latest value. The list returned can be optionally restricted to one or more resources and/or sources.
**Permission Required:** `kelvin.permission.storage.read`.
``listTimeseries``: ``POST`` ``/api/v4/timeseries/list``
Args:
pagination_type : :obj:`Literal['limits', 'cursor', 'stream']`
Method of pagination to use for return results where `total_items` is
greater than `page_size`. `cursor` and `limits` will return one `page`
of results, `stream` will return all results. ('limits', 'cursor',
'stream')
page_size : :obj:`int`
Number of objects to be returned in each page. Page size can range
between 1 and 10000 objects.
page : :obj:`int`
An integer for the wanted page of results. Used only with
`pagination_type` set as `limits`.
next : :obj:`str`
An alphanumeric string bookmark to indicate where to start for the
next page. Used only with `pagination_type` set as `cursor`.
previous : :obj:`str`
An alphanumeric string bookmark to indicate where to end for the
previous page. Used only with `pagination_type` set as `cursor`.
direction : :obj:`Literal['asc', 'desc']`
Sorting order according to the `sort_by` parameter. ('asc', 'desc')
nulls : :obj:`Literal['first', 'last']`
Null ordering according to the `sort_by` parameter. Defaults to
`first` for ascending order and `last` for descending order. ('first',
'last')
sort_by : :obj:`Sequence[str]`
data: requests.TimeseriesList, optional
**kwargs:
Extra parameters for requests.TimeseriesList
- list_timeseries: dict
"""
result_types = {
"200": responses.TimeseriesListPaginatedResponseCursor,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
# override pagination_type
# stream type is only supported for raw responses
if not _get_response and pagination_type == "stream":
pagination_type = "cursor"
_request = cls._prepare_request(
method="POST",
path="/api/v4/timeseries/list",
values={},
params={
"pagination_type": pagination_type,
"page_size": page_size,
"page": page,
"next": next,
"previous": previous,
"direction": direction,
"nulls": nulls,
"sort_by": sort_by,
},
files={},
headers={},
data=data,
body_type=requests.TimeseriesList,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
if pagination_type == "limits":
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.TimeseriesListPaginatedResponseLimits,
)
else: # default pagination_type is cursor
result = cls._process_response(
response=_response,
result_types=result_types,
result_type=responses.TimeseriesListPaginatedResponseCursor,
)
if fetch:
return cls._fetch_pages(
client=_client,
path=_request.path,
api_response=result,
method="POST",
params=_request.params,
data=_request.data,
result_types=result_types,
error_type=response.Error,
response_type=KList[type.TimeseriesData],
)
return result
@overload
@classmethod
def download_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeDownload, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def download_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeDownload, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def download_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeDownload, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> str: ...
[docs]
@classmethod
def download_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeDownload, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[str, dict[str, object], Response]:
"""Returns a **CSV file** with Time Series data within the specified time range one or more resources (Asset /Data Stream pairs). Optional to preprocess and aggregate the data on the server using `agg` and `time_bucket` before downloading.
**Permission Required:** `kelvin.permission.storage.read`'.
``downloadTimeseriesRange``: ``POST`` ``/api/v4/timeseries/range/download``
Args:
data: requests.TimeseriesRangeDownload, optional
**kwargs:
Extra parameters for requests.TimeseriesRangeDownload
- download_timeseries_range: str
"""
result_types = {"200": str, "400": response.Error, "401": response.Error, "500": response.Error}
_request = cls._prepare_request(
method="POST",
path="/api/v4/timeseries/range/download",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.TimeseriesRangeDownload,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=False,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
result = cls._process_str_response(response=_response)
if result is None:
raise ResponseError(
f"Unexpected empty response",
_response,
)
return result
@overload
@classmethod
def get_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeGet, Mapping[str, object]]] = None,
*,
_dry_run: Literal[True],
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> dict[str, object]: ...
@overload
@classmethod
def get_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
*,
_get_response: Literal[True],
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Response: ...
@overload
@classmethod
def get_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeGet, Mapping[str, object]]] = None,
_dry_run: Literal[False] = False,
_get_response: Literal[False] = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> KIterator[responses.TimeseriesRangeGet]: ...
[docs]
@classmethod
def get_timeseries_range(
cls,
data: Optional[Union[requests.TimeseriesRangeGet, Mapping[str, object]]] = None,
_dry_run: bool = False,
_get_response: bool = False,
_client: Optional[SyncBaseClient] = None,
**kwargs: object,
) -> Union[KIterator[responses.TimeseriesRangeGet], dict[str, object], Response]:
"""Streams timeseries points within the requested time range for one or more
resources.
For field projection behavior, see the `fields` property description.
When aggregation is enabled, `fill` controls empty aggregation buckets.
**Permission Required:** `kelvin.permission.storage.read`.
``getTimeseriesRange``: ``POST`` ``/api/v4/timeseries/range/get``
Args:
data: requests.TimeseriesRangeGet, optional
**kwargs:
Extra parameters for requests.TimeseriesRangeGet
- get_timeseries_range: dict
"""
result_types = {
"200": responses.TimeseriesRangeGet,
"400": response.Error,
"401": response.Error,
"500": response.Error,
}
_request = cls._prepare_request(
method="POST",
path="/api/v4/timeseries/range/get",
values={},
params={},
files={},
headers={},
data=data,
body_type=requests.TimeseriesRangeGet,
array_body=False,
**kwargs,
)
if _dry_run:
return _request.to_dict()
_response = cls._make_request(
client=_client,
request=_request,
stream=True,
)
if _get_response:
return _response
cls._raise_api_error(
response=_response,
result_types=result_types,
error_type=response.Error,
)
result = cls._process_response_kiterator(
response=_response,
result_types=result_types,
return_type=responses.TimeseriesRangeGet,
)
if result is None:
raise ResponseError(
f"Unexpected empty response",
_response,
)
return result