add config field server_timezone (#84)

bfabric/bfabric_config.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
+import dataclasses
 import logging
 import os
-from typing import Optional, Dict, Tuple, Union
-import dataclasses
-import yaml
 from pathlib import Path
 
+import yaml
+
+from bfabric.src.errors import BfabricConfigError
+
 
 @dataclasses.dataclass(frozen=True)
 class BfabricAuth:
@@ -15,10 +17,10 @@ class BfabricAuth:
     login: str
     password: str
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"BfabricAuth(login={repr(self.login)}, password=...)"
 
-    def __str__(self):
+    def __str__(self) -> str:
         return repr(self)
 
 
@@ -29,25 +31,28 @@ class BfabricConfig:
         base_url (optional): The API base url
         application_ids (optional): Map of application names to ids.
         job_notification_emails (optional): Space-separated list of email addresses to notify when a job finishes.
+        server_timezone (optional): Timezone name of the server (used for queries)
     """
 
     def __init__(
         self,
-        base_url: Optional[str] = None,
-        application_ids: Optional[Dict[str, int]] = None,
-        job_notification_emails: Optional[str] = None
-    ):
+        base_url: str | None = None,
+        application_ids: dict[str, int] = None,
+        job_notification_emails: str | None = None,
+        server_timezone: str = "Europe/Zurich",
+    ) -> None:
         self._base_url = base_url or "https://fgcz-bfabric.uzh.ch/bfabric"
         self._application_ids = application_ids or {}
         self._job_notification_emails = job_notification_emails or ""
+        self._server_timezone = server_timezone
 
     @property
     def base_url(self) -> str:
         """The API base url."""
         return self._base_url
 
     @property
-    def application_ids(self) -> Dict[str, int]:
+    def application_ids(self) -> dict[str, int]:
         """Map of known application names to ids."""
         return self._application_ids
 
@@ -56,26 +61,33 @@ def job_notification_emails(self) -> str:
         """Space-separated list of email addresses to notify when a job finishes."""
         return self._job_notification_emails
 
+    @property
+    def server_timezone(self) -> str:
+        """Timezone name of the server (used for queries)."""
+        return self._server_timezone
+
     def copy_with(
         self,
-        base_url: Optional[str] = None,
-        application_ids: Optional[Dict[str, int]] = None,
+        base_url: str | None = None,
+        application_ids: dict[str, int] | None = None,
     ) -> BfabricConfig:
         """Returns a copy of the configuration with new values applied, if they are not None."""
         return BfabricConfig(
             base_url=base_url if base_url is not None else self.base_url,
-            application_ids=application_ids
-            if application_ids is not None
-            else self.application_ids,
+            application_ids=(application_ids if application_ids is not None else self.application_ids),
+            job_notification_emails=self.job_notification_emails,
+            server_timezone=self.server_timezone,
         )
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return (
             f"BfabricConfig(base_url={repr(self.base_url)}, application_ids={repr(self.application_ids)}, "
-            f"job_notification_emails={repr(self.job_notification_emails)})"
+            f"job_notification_emails={repr(self.job_notification_emails)}, "
+            f"server_timezone={repr(self.server_timezone)})"
         )
 
-def _read_config_env_as_dict(config_path: Union[str, Path], config_env: str = None) -> Tuple[str, dict]:
+
+def _read_config_env_as_dict(config_path: Path, config_env: str | None = None) -> tuple[str, dict]:
     """
     Reads and partially parses a bfabricpy.yml file
     :param config_path:  Path to the configuration file. It is assumed that it exists
@@ -86,39 +98,50 @@ def _read_config_env_as_dict(config_path: Union[str, Path], config_env: str = No
     logger = logging.getLogger(__name__)
     logger.info(f"Reading configuration from: {config_path}")
 
-    if os.path.splitext(config_path)[1] != '.yml':
-        raise IOError(f"Expected config file with .yml extension, got {config_path}")
+    if config_path.suffix != ".yml":
+        raise OSError(f"Expected config file with .yml extension, got {config_path}")
 
     # Read the config file
-    config_dict = yaml.safe_load(Path(config_path).read_text())
+    config_dict = yaml.safe_load(config_path.read_text())
 
-    if "GENERAL" not in config_dict:
-        raise IOError("Config file must have a general section")
-    if 'default_config' not in config_dict['GENERAL']:
-        raise IOError("Config file must provide a default environment")
-    config_env_default = config_dict['GENERAL']['default_config']
+    if "default_config" not in config_dict.get("GENERAL", {}):
+        raise BfabricConfigError("Config file must provide a `default_config` in the `GENERAL` section")
+    config_env_default = config_dict["GENERAL"]["default_config"]
 
     # Determine which environment we will use
     # By default, use the one provided by config_env
-    if config_env is None:
-        # Try to find a relevant
+    config_env = _select_config_env(
+        explicit_config_env=config_env, config_file_default_config=config_env_default, logger=logger
+    )
+    if config_env not in config_dict:
+        raise BfabricConfigError(f"The requested config environment {config_env} is not present in the config file")
+
+    return config_env, config_dict[config_env]
+
+
+def _select_config_env(explicit_config_env: str | None, config_file_default_config: str, logger: logging.Logger) -> str:
+    """Selects the appropriate configuration environment to use, based on the provided arguments.
+    :param explicit_config_env: Explicitly provided configuration environment to use (i.e. from a function argument)
+    :param config_file_default_config: Default configuration environment to use, as specified in the config file
+    :param logger: Logger to use for output
+    """
+    if explicit_config_env is None:
         config_env = os.getenv("BFABRICPY_CONFIG_ENV")
         if config_env is None:
-            logger.info(f"BFABRICPY_CONFIG_ENV not found, using default environment {config_env_default}")
-            config_env = config_env_default
+            logger.info(f"BFABRICPY_CONFIG_ENV not found, using default environment {config_file_default_config}")
+            config_env = config_file_default_config
         else:
             logger.info(f"found BFABRICPY_CONFIG_ENV = {config_env}")
     else:
+        config_env = explicit_config_env
         logger.info(f"config environment specified explicitly as {config_env}")
+    return config_env
 
-    if config_env not in config_dict:
-        raise IOError(f"The requested config environment {config_env} is not present in the config file")
 
-    return config_env, config_dict[config_env]
+def _have_all_keys(dict_: dict, expected_keys: list) -> bool:
+    """Returns True if all elements in list l are present as keys in dict d, otherwise false"""
+    return all(k in dict_ for k in expected_keys)
 
-def _have_all_keys(d: dict, l: list) -> bool:
-    """True if all elements in list l are present as keys in dict d, otherwise false"""
-    return all([k in d for k in l])
 
 def _parse_dict(d: dict, mandatory_keys: list, optional_keys: list = None, error_prefix: str = " ") -> dict:
     """
@@ -132,27 +155,27 @@ def _parse_dict(d: dict, mandatory_keys: list, optional_keys: list = None, error
     """
     missing_keys = set(mandatory_keys) - set(d)
     if missing_keys:
-        raise ValueError(f"{error_prefix}{missing_keys}")
+        raise BfabricConfigError(f"{error_prefix}{missing_keys}")
     result_keys = set(mandatory_keys) | set(optional_keys or [])
     d_rez = {k: d[k] for k in result_keys if k in d}
 
     # Ignore all other fields
     return d_rez
 
-def read_config(config_path: Union[str, Path], config_env: str = None,
-                optional_auth: bool = False) -> Tuple[BfabricConfig, Optional[BfabricAuth]]:
+
+def read_config(
+    config_path: str | Path,
+    config_env: str = None,
+) -> tuple[BfabricConfig, BfabricAuth | None]:
     """
     Reads bfabricpy.yml file, parses it, extracting authentication and configuration data
     :param config_path:   Path to the configuration file. It is assumed the file exists
     :param config_env:    Configuration environment to use. If not given, it is deduced.
-    :param optional_auth: Whether authentication is optional.
-        If not, both login and password must be present in the config file, otherwise an exception is thrown
-        If yes, missing login and password would result in authentication class being None, but no exception
     :return: Configuration and Authentication class instances
 
     NOTE: BFabricPy expects a .bfabricpy.yml of the format, as seen in bfabricPy/tests/unit/example_config.yml
     * The general field always has to be present
-    * There may be any number of environments, and they may have arbitrary names. Here, they are called PRODUCTION and TEST
+    * There may be any number of environments, with arbitrary names. Here, they are called PRODUCTION and TEST
     * Must specify correct login, password and base_url for each environment.
     * application and job_notification_emails fields are optional
     * The default environment will be selected as follows:
@@ -161,22 +184,24 @@ def read_config(config_path: Union[str, Path], config_env: str = None,
         - If not, finally, the parser will select the default_config specified in [GENERAL] of the .bfabricpy.yml file
     """
 
-
-    config_env_final, config_dict = _read_config_env_as_dict(config_path, config_env=config_env)
+    config_env_final, config_dict = _read_config_env_as_dict(Path(config_path), config_env=config_env)
 
     error_prefix = f"Config environment {config_env_final} does not have a compulsory field: "
 
-    # Parse authentification
-    if optional_auth and not _have_all_keys(config_dict, ['login', 'password']):
-        # Allow returning None auth if enabled
+    # Parse authentication
+    if not _have_all_keys(config_dict, ["login", "password"]):
         auth = None
     else:
-        auth_dict = _parse_dict(config_dict, ['login', 'password'], error_prefix=error_prefix)
+        auth_dict = _parse_dict(config_dict, ["login", "password"], error_prefix=error_prefix)
         auth = BfabricAuth(**auth_dict)
 
     # Parse config
-    config_dict = _parse_dict(config_dict, ['base_url'], optional_keys=['application_ids', 'job_notification_emails'],
-                              error_prefix=error_prefix)
+    config_dict = _parse_dict(
+        config_dict,
+        ["base_url"],
+        optional_keys=["application_ids", "job_notification_emails", "server_timezone"],
+        error_prefix=error_prefix,
+    )
     config = BfabricConfig(**config_dict)
 
     return config, auth

bfabric/src/errors.py

@@ -1,18 +1,25 @@
-from typing import List
+from __future__ import annotations
 
 
 class BfabricRequestError(Exception):
     """An error that is returned by the server in response to a full request."""
-    def __init__(self, message: str):
+
+    def __init__(self, message: str) -> None:
         self.message = message
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"RequestError(message={repr(self.message)})"
 
+
+class BfabricConfigError(RuntimeError):
+    """An error that is raised when the configuration is invalid."""
+    pass
+
+
 # TODO: Also test for response-level errors
-def get_response_errors(response, endpoint: str) -> List[BfabricRequestError]:
+def get_response_errors(response, endpoint: str) -> list[BfabricRequestError]:
     """
-    :param response:  A raw response to a query from an underlying engine 
+    :param response:  A raw response to a query from an underlying engine
     :param endpoint:  The target endpoint
     :return:          A list of errors for each query result, if that result failed
         Thus, a successful query would result in an empty list

bfabric/tests/unit/example_config.yml

@@ -15,6 +15,7 @@ TEST:
     Proteomics/DOG_552: 6
     Proteomics/DUCK_666: 12
   job_notification_emails: [email protected] [email protected]
+  server_timezone: UTC
 
 STANDBY:
   base_url: https://standby-server.uzh.ch/mystandby

bfabric/tests/unit/test_bfabric_config.py

@@ -2,7 +2,7 @@
 import unittest
 from pathlib import Path
 
-from bfabric.bfabric_config import BfabricConfig, BfabricAuth, read_config
+from bfabric.bfabric_config import BfabricAuth, BfabricConfig, read_config
 
 
 class TestBfabricAuth(unittest.TestCase):
@@ -22,6 +22,7 @@ def setUp(self):
         self.config = BfabricConfig(
             base_url="url",
             application_ids={"app": 1},
+            server_timezone="t/z",
         )
         self.example_config_path = Path(__file__).parent / "example_config.yml"
 
@@ -58,7 +59,7 @@ def test_copy_with_replaced_when_none(self):
     # TODO: Test that logging is consistent with initialization
     def test_read_yml_bypath_default(self):
         # Ensure environment variable is not available, and the default is environment is loaded
-        os.environ.pop('BFABRICPY_CONFIG_ENV', None)
+        os.environ.pop("BFABRICPY_CONFIG_ENV", None)
 
         config, auth = read_config(self.example_config_path)
         self.assertEqual("my_epic_production_login", auth.login)
@@ -80,7 +81,7 @@ def test_read_yml_bypath_environment_variable(self):
     # TODO: Test that logging is consistent with default config
     def test_read_yml_bypath_all_fields(self):
         with self.assertLogs(level="INFO") as log_context:
-            config, auth = read_config(self.example_config_path, config_env='TEST')
+            config, auth = read_config(self.example_config_path, config_env="TEST")
 
         # # Testing log
         # self.assertEqual(
@@ -96,42 +97,47 @@ def test_read_yml_bypath_all_fields(self):
         self.assertEqual("https://mega-test-server.uzh.ch/mytest", config.base_url)
 
         applications_dict_ground_truth = {
-            'Proteomics/CAT_123': 7,
-            'Proteomics/DOG_552': 6,
-            'Proteomics/DUCK_666': 12
+            "Proteomics/CAT_123": 7,
+            "Proteomics/DOG_552": 6,
+            "Proteomics/DUCK_666": 12,
         }
 
         job_notification_emails_ground_truth = "[email protected] [email protected]"
 
         self.assertEqual(applications_dict_ground_truth, config.application_ids)
         self.assertEqual(job_notification_emails_ground_truth, config.job_notification_emails)
+        self.assertEqual("UTC", config.server_timezone)
 
     # Testing that we can load base_url without authentication if correctly requested
     def test_read_yml_when_empty_optional(self):
         with self.assertLogs(level="INFO"):
-            config, auth = read_config(self.example_config_path, config_env='STANDBY', optional_auth=True)
+            config, auth = read_config(self.example_config_path, config_env="STANDBY")
 
         self.assertIsNone(auth)
         self.assertEqual("https://standby-server.uzh.ch/mystandby", config.base_url)
         self.assertEqual({}, config.application_ids)
         self.assertEqual("", config.job_notification_emails)
+        self.assertEqual("Europe/Zurich", config.server_timezone)
 
+    # TODO delete if no mandatory fields are reintroduced
     # Test that missing authentication will raise an error if required
-    def test_read_yml_when_empty_mandatory(self):
-        with self.assertRaises(ValueError):
-            read_config(self.example_config_path, config_env='STANDBY', optional_auth=False)
+    #def test_read_yml_when_empty_mandatory(self):
+    #    with self.assertRaises(BfabricConfigError):
+    #        read_config(self.example_config_path, config_env="STANDBY")
 
     def test_repr(self):
         rep = repr(self.config)
         self.assertEqual(
-            "BfabricConfig(base_url='url', application_ids={'app': 1}, job_notification_emails='')",
+            "BfabricConfig(base_url='url', application_ids={'app': 1}, "
+            "job_notification_emails='', server_timezone='t/z')",
             rep,
         )
 
     def test_str(self):
         rep = str(self.config)
         self.assertEqual(
-            "BfabricConfig(base_url='url', application_ids={'app': 1}, job_notification_emails='')",
+            "BfabricConfig(base_url='url', application_ids={'app': 1}, "
+            "job_notification_emails='', server_timezone='t/z')",
             rep,
         )