Issue #389/#390 Add now_utc helper

Author: soxofaan

openeo_driver/dummy/dummy_backend.py

@@ -70,6 +70,7 @@
     AggregatePolygonSpatialResult,
 )
 from openeo_driver.users import User
+from openeo_driver.util.date_math import now_utc
 from openeo_driver.util.http import UrlSafeStructCodec
 from openeo_driver.utils import EvalEnv, WhiteListEvalEnv, generate_unique_id
 
@@ -704,7 +705,7 @@ def create_job(
             id=job_id,
             status=JOB_STATUS.CREATED,
             process=process,
-            created=datetime.datetime.now(tz=datetime.timezone.utc),
+            created=now_utc(),
             job_options=job_options,
             title=metadata.get("title"),
             description=metadata.get("description"),

openeo_driver/util/date_math.py

@@ -4,6 +4,14 @@
 import pandas as pd
 
 
+def now_utc() -> dt.datetime:
+    """
+    Current timezone-aware (UTC) datetime
+    to be used instead of deprecated naive datetime.utcnow()
+    """
+    return dt.datetime.now(tz=dt.timezone.utc)
+
+
 def month_shift(
         d: Union[dt.date, dt.datetime, pd.Timestamp],
         months: int = 1

openeo_driver/util/view_helpers.py

@@ -3,6 +3,7 @@
 
 import flask
 
+from openeo_driver.util.date_math import now_utc
 
 
 def cache_control(
@@ -27,7 +28,7 @@ def add_cache_control_headers(response: flask.Response) -> flask.Response:
             for key, value in settings.items():
                 setattr(response.cache_control, key, value)
             if max_age is not None:
-                response.expires = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=max_age)
+                response.expires = now_utc() + datetime.timedelta(seconds=max_age)
         return response
 
     def decorator(func):

openeo_driver/utils.py

@@ -14,6 +14,7 @@
 
 from openeo.util import Rfc3339, rfc3339
 from openeo.utils.version import ComparableVersion
+from openeo_driver.util.date_math import now_utc
 
 _log = logging.getLogger(__name__)
 
@@ -358,7 +359,7 @@ def generate_unique_id(prefix: Optional[str] = None, *, date_prefix: Union[bool,
     if date_prefix:
         if not isinstance(date_prefix, str):
             date_prefix = "%y%m%d%H%M%S"
-        date_repr = datetime.datetime.now(datetime.timezone.utc).strftime(date_prefix)
+        date_repr = now_utc().strftime(date_prefix)
         id = f"{date_repr}{id[len(date_repr):]}"
     if prefix:
         id = f"{prefix}-{id}"

tests/test_server.py

@@ -1,12 +1,12 @@
 import re
-import datetime
 
 from openeo_driver.server import build_backend_deploy_metadata
+from openeo_driver.util.date_math import now_utc
 
 
 def test_build_backend_deploy_metadata():
     data = build_backend_deploy_metadata(packages=["openeo", "openeo_driver", "foobarblerghbwop"])
-    assert data["date"].startswith(datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H"))
+    assert data["date"].startswith(now_utc().strftime("%Y-%m-%dT%H"))
     assert re.match(r"\d+\.\d+\.\d+", data["versions"]["openeo"])
     assert re.match(r"\d+\.\d+\.\d+", data["versions"]["openeo_driver"])
     assert data["versions"]["foobarblerghbwop"] == "n/a"

tests/test_workspace.py

@@ -6,6 +6,7 @@
 from pystac import Asset, Collection, Extent, Item, SpatialExtent, TemporalExtent, CatalogType, Link, RelType
 import pytest
 
+from openeo_driver.util.date_math import now_utc
 from openeo_driver.workspace import DiskWorkspace
 
 
@@ -200,7 +201,7 @@ def _collection(
         extent=Extent(spatial_extent, temporal_extent),
     )
 
-    item = Item(id=asset_filename, geometry=None, bbox=None, datetime=dt.datetime.now(dt.timezone.utc), properties={})
+    item = Item(id=asset_filename, geometry=None, bbox=None, datetime=now_utc(), properties={})
 
     asset_path = root_path / item.id / asset_filename
     asset = Asset(href=asset_path.name)  # relative to item

tests/util/test_date_math.py

@@ -4,7 +4,18 @@
 import pytest
 import time_machine
 
-from openeo_driver.util.date_math import month_shift, simple_job_progress_estimation
+from openeo_driver.util.date_math import month_shift, simple_job_progress_estimation, now_utc
+
+
+def test_now_utc_timezone():
+    now = now_utc()
+    assert now.tzinfo == dt.timezone.utc
+
+
+@time_machine.travel("2025-04-08T10:11:12+00")
+def test_now_utc():
+    now = now_utc()
+    assert now == dt.datetime(2025, 4, 8, 10, 11, 12, tzinfo=dt.timezone.utc)
 
 
 def test_month_shift_date():