From b36bb49e398df439708f1354f055c3ab08fe609a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 11:37:13 +0200 Subject: [PATCH 1/6] Change MsalCredential to TokenCredential and fix other basic pylance type validation errors as well --- pyproject.toml | 2 +- src/omnia_timeseries/api.py | 4 ++-- src/omnia_timeseries/http_client.py | 14 +++++++------- src/omnia_timeseries/models.py | 2 +- 4 files changed, 11 insertions(+), 11 deletions(-) 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..e6a7c4b 100644 --- a/src/omnia_timeseries/api.py +++ b/src/omnia_timeseries/api.py @@ -1,5 +1,5 @@ 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, @@ -103,7 +103,7 @@ class TimeseriesAPI: """ def __init__( - self, azure_credential: MsalCredential, environment: TimeseriesEnvironment + self, azure_credential: TokenCredential, environment: TimeseriesEnvironment ): self._http_client = HttpClient( azure_credential=azure_credential, resource_id=environment.resource_id 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..56073a9 100644 --- a/src/omnia_timeseries/models.py +++ b/src/omnia_timeseries/models.py @@ -212,5 +212,5 @@ def message(self) -> str: return self._message @property - def trace_id(self) -> str: + def trace_id(self) -> Optional[str]: return self._trace_id From bda50991c27265c363f8b52dc7dc5fc80629b06d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 12:13:19 +0200 Subject: [PATCH 2/6] Add APIs for IMS Metadata API and Subscriptions API --- src/omnia_timeseries/api.py | 362 ++++++++++++++++++++++++++++++--- src/omnia_timeseries/models.py | 48 +++++ 2 files changed, 381 insertions(+), 29 deletions(-) diff --git a/src/omnia_timeseries/api.py b/src/omnia_timeseries/api.py index e6a7c4b..22aa53d 100644 --- a/src/omnia_timeseries/api.py +++ b/src/omnia_timeseries/api.py @@ -1,4 +1,6 @@ +from dataclasses import dataclass from typing import Any, Dict, List, Literal, Optional +from typing_extensions import Self # Python 3.8 compatible from azure.core.credentials import TokenCredential from omnia_timeseries.http_client import HttpClient, ContentType from omnia_timeseries.models import ( @@ -11,16 +13,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 +49,129 @@ 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) -> Self: + """ + 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) -> Self: """ - 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) -> Self: """ - 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: @@ -92,23 +181,238 @@ def resource_id(self) -> str: 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: + return self._resource_id + + @property + 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" + ) + 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" + ) + 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: 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/models.py b/src/omnia_timeseries/models.py index 56073a9..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: From 8d8ba3214878297c8f6f7b56fae336c7eb4249cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 12:15:40 +0200 Subject: [PATCH 3/6] Update poetry dependencies --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index eb25607..136a7e3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,6 +15,7 @@ opentelemetry-instrumentation-requests = ">=0.54b1,<1.0.0" opentelemetry-api = ">=1.27.0,<2.0.0" cryptography = ">=46.0.5,<47.0.0" pyjwt = ">=2.9.0,<3.0.0" +typing_extensions = ">=4.15.0" [tool.poetry.group.test.dependencies] pytest = ">=7.4.0,<9.0.0" From d3cb81c5999b1d5e7abaddd34b67539d70ba9126 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 13:36:42 +0200 Subject: [PATCH 4/6] Fix couple endpoints and ensure forward reference is correctly used for Python v3.8 --- pyproject.toml | 1 - src/omnia_timeseries/api.py | 11 +++++------ 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 136a7e3..eb25607 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,6 @@ opentelemetry-instrumentation-requests = ">=0.54b1,<1.0.0" opentelemetry-api = ">=1.27.0,<2.0.0" cryptography = ">=46.0.5,<47.0.0" pyjwt = ">=2.9.0,<3.0.0" -typing_extensions = ">=4.15.0" [tool.poetry.group.test.dependencies] pytest = ">=7.4.0,<9.0.0" diff --git a/src/omnia_timeseries/api.py b/src/omnia_timeseries/api.py index 22aa53d..873bc2f 100644 --- a/src/omnia_timeseries/api.py +++ b/src/omnia_timeseries/api.py @@ -1,6 +1,5 @@ from dataclasses import dataclass from typing import Any, Dict, List, Literal, Optional -from typing_extensions import Self # Python 3.8 compatible from azure.core.credentials import TokenCredential from omnia_timeseries.http_client import HttpClient, ContentType from omnia_timeseries.models import ( @@ -62,21 +61,21 @@ class TimeseriesEnvironment: kind: Environment @classmethod - def Dev(cls) -> Self: + def Dev(cls) -> "TimeseriesEnvironment": """ Set non-production dev environment """ return cls(kind=Environment.Dev) @classmethod - def Test(cls) -> Self: + def Test(cls) -> "TimeseriesEnvironment": """ Set non-production test environment """ return cls(kind=Environment.Test) @classmethod - def Prod(cls) -> Self: + def Prod(cls) -> "TimeseriesEnvironment": """ Set production environment """ @@ -310,7 +309,7 @@ def search_by_uid( if continuationToken is not None: params["continuationToken"] = continuationToken url = ( - f"{self._base_url}/uid" + f"{self._base_url}/uid/{uid}" ) return self._http_client.request(request_type="get", url=url, params=params) @@ -329,7 +328,7 @@ def search_by_timeseries_id( if continuationToken is not None: params["continuationToken"] = continuationToken url = ( - f"{self._base_url}/timeseriesId" + f"{self._base_url}/timeseriesId/{timeseriesId}" ) return self._http_client.request(request_type="get", url=url, params=params) From cab3091aec4d4ff7d641d798d002f85d1a7f50f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 13:57:04 +0200 Subject: [PATCH 5/6] Add other relevant APIs for user in import setup --- examples/usage_examples.ipynb | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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)" ] }, { From 0b7808d33a202023e7652839847f17a237f150c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Inge=20Nyg=C3=A5rd=20=28Omega=20AS=29?= Date: Fri, 17 Apr 2026 14:33:26 +0200 Subject: [PATCH 6/6] Fix unit tests --- tests/test_api.py | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) 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