diff --git a/examples/usage_examples.ipynb b/examples/usage_examples.ipynb index 0c85929..b2861da 100644 --- a/examples/usage_examples.ipynb +++ b/examples/usage_examples.ipynb @@ -31,7 +31,7 @@ "metadata": {}, "outputs": [], "source": [ - "from omnia_timeseries.api import TimeseriesAPI, TimeseriesEnvironment, FederationSource\n", + "from omnia_timeseries.api import TimeseriesAPI, IMSMetadataAPI, IMSSubscriptionsAPI, TimeseriesEnvironment, FederationSource\n", "import os" ] }, @@ -122,7 +122,8 @@ "metadata": {}, "outputs": [], "source": [ - "api = TimeseriesAPI(azure_credential=credentials, environment=TimeseriesEnvironment.Dev())" + "environment=TimeseriesEnvironment.Dev()\n", + "api = TimeseriesAPI(azure_credential=credentials, environment=environment)" ] }, { diff --git a/pyproject.toml b/pyproject.toml index 28bce1c..eb25607 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "omnia_timeseries" -version = "1.4.3" +version = "1.4.4" description = "Official Python SDK for the Omnia Timeseries API" readme = "README.md" authors = ["Equinor Omnia Industrial IoT Team "] diff --git a/src/omnia_timeseries/api.py b/src/omnia_timeseries/api.py index 6340950..873bc2f 100644 --- a/src/omnia_timeseries/api.py +++ b/src/omnia_timeseries/api.py @@ -1,5 +1,6 @@ +from dataclasses import dataclass from typing import Any, Dict, List, Literal, Optional -from azure.identity._internal.msal_credentials import MsalCredential +from azure.core.credentials import TokenCredential from omnia_timeseries.http_client import HttpClient, ContentType from omnia_timeseries.models import ( DatapointModel, @@ -11,16 +12,21 @@ GetHistoryResponseModel, GetMultipleDatapointsRequestItem, GetTimeseriesResponseModel, + GetIMSMetadataResponseModel, SourceDataModel, MessageModel, StreamSubscriptionRequestModel, StreamSubscriptionDataModel, + SubscriptionCounterModel, + SubscriptionPatchRequestItem, TimeseriesPatchRequestItem, TimeseriesRequestItem, ) import logging from enum import Enum +IMSMetadataVersion = Literal["1.3"] +IMSSubscriptionsVersion = Literal["1.2"] TimeseriesVersion = Literal["1.6", "1.7"] logger = logging.getLogger(__name__) @@ -42,47 +48,162 @@ def __str__(self): TSDB = "TSDB" DataLake = "DataLake" +# Internal enum to represent environment chosen by user via TimeseriesEnvironment +class Environment(Enum): + Dev = "dev" + Test = "test" + Prod = "prod" +# User-facing, immutable token accepted by all APIs with Python v3.8 slots-like behaviour +@dataclass(frozen=True) class TimeseriesEnvironment: - def __init__(self, resource_id: str, base_url: str): - """ - Wrapper class for defining which timeseries environment the TimeseriesAPI client will interface to + __slots__ = ("kind",) + kind: Environment - :param str resource_id: Resource/client id of API app registration (Azure SPN) - :param str base_url: Baze url of API + @classmethod + def Dev(cls) -> "TimeseriesEnvironment": + """ + Set non-production dev environment """ - self._resource_id = resource_id - self._base_url = base_url + return cls(kind=Environment.Dev) @classmethod - def Dev(cls, version: TimeseriesVersion = "1.7"): + def Test(cls) -> "TimeseriesEnvironment": """ - Sets up non-production dev environment + Set non-production test environment """ - return cls( - resource_id="32f2a909-8a98-4eb8-b22d-1208d9350cb0", - base_url=f"https://api-dev.gateway.equinor.com/plant/timeseries/v{version}", - ) + return cls(kind=Environment.Test) @classmethod - def Test(cls, version: TimeseriesVersion = "1.7"): + def Prod(cls) -> "TimeseriesEnvironment": """ - Sets up non-production environment + Set production environment """ - return cls( - resource_id="32f2a909-8a98-4eb8-b22d-1208d9350cb0", - base_url=f"https://api-test.gateway.equinor.com/plant/timeseries/v{version}", - ) + return cls(kind=Environment.Prod) - @classmethod - def Prod(cls, version: TimeseriesVersion = "1.7"): +class IMSMetadataApiEnvironment: + def __init__(self, environment: TimeseriesEnvironment, version: IMSMetadataVersion = "1.3"): """ - Sets up production environment + Wrapper class for defining which environment the IMS Metadata API client will interface to + + :param TimeseriesEnvironment environment: Value with environment chosen as dev, test or prod """ - return cls( - resource_id="141369bd-3dca-4b55-825b-56ad4a69b1fc", - base_url=f"https://api.gateway.equinor.com/plant/timeseries/v{version}", - ) + + # Prepare API connection details + if environment.kind is Environment.Dev: + self._resource_id = "310547f5-022b-4fb5-b13e-2b468b8bf658" + self._base_url = f"https://api-dev.gateway.equinor.com/plant/ims-metadata/v{version}" + + if environment.kind is Environment.Test: + self._resource_id = "310547f5-022b-4fb5-b13e-2b468b8bf658" + self._base_url = f"https://api-test.gateway.equinor.com/plant/ims-metadata/v{version}" + + if environment.kind is Environment.Prod: + self._resource_id = "71415e4e-7131-4a81-9596-f921978cdbee" + self._base_url = f"https://api.gateway.equinor.com/plant/ims-metadata/v{version}" + + # Safeguard in case of new, unsupported environment + if not hasattr(self, "_resource_id") or not hasattr(self, "_base_url"): + raise ValueError(f"Unsupported environment: {environment.kind!r}") + + @property + def resource_id(self) -> str: + return self._resource_id + + @property + def base_url(self) -> str: + return self._base_url + +class IMSSubscriptionsAPIEnvironment: + def __init__(self, environment: TimeseriesEnvironment, version: IMSSubscriptionsVersion = "1.2"): + """ + Wrapper class for defining which environment the IMS Subscriptions API client will interface to + + :param TimeseriesEnvironment environment: Value with environment chosen as dev, test or prod + """ + + # Prepare API connection details + if environment.kind is Environment.Dev: + self._resource_id = "789d768e-32ac-417b-b286-be5e0d460809" + self._base_url = f"https://api-dev.gateway.equinor.com/plant/ims-subscriptions/v{version}" + + if environment.kind is Environment.Test: + self._resource_id = "789d768e-32ac-417b-b286-be5e0d460809" + self._base_url = f"https://api-test.gateway.equinor.com/plant/ims-subscriptions/v{version}" + + if environment.kind is Environment.Prod: + self._resource_id = "cae23674-f25d-40a2-b9e4-81649b33b957" + self._base_url = f"https://api.gateway.equinor.com/plant/ims-subscriptions/v{version}" + + # Safeguard in case of new, unsupported environment + if not hasattr(self, "_resource_id") or not hasattr(self, "_base_url"): + raise ValueError(f"Unsupported environment: {environment.kind!r}") + + @property + def resource_id(self) -> str: + return self._resource_id + + @property + def base_url(self) -> str: + return self._base_url + +class IMSSubscriptionsManagementAPIEnvironment: + def __init__(self, environment: TimeseriesEnvironment, version: IMSSubscriptionsVersion = "1.2"): + """ + Wrapper class for defining which environment the IMS Subscriptions Management API client will interface to + + :param TimeseriesEnvironment environment: Value with environment chosen as dev, test or prod + """ + + # Prepare API connection details + if environment.kind is Environment.Dev: + self._resource_id = "789d768e-32ac-417b-b286-be5e0d460809" + self._base_url = f"https://api-dev.gateway.equinor.com/plant/ims-subscriptions-management/v{version}" + + if environment.kind is Environment.Test: + self._resource_id = "789d768e-32ac-417b-b286-be5e0d460809" + self._base_url = f"https://api-test.gateway.equinor.com/plant/ims-subscriptions-management/v{version}" + + if environment.kind is Environment.Prod: + self._resource_id = "cae23674-f25d-40a2-b9e4-81649b33b957" + self._base_url = f"https://api.gateway.equinor.com/plant/ims-subscriptions-management/v{version}" + + # Safeguard in case of new, unsupported environment + if not hasattr(self, "_resource_id") or not hasattr(self, "_base_url"): + raise ValueError(f"Unsupported environment: {environment.kind!r}") + + @property + def resource_id(self) -> str: + return self._resource_id + + @property + def base_url(self) -> str: + return self._base_url + +class TimeseriesApiEnvironment: + def __init__(self, environment: TimeseriesEnvironment, version: TimeseriesVersion = "1.7"): + """ + Wrapper class for defining which environment the Timeseries API client will interface to + + :param TimeseriesEnvironment environment: Value with environment chosen as dev, test or prod + """ + + # Prepare API connection details + if environment.kind is Environment.Dev: + self._resource_id = "32f2a909-8a98-4eb8-b22d-1208d9350cb0" + self._base_url = f"https://api-dev.gateway.equinor.com/plant/timeseries/v{version}" + + if environment.kind is Environment.Test: + self._resource_id = "32f2a909-8a98-4eb8-b22d-1208d9350cb0" + self._base_url = f"https://api-test.gateway.equinor.com/plant/timeseries/v{version}" + + if environment.kind is Environment.Prod: + self._resource_id = "141369bd-3dca-4b55-825b-56ad4a69b1fc" + self._base_url = f"https://api.gateway.equinor.com/plant/timeseries/v{version}" + + # Safeguard in case of new, unsupported environment + if not hasattr(self, "_resource_id") or not hasattr(self, "_base_url"): + raise ValueError(f"Unsupported environment: {environment.kind!r}") @property def resource_id(self) -> str: @@ -92,23 +213,205 @@ def resource_id(self) -> str: def base_url(self) -> str: return self._base_url +class IMSMetadataAPI: + """ + Wrapper class for interacting with the Omnia Industrial IIoT IMS Metadata API. + For more information, see https://github.com/equinor/OmniaPlant/wiki or consult with the Omnia IIoT team. + + :param MsalCredential azure_credential: Azure credential instance used for authenticating + :param TimeseriesEnvironment environment: API deployment environment + """ + + def __init__( + self, azure_credential: TokenCredential, environment: TimeseriesEnvironment + ): + if not isinstance(environment, TimeseriesEnvironment): + raise TypeError(f"Environment must be TimeseriesEnvironment, got: {type(environment).__name__}") + apiEnvironment=IMSMetadataApiEnvironment(environment) + self._http_client = HttpClient( + azure_credential=azure_credential, resource_id=apiEnvironment.resource_id + ) + self._base_url = apiEnvironment.base_url.rstrip("/") + + def search( + self, + tag: Optional[str] = None, + uid: Optional[str] = None, + continuationToken: Optional[str] = None, + **kwargs, + ) -> GetIMSMetadataResponseModel: + """ + Search IMS Metadata API + """ + params = kwargs or {} + if tag is not None: + params["tag"] = tag + if uid is not None: + params["uid"] = uid + if continuationToken is not None: + params["continuationToken"] = continuationToken + url = ( + f"{self._base_url}/search" + ) + return self._http_client.request(request_type="get", url=url, params=params) + +class IMSSubscriptionsAPI: + """ + Wrapper class for interacting with the Omnia Industrial IIoT IMS Subscriptions API. + For more information, see https://github.com/equinor/OmniaPlant/wiki or consult with the Omnia IIoT team. + + :param MsalCredential azure_credential: Azure credential instance used for authenticating + :param TimeseriesEnvironment environment: API deployment environment + """ + + def __init__( + self, azure_credential: TokenCredential, environment: TimeseriesEnvironment + ): + if not isinstance(environment, TimeseriesEnvironment): + raise TypeError(f"Environment must be TimeseriesEnvironment, got: {type(environment).__name__}") + apiEnvironment=IMSSubscriptionsAPIEnvironment(environment) + self._http_client = HttpClient( + azure_credential=azure_credential, resource_id=apiEnvironment.resource_id + ) + self._base_url = apiEnvironment.base_url.rstrip("/") + + def search( + self, + limit: Optional[int] = 1000, + continuationToken: Optional[str] = None, + **kwargs, + ) -> GetIMSMetadataResponseModel: + """ + Search IMS Subscriptions API + """ + params = kwargs or {} + if limit is not None: + params["limit"] = limit + if continuationToken is not None: + params["continuationToken"] = continuationToken + url = ( + f"{self._base_url}" + ) + return self._http_client.request(request_type="get", url=url, params=params) + + def search_by_uid( + self, + uid: Optional[str] = None, + continuationToken: Optional[str] = None, + **kwargs, + ) -> GetIMSMetadataResponseModel: + """ + Search IMS Subscriptions API + """ + params = kwargs or {} + if uid is not None: + params["uid"] = uid + if continuationToken is not None: + params["continuationToken"] = continuationToken + url = ( + f"{self._base_url}/uid/{uid}" + ) + return self._http_client.request(request_type="get", url=url, params=params) + + def search_by_timeseries_id( + self, + timeseriesId: Optional[str] = None, + continuationToken: Optional[str] = None, + **kwargs, + ) -> GetIMSMetadataResponseModel: + """ + Search IMS Subscriptions API + """ + params = kwargs or {} + if timeseriesId is not None: + params["timeseriesId"] = timeseriesId + if continuationToken is not None: + params["continuationToken"] = continuationToken + url = ( + f"{self._base_url}/timeseriesId/{timeseriesId}" + ) + return self._http_client.request(request_type="get", url=url, params=params) + + def search_by_system_code( + self, + systemCode: str, + **kwargs, + ) -> SubscriptionCounterModel: + """ + Search IMS Subscriptions API + """ + params = kwargs or {} + url = ( + f"{self._base_url}/{systemCode}" + ) + return self._http_client.request(request_type="get", url=url, params=params) + +class IMSSubscriptionsManagementAPI: + """ + Wrapper class for interacting with the Omnia Industrial IIoT IMS Subscriptions Management API. + For more information, see https://github.com/equinor/OmniaPlant/wiki or consult with the Omnia IIoT team. + + :param MsalCredential azure_credential: Azure credential instance used for authenticating + :param TimeseriesEnvironment environment: API deployment environment + """ + + def __init__( + self, azure_credential: TokenCredential, environment: TimeseriesEnvironment + ): + if not isinstance(environment, TimeseriesEnvironment): + raise TypeError(f"Environment must be TimeseriesEnvironment, got: {type(environment).__name__}") + apiEnvironment=IMSSubscriptionsManagementAPIEnvironment(environment) + self._http_client = HttpClient( + azure_credential=azure_credential, resource_id=apiEnvironment.resource_id + ) + self._base_url = apiEnvironment.base_url.rstrip("/") + + def get_subscription_counter_by_system_code( + self, + systemCode: str, + **kwargs, + ) -> SubscriptionCounterModel: + """ + Search IMS Subscriptions Management API + """ + params = kwargs or {} + url = ( + f"{self._base_url}/{systemCode}/counter" + ) + return self._http_client.request(request_type="get", url=url, params=params) + + def patch_subscription_by_uid( + self, + uid: str, + request: SubscriptionPatchRequestItem + ) -> GetIMSMetadataResponseModel: + """ + Search IMS Subscriptions Management API + """ + url = ( + f"{self._base_url}/uid/{uid}" + ) + return self._http_client.request(request_type="patch", url=url, payload=request) class TimeseriesAPI: """ Wrapper class for interacting with the Omnia Industrial IIoT Timeseries API. For more information, see https://github.com/equinor/OmniaPlant/wiki or consult with the Omnia IIoT team. - :param MsalCredential azure_credential: Azure credential instance used for authenticating + :param TokenCredential azure_credential: Azure token credential instance used for authenticating :param TimeseriesEnvironment environment: API deployment environment """ def __init__( - self, azure_credential: MsalCredential, environment: TimeseriesEnvironment + self, azure_credential: TokenCredential, environment: TimeseriesEnvironment ): + if not isinstance(environment, TimeseriesEnvironment): + raise TypeError(f"Environment must be TimeseriesEnvironment, got: {type(environment).__name__}") + apiEnvironment=TimeseriesApiEnvironment(environment) self._http_client = HttpClient( - azure_credential=azure_credential, resource_id=environment.resource_id + azure_credential=azure_credential, resource_id=apiEnvironment.resource_id ) - self._base_url = environment.base_url.rstrip("/") + self._base_url = apiEnvironment.base_url.rstrip("/") self._debug_mode = False def _add_debug_param(self, params: Dict[str, Any]) -> Dict[str, Any]: diff --git a/src/omnia_timeseries/http_client.py b/src/omnia_timeseries/http_client.py index 11f9b31..dbf3197 100644 --- a/src/omnia_timeseries/http_client.py +++ b/src/omnia_timeseries/http_client.py @@ -1,5 +1,5 @@ -from typing import Literal, List, Optional, Union, Dict, Any -from azure.identity._internal.msal_credentials import MsalCredential +from typing import Literal, List, Optional, Union, Dict, Any, Mapping, Sequence +from azure.core.credentials import TokenCredential import requests import logging @@ -31,8 +31,8 @@ def _request( request_type: RequestType, url: str, headers: Dict[str, Any], - payload: Optional[Union[Dict, Dict, List]] = None, - params: Optional[Dict[str, Any]] = None, + payload: Optional[Mapping[str, Any] | Sequence[Any]] = None, + params: Optional[Mapping[str, Any]] = None, ) -> Union[Dict[str, Any], bytes]: response = requests.request( @@ -47,7 +47,7 @@ def _request( class HttpClient: - def __init__(self, azure_credential: MsalCredential, resource_id: str): + def __init__(self, azure_credential: TokenCredential, resource_id: str): self._azure_credential = azure_credential self._resource_id = resource_id @@ -56,8 +56,8 @@ def request( request_type: RequestType, url: str, accept: ContentType = "application/json", - payload: Optional[Union[Dict, Dict, List]] = None, - params: Optional[Dict[str, Any]] = None, + payload: Optional[Mapping[str, Any] | Sequence[Any]] = None, + params: Optional[Mapping[str, Any]] = None, ) -> Any: access_token = self._azure_credential.get_token( diff --git a/src/omnia_timeseries/models.py b/src/omnia_timeseries/models.py index 9028a56..46375e4 100644 --- a/src/omnia_timeseries/models.py +++ b/src/omnia_timeseries/models.py @@ -78,6 +78,44 @@ class GetAggregatesResponseModel(TypedDict): count: Optional[int] continuationToken: Optional[str] +class IMSMetadataItemsModel(TypedDict): + items: List[IMSMetadataModel] + +class GetIMSMetadataResponseModel(TypedDict): + data: IMSMetadataItemsModel + count: Optional[int] + continuationToken: Optional[str] + +class IMSMetadataModel(TypedDict): + id: str + uid: str + imsType: str + systemCode: str + recordId: bool + plantSapCode: str + plantStidCode: str + tag: str + terminal: str + isDefaultTerminal: str + description: str + engUnits: str + standardUnit: str + source: str + type: str + sourceTag: str + stepped: str + changeDate: str + creationDate: str + currentValueDate: str + compressed: str + compressionDeviation: str + compressionMaximum: str + compressionMinimum: str + compressionDeviationPercent: str + plantArea: str + significantDigit: str + maxTimeInterval: str + fieldId: str class TimeseriesModel(TypedDict): id: str @@ -103,6 +141,13 @@ class GetTimeseriesResponseModel(TypedDict): count: Optional[int] continuationToken: Optional[str] +class SubscriptionPatchRequestItem(TypedDict, total=False): + name: Optional[str] + plantStidCode: Optional[str] + plantSapCode: Optional[bool] + terminal: Optional[str] + fieldId: Optional[str] + timeseriesId: Optional[str] class TimeseriesRequestItem(TypedDict, total=False): name: str @@ -182,6 +227,9 @@ class StreamSubscriptionItemsModel(TypedDict): class StreamSubscriptionDataModel(TypedDict): data: StreamSubscriptionItemsModel +class SubscriptionCounterModel(TypedDict): + quota: int + count: int class TimeseriesRequestFailedException(Exception): def __init__(self, response: Response) -> None: @@ -212,5 +260,5 @@ def message(self) -> str: return self._message @property - def trace_id(self) -> str: + def trace_id(self) -> Optional[str]: return self._trace_id diff --git a/tests/test_api.py b/tests/test_api.py index 2998ce1..c3899fc 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,24 +1,19 @@ import requests_mock import pytest -from azure.identity._internal.msal_credentials import MsalCredential from azure.core.credentials import AccessToken from omnia_timeseries import TimeseriesAPI, TimeseriesEnvironment from omnia_timeseries.http_client import TimeseriesRequestFailedException - -class DummyCredentials(MsalCredential): - - def get_token(self, *scopes, **kwargs): +class DummyCredentials: + def get_token(self, *scopes: str, **kwargs) -> AccessToken: return AccessToken("dummytoken", 0) - def _get_app(self): - return None - @pytest.fixture def api(): - env = TimeseriesEnvironment("dummy", "https://test") - api = TimeseriesAPI(DummyCredentials("dummy"), env) + env = TimeseriesEnvironment.Dev() + api = TimeseriesAPI(azure_credential=DummyCredentials(), environment=env) + api._base_url = "https://test" return api