feat: app registration authentication for equnor pi webapi

examples/demo.py

@@ -0,0 +1,36 @@
+import tagreader
+from tagreader.utils import IMSType
+from tagreader.web_handlers import get_auth_aspen, get_auth_pi, get_url_aspen
+
+use_internal = False
+
+
+print(
+    tagreader.list_sources(
+        imstype=IMSType.ASPENONE,
+        url=get_url_aspen(use_internal=use_internal),
+        auth=get_auth_aspen(use_internal=use_internal),
+    )
+)
+
+c = tagreader.IMSClient(
+    "GRA",
+    handler_options={"max_rows": 1e7},
+    auth=get_auth_aspen(use_internal=use_internal),
+    url=get_url_aspen(use_internal=use_internal),
+)
+t = c.search("*PDIC*")
+d = c.read(tags=t[0][0], start_time="2024-Jan-01", end_time="2024-Jan-02")
+
+
+print(
+    tagreader.list_sources(
+        imstype=IMSType.PIWEBAPI, auth=get_auth_pi(use_internal=use_internal)
+    )
+)
+c = tagreader.IMSClient("JSV")
+t4 = c.search("*PDIC*")
+d = c.read(tags=t4[0][0], start_time="2024-Jan-02", end_time="2024-Jan-02")
+
+
+pass

tagreader/web_handlers.py

@@ -14,6 +14,7 @@
 import requests
 import urllib3
 from Crypto.Hash import MD4 as _MD4
+from msal_bearer import BearerAuth
 from requests_kerberos import OPTIONAL, HTTPKerberosAuth
 from urllib3.exceptions import InsecureRequestWarning
 
@@ -55,20 +56,28 @@ def get_verify_ssl() -> Union[bool, str]:
     return "/etc/ssl/certs/ca-bundle.trust.crt"
 
 
-def get_auth_pi() -> HTTPKerberosAuth:
-    return HTTPKerberosAuth(mutual_authentication=OPTIONAL)
+def get_auth_pi(use_internal: bool = True) -> Union[HTTPKerberosAuth, BearerAuth]:
+    if use_internal:
+        return HTTPKerberosAuth(mutual_authentication=OPTIONAL)
+
+    tenant_id = "3aa4a235-b6e2-48d5-9195-7fcf05b459b0"
+    client_id = "98fe146b-2687-4db9-9c84-45f4cd9063af"
+
+    scopes = ["https://piwebapi.equinor.com//user_impersonation"]
+
+    return BearerAuth.get_auth(
+        tenantID=tenant_id, clientID=client_id, scopes=scopes, verbose=True
+    )
 
 
 def get_url_pi() -> str:
     return r"https://piwebapi.equinor.com/piwebapi"
 
 
-def get_auth_aspen(use_internal: bool = True):
+def get_auth_aspen(use_internal: bool = True) -> Union[HTTPKerberosAuth, BearerAuth]:
     if use_internal:
         return HTTPKerberosAuth(mutual_authentication=OPTIONAL)
 
-    from msal_bearer import BearerAuth
-
     tenantID = "3aa4a235-b6e2-48d5-9195-7fcf05b459b0"
     clientID = "7adaaa99-897f-428c-8a5f-4053db565b32"
     scopes = [
@@ -115,6 +124,8 @@ def list_aspenone_sources(
     except JSONDecodeError as e:
         logger.error(f"Could not decode JSON response: {e}")
 
+    return []
+
 
 def list_piwebapi_sources(
     url: Optional[str] = None,
@@ -134,7 +145,14 @@ def list_piwebapi_sources(
         urllib3.disable_warnings(InsecureRequestWarning)
 
     url_ = urljoin(url, "dataservers")
-    res = requests.get(url_, auth=auth, verify=verify_ssl, timeout=300)
+    res = requests.get(
+        url_,
+        auth=auth,
+        verify=verify_ssl,
+        timeout=300,
+        headers={"Accept": "application/json"},
+        allow_redirects=False,
+    )
 
     res.raise_for_status()
     try:
@@ -143,6 +161,8 @@ def list_piwebapi_sources(
     except JSONDecodeError as e:
         logger.error(f"Could not decode JSON response: {e}")
 
+    return []
+
 
 def get_piwebapi_source_to_webid_dict(
     url: Optional[str] = None,
@@ -170,6 +190,8 @@ def get_piwebapi_source_to_webid_dict(
     except JSONDecodeError as e:
         logger.error(f"Could not decode JSON response: {e}")
 
+    return []
+
 
 class BaseHandlerWeb(ABC):
     def __init__(