The great refactor [EWB-4557]

README.md

@@ -1,3 +1,5 @@
+from zepben.eas.client.auth_method import TokenAuth
+
 # Evolve App Server Python Client #
 
 This library provides a wrapper to the Evolve App Server's API, allowing users of the evolve SDK to authenticate with
@@ -7,16 +9,18 @@ the Evolve App Server and upload studies.
 
 ```python
 from geojson import FeatureCollection
-from zepben.eas import EasClient, Study, Result, Section, GeoJsonOverlay
+from zepben.eas import EasClient, Study, Result, Section, GeoJsonOverlay, TokenAuth
 
 eas_client = EasClient(
-    host="<host>",
-    port=1234,
-    access_token="<access_token>",
-    client_id="<client_id>",
-    username="<username>",
-    password="<password>",
-    client_secret="<client_secret>"
+    TokenAuth(
+        host="<host>",
+        port=1234,
+        access_token="<access_token>",
+        client_id="<client_id>",
+        username="<username>",
+        password="<password>",
+        client_secret="<client_secret>"
+    )
 )
 
 eas_client.upload_study(
@@ -66,17 +70,19 @@ To use the asyncio API use `async_upload_study` like so:
 ```python
 from aiohttp import ClientSession
 from geojson import FeatureCollection
-from zepben.eas import EasClient, Study, Result, Section, GeoJsonOverlay
+from zepben.eas import EasClient, Study, Result, Section, GeoJsonOverlay, TokenAuth
 
 async def upload():
     eas_client = EasClient(
-        host="<host>",
-        port=1234,
-        access_token="<access_token>",
-        client_id="<client_id>",
-        username="<username>",
-        password="<password>",
-        client_secret="<client_secret>",
+        TokenAuth(
+            host="<host>",
+            port=1234,
+            access_token="<access_token>",
+            client_id="<client_id>",
+            username="<username>",
+            password="<password>",
+            client_secret="<client_secret>",
+        ),
         session=ClientSession(...)
     )
 

changelog.md

@@ -77,6 +77,8 @@ and are now `before_cut_off_profile` and `after_cut_off_profile` respectively.
 ### Breaking Changes
 * Renamed the parameter `calibration_id` to `calibration_name` for the following methods `get_transformer_tap_settings` and `async_get_transformer_tap_settings`. This better reflects that
   this parameter is the user supplied calibration name rather than EAS's internal calibration run ID. 
+* EasClient will now need `auth=` passed with an auth object. either `BaseAuthMethod` or `TokenAuth`, this allows
+  cleaner documenting of accepted constructor arguments.
 
 ### New Features
 * Added optional fields to `ModelConfig` to control network simplification: `simplify_network`, `collapse_negligible_impedances`, and `combine_common_impedances`.
@@ -90,7 +92,9 @@ and are now `before_cut_off_profile` and `after_cut_off_profile` respectively.
 * Added optional field `inverterControlConfig` to `ModelConfig`. This `PVVoltVARVoltWattConfig` allows the configuration of advanced inverter control profiles.
 
 ### Enhancements
-* None.
+* Internal: query bodys are now mostly self generating with `to_json` and `build_gql_query_object_model` methods.
+* All request handling logic has been refactored into a single method.
+* `catch_warnings` wrapper func to handle standard warning catching.
 
 ### Fixes
 * `TimePeriod` no longer truncates the `start_time` and `end_time` to midnight(`00:00:00`). `TimePeriod` will now preserve arbitrary start and end times to minute precision.

setup.py

@@ -1,4 +1,4 @@
-#  Copyright 2020 Zeppelin Bend Pty Ltd
+#  Copyright 2025 Zeppelin Bend Pty Ltd
 #
 #  This Source Code Form is subject to the terms of the Mozilla Public
 #  License, v. 2.0. If a copy of the MPL was not distributed with this

src/zepben/eas/__init__.py

@@ -6,5 +6,6 @@
 #
 
 from zepben.eas.client.eas_client import EasClient
+from zepben.eas.client.auth_method import *
 from zepben.eas.client.study import *
 from zepben.eas.client.work_package import *

src/zepben/eas/client/auth_method.py

@@ -0,0 +1,193 @@
+#  Copyright 2025 Zeppelin Bend Pty Ltd
+#
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+__all__ = ['BaseAuthMethod', 'TokenAuth']
+
+
+from hashlib import sha256
+from typing import Optional, overload
+
+from aiohttp import ClientSession
+from zepben.auth import ZepbenTokenFetcher, create_token_fetcher, AuthMethod, create_token_fetcher_managed_identity
+
+
+class EasClient:
+    protocol: str = "https",
+    verify_certificate: bool = True,
+    ca_filename: Optional[str] = None,
+    session: ClientSession = None,
+    json_serialiser=None
+
+
+class BaseAuthMethod:
+    def __init__(self, host, port, protocol='https', verify_certificate=True):
+        """
+
+        :param host: The domain of the Evolve App Server, e.g. "evolve.local"
+        :param port: The port on which to make requests to the Evolve App Server, e.g. 7624
+        :param protocol: The protocol of the Evolve App Server. Should be either "http" or "https". Must be "https" if
+                         auth is configured. (Defaults to "https")
+        :param verify_certificate: Set this to "False" to disable certificate verification. This will also apply to the
+                                   auth provider if auth is initialised via client id + username + password or
+                                   client_id + client_secret. (Defaults to True)
+        """
+        self._host = host
+        self._port = port
+        self.protocol = protocol
+        self.verify_certificate = verify_certificate
+
+    @property
+    def base_url_args(self) -> dict:
+        return dict(host=self._host, port=self._port, protocol=self.protocol)
+
+
+class TokenAuth(BaseAuthMethod):
+    """
+    Token Auth Method for Evolve App Server python client when connecting to HTTPS servers.
+
+    Token Authentication may be configured in one of three ways:
+        - Providing an access token via the access_token parameter
+        - Specifying the client ID of the Auth0 application via the client_id parameter, plus one of the following:
+            - A username and password pair via the username and password parameters (account authentication)
+            - The client secret via the client_secret parameter (M2M authentication)
+          If this method is used, the auth configuration will be fetched from the Evolve App Server at the path
+          "/api/config/auth".
+        - Specifying a ZepbenTokenFetcher directly via the token_fetcher parameter
+
+    ..code-block:: python::
+
+        TokenAuth(access_token='...')
+        TokenAuth(token_fetcher='...')
+        TokenAuth(client_id='...' username='...' password='...')
+        TokenAuth(client_id='...', client_secret='...')
+
+    """
+    @overload
+    def __init__(self, host, port, protocol='https', verify_certificate=True, *, access_token: str):
+        """
+        :param access_token: The access token used for authentication, generated by Evolve App Server.
+        """
+        ...
+
+    @overload
+    def __init__(self, host, port, protocol='https', verify_certificate=True, *, token_fetcher: ZepbenTokenFetcher):
+        """
+        :param token_fetcher: A ZepbenTokenFetcher used to fetch auth tokens for access to the Evolve App Server.
+        """
+        ...
+
+    @overload
+    def __init__(self, host, port, protocol='https', verify_certificate=True, *, client_id: str, username: str, password: str, client_secret: Optional[str]):
+        """
+        :param client_id: The Auth0 client ID used to specify to the auth server which application to request a token for.
+        :param username: The username used for account authentication.
+        :param password: The password used for account authentication.
+        :param client_secret: The Auth0 client secret used for M2M authentication. (Optional)
+        """
+        ...
+
+    @overload
+    def __init__(self, host, port, protocol='https', verify_certificate=True, *, client_id: str, client_secret: str):
+        """
+        :param client_id: The Auth0 client ID used to specify to the auth server which application to request a token for.
+        :param client_secret: The Auth0 client secret used for M2M authentication.
+        """
+        ...
+
+    def __init__(self, host, port, protocol='https', verify_certificate=True, **kwargs):
+        if protocol != 'https':  # TODO: this exists because of an existing test, but given we can force it, we should
+            raise ValueError(
+                "Incompatible arguments passed to connect to secured Evolve App Server. "
+                "Authentication tokens must be sent via https. "
+                "To resolve this issue, exclude the \"protocol\" argument when initialising the EasClient.")
+
+        super().__init__(host, port, protocol, verify_certificate)
+        self._token_fetcher = None
+        self._access_token = None
+        self._init_func = None
+        self._configure(kwargs)
+
+    @property
+    def token(self) -> Optional[str]:
+        if self._access_token:
+            return f"Bearer {self._access_token}"
+        elif self._token_fetcher:
+            return self._token_fetcher.fetch_token()
+        raise AttributeError("access_token or token_fetcher method not configured")
+
+    def _configure(self, kwargs: dict):
+        """
+        Validates that the kwargs that end up being passed to the non-overloaded `__init__` method are of a valid
+        combination.
+        """
+        match list(kwargs.keys()):
+            case ['access_token']:
+                self._access_token = kwargs['access_token']
+            case ['token_fetcher']:
+                self._token_fetcher = kwargs['token_fetcher']
+            case ['client_id', 'client_secret', 'username', 'password']:
+                self._configure_client_id(**kwargs)
+            case ['client_id', 'username', 'password']:
+                self._configure_client_id(**kwargs)
+            case ['client_id', 'client_secret']:
+                self._configure_client_id(**kwargs)
+            case _:
+                raise ValueError("Incompatible arguments passed to connect to secured Evolve App Server.")
+
+        if kwargs.get('client_id'):
+            self._token_fetcher = create_token_fetcher(
+                conf_address=f"{self.protocol}://{self._host}:{self._port}/api/config/auth",
+                verify_conf=self.verify_certificate,
+            )
+
+    def _configure_client_id(
+        self, client_id: str = None,
+        username: str = None,
+        password: str = None,
+        client_secret: str = None
+    ):
+        self._token_fetcher = create_token_fetcher(
+            conf_address=f"{self.protocol}://{self._host}:{self._port}/api/config/auth",
+            verify_conf=self.verify_certificate,
+        )
+        if self._token_fetcher:
+            scope = (
+                'trusted' if self._token_fetcher.auth_method is AuthMethod.SELF else 'offline_access openid profile email0'
+            )
+
+            self._token_fetcher.token_request_data.update({
+                'client_id': client_id,
+                'scope': scope
+            })
+            self._token_fetcher.refresh_request_data.update({
+                "grant_type": "refresh_token",
+                'client_id': client_id,
+                'scope': scope
+            })
+            if username and password:
+                self._token_fetcher.token_request_data.update({
+                    'grant_type': 'password',
+                    'username': username,
+                    'password':
+                        sha256(password.encode('utf-8')).hexdigest()
+                        if self._token_fetcher.auth_method is AuthMethod.SELF
+                        else password
+                })
+                if client_secret:
+                    self._token_fetcher.token_request_data.update({'client_secret': client_secret})
+
+            elif client_secret:
+                self._token_fetcher.token_request_data.update({
+                    'grant_type': 'client_credentials',
+                    'client_secret': client_secret
+                })
+            else:
+                # Attempt azure managed identity (what a hack)
+                url = "http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01"
+                self._token_fetcher = create_token_fetcher_managed_identity(
+                    identity_url=f"{url}&resource={client_id}",
+                    verify_auth=self.verify_certificate
+                )

src/zepben/eas/client/decorators.py

@@ -0,0 +1,19 @@
+#  Copyright 2025 Zeppelin Bend Pty Ltd
+#
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+__all__ = ['catch_warnings']
+
+import functools
+import warnings
+from typing import Callable
+
+
+def catch_warnings(func: Callable) -> Callable:
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        with warnings.catch_warnings():
+            return func(*args, **kwargs)
+    return wrapper