Add datetime type with guaranteed UTC timezone (#76)

- Add pydantic datetime type DateTimeUTC with guaranteed UTC timezone
- Add assert_tz_is_utc() to verify that UTC is the default timezone
- Add now_as_utc() to create the current datetime with UTC timezone
- Add some more type hints to the utils module and specify exported names in __all__

Bumps version 0.16.0.

Co-authored-by:  Kersten Breuer <[email protected]>

examples/api/hello_world_web_server/__main__.py

@@ -18,13 +18,15 @@
 import asyncio
 
 from ghga_service_chassis_lib.api import run_server
+from ghga_service_chassis_lib.utils import assert_tz_is_utc
 
 from .api import app  # noqa: F401 pylint: disable=unused-import
 from .config import get_config
 
 
 def run():
     """Run the service"""
+    assert_tz_is_utc()
     asyncio.run(
         run_server(app="hello_world_web_server.__main__:app", config=get_config())
     )

ghga_service_chassis_lib/__init__.py

@@ -15,4 +15,4 @@
 
 """A library that contains the basic chassis functionality used in services of GHGA"""
 
-__version__ = "0.15.2"
+__version__ = "0.16.0"

ghga_service_chassis_lib/utils.py

@@ -15,14 +15,33 @@
 
 """General utilities that don't require heavy dependencies."""
 
+from __future__ import annotations
+
 import os
 import signal
+from abc import ABC
 from contextlib import contextmanager
+from datetime import datetime, timezone
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from typing import Callable, Optional
+from typing import Any, BinaryIO, Callable, Generator, Optional, TypeVar, cast
+
+from pydantic import BaseSettings, parse_obj_as
+
+__all__ = [
+    "AsyncDaoGenericBase",
+    "DaoGenericBase",
+    "DateTimeUTC",
+    "OutOfContextError",
+    "UTC",
+    "big_temp_file",
+    "assert_tz_is_utc",
+    "create_fake_drs_uri",
+    "exec_with_timeout",
+    "now_as_utc",
+]
 
-from pydantic import BaseSettings
+T = TypeVar("T")
 
 TEST_FILE_DIR = Path(__file__).parent.resolve() / "test_files"
 
@@ -32,6 +51,8 @@
     if filename.startswith("test_") and filename.endswith(".yaml")
 ]
 
+UTC = timezone.utc
+
 
 class OutOfContextError(RuntimeError):
     """Thrown when a context manager is used out of context."""
@@ -86,17 +107,17 @@ async def _aexit__(self, err_type, err_value, err_traceback):
         ...
 
 
-def raise_timeout_error(_, __):
+def raise_timeout_error(_signalnum, _handler) -> None:
     """Raise a TimeoutError"""
     raise TimeoutError()
 
 
 def exec_with_timeout(
-    func: Callable,
+    func: Callable[..., T],
     timeout_after: int,
     func_args: Optional[list] = None,
     func_kwargs: Optional[dict] = None,
-):
+) -> T:
     """
     Exec a function (`func`) with a specified timeout (`timeout_after` in seconds).
     If the function doesn't finish before the timeout, a TimeoutError is thrown.
@@ -118,13 +139,19 @@ def exec_with_timeout(
     return result
 
 
-def create_fake_drs_uri(object_id: str):
+def create_fake_drs_uri(object_id: str) -> str:
     """Create a fake DRS URI based on an object id."""
     return f"drs://www.example.org/{object_id}"
 
 
+class NamedBinaryIO(ABC, BinaryIO):
+    """Return type of NamedTemporaryFile."""
+
+    name: str
+
+
 @contextmanager
-def big_temp_file(size: int):
+def big_temp_file(size: int) -> Generator[NamedBinaryIO, None, None]:
     """Generates a big file with approximately the specified size in bytes."""
     current_size = 0
     current_number = 0
@@ -138,4 +165,51 @@ def big_temp_file(size: int):
             current_number = next_number
             next_number = previous_number + current_number
         temp_file.flush()
-        yield temp_file
+        yield cast(NamedBinaryIO, temp_file)
+
+
+class DateTimeUTC(datetime):
+    """A pydantic type for values that should have an UTC timezone.
+
+    This behaves exactly like the normal datetime type, but requires that the value
+    has a timezone and converts the timezone to UTC if necessary.
+    """
+
+    @classmethod
+    def construct(cls, *args, **kwargs) -> DateTimeUTC:
+        """Construct a datetime with UTC timezone."""
+        if kwargs.get("tzinfo") is None:
+            kwargs["tzinfo"] = UTC
+        return cls(*args, **kwargs)
+
+    @classmethod
+    def __get_validators__(cls) -> Generator[Callable[[Any], datetime], None, None]:
+        """Get all validators."""
+        yield cls.validate
+
+    @classmethod
+    def validate(cls, value: Any) -> datetime:
+        """Validate the given value."""
+        date_value = parse_obj_as(datetime, value)
+        if date_value.tzinfo is None:
+            raise ValueError(f"Date-time value is missing a timezone: {value!r}")
+        if date_value.tzinfo is not UTC:
+            date_value = date_value.astimezone(UTC)
+        return date_value
+
+
+def assert_tz_is_utc() -> None:
+    """Verifies that the default timezone is set to UTC.
+
+    Raises a Runtimeerror if the default timezone is set differently.
+    """
+    if datetime.now().astimezone().tzinfo != UTC:
+        raise RuntimeError("System must be configured to use UTC.")
+
+
+def now_as_utc() -> DateTimeUTC:
+    """Return the current datetime with UTC timezone.
+
+    Note: This is different from datetime.utcnow() which has no timezone.
+    """
+    return DateTimeUTC.now(UTC)

tests/integration/test_s3.py

@@ -61,11 +61,11 @@ def test_typical_workflow(
         big_temp_file(size=20 * MEBIBYTE) if use_multipart_upload else nullcontext()
     ) as temp_file:
         object_fixture = (
-            ObjectFixture(
+            s3_fixture.non_existing_objects[0]
+            if temp_file is None
+            else ObjectFixture(
                 file_path=temp_file.name, bucket_id="", object_id="some-big-file"
             )
-            if use_multipart_upload
-            else s3_fixture.non_existing_objects[0]
         )
 
         typical_workflow(

tests/unit/test_utils.py

@@ -0,0 +1,122 @@
+# Copyright 2021 - 2022 Universität Tübingen, DKFZ and EMBL
+# for the German Human Genome-Phenome Archive (GHGA)
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test the utils module."""
+
+from datetime import datetime, timedelta, timezone
+from zoneinfo import ZoneInfo
+
+from pydantic import BaseModel
+from pytest import mark, raises
+
+from ghga_service_chassis_lib.utils import UTC, DateTimeUTC, now_as_utc
+
+
+@mark.parametrize(
+    "value",
+    [
+        "2022-11-15 12:00:00",
+        "2022-11-15T12:00:00",
+        datetime(2022, 11, 15, 12, 0, 0),
+        datetime.now(),
+        datetime.utcnow(),
+        datetime.utcfromtimestamp(0),
+    ],
+)
+def test_does_not_accept_naive_datetimes(value):
+    """Test that DateTimeUTC does not accept naive datetimes."""
+
+    class Model(BaseModel):
+        """Test model"""
+
+        d: DateTimeUTC
+
+    with raises(ValueError, match="missing a timezone"):
+        Model(d=value)
+
+
+@mark.parametrize(
+    "value",
+    [
+        "2022-11-15T12:00:00+00:00",
+        "2022-11-15T12:00:00Z",
+        datetime(2022, 11, 15, 12, 0, 0, tzinfo=UTC),
+        datetime.now(timezone.utc),
+        datetime.fromtimestamp(0, UTC),
+    ],
+)
+def test_accept_aware_datetimes_in_utc(value):
+    """Test that DateTimeUTC does not accepts timezone aware UTC datetimes."""
+
+    class Model(BaseModel):
+        """Test model"""
+
+        dt: datetime
+        du: DateTimeUTC
+
+    model = Model(dt=value, du=value)
+
+    assert model.dt == model.du
+
+
+@mark.parametrize(
+    "value",
+    [
+        "2022-11-15T12:00:00+03:00",
+        "2022-11-15T12:00:00-03:00",
+        datetime(2022, 11, 15, 12, 0, 0, tzinfo=ZoneInfo("America/Los_Angeles")),
+        datetime.now(ZoneInfo("Asia/Tokyo")),
+    ],
+)
+def test_converts_datetimes_to_utc(value):
+    """Test that DateTimeUTC converts other time zones to UTC."""
+
+    class Model(BaseModel):
+        """Test model"""
+
+        dt: datetime
+        du: DateTimeUTC
+
+    model = Model(dt=value, du=value)
+
+    assert model.dt.tzinfo is not None
+    assert model.dt.tzinfo is not UTC
+    assert model.dt.utcoffset() != timedelta(0)
+    assert model.du.tzinfo is UTC
+    assert model.du.utcoffset() == timedelta(0)
+
+    assert model.dt == model.du
+
+
+def test_datetime_utc_constructor():
+    """Test the constructor for DateTimeUTC values."""
+
+    date = DateTimeUTC.construct(2022, 11, 15, 12, 0, 0)
+    assert isinstance(date, DateTimeUTC)
+    assert date.tzinfo is UTC
+    assert date.utcoffset() == timedelta(0)
+
+    date = DateTimeUTC.construct(2022, 11, 15, 12, 0, 0, tzinfo=UTC)
+    assert isinstance(date, DateTimeUTC)
+    assert date.tzinfo is UTC
+    assert date.utcoffset() == timedelta(0)
+
+
+def test_now_as_utc():
+    """Test the now_as_utc function."""
+    assert isinstance(now_as_utc(), DateTimeUTC)
+    assert now_as_utc().tzinfo is UTC
+    assert now_as_utc().utcoffset() == timedelta(0)
+    assert abs(now_as_utc().timestamp() - datetime.now().timestamp()) < 5