#495 #496 Python 3.12 requires datetime.now(timezone.utc) i.s.o. datetime.now()

Author: justb4

GeoHealthCheck/healthcheck.py

@@ -27,7 +27,7 @@
 #
 # =================================================================
 
-from datetime import datetime
+from datetime import datetime, timezone
 import logging
 import json
 from urllib.request import urlopen
@@ -95,7 +95,7 @@ def run_resource(resourceid):
     # Run test
     result = run_test_resource(resource)
 
-    run1 = Run(resource, result, datetime.utcnow())
+    run1 = Run(resource, result, datetime.now(timezone.utc))
 
     DB.session.add(run1)
 
@@ -144,7 +144,7 @@ def sniff_test_resource(config, resource_type, url):
         raise RuntimeError(msg2)
 
     title = None
-    start_time = datetime.utcnow()
+    start_time = datetime.now(timezone.utc)
     message = None
     resource_type_map = {'OGC:WMS': [partial(WebMapService, version='1.3.0'),
                                      partial(WebMapService, version='1.1.1')],
@@ -223,7 +223,7 @@ def sniff_test_resource(config, resource_type, url):
             title = urlparse(url).hostname
         elif resource_type == 'OSGeo:GeoNode':
             endpoints = ows
-            end_time = datetime.utcnow()
+            end_time = datetime.now(timezone.utc)
             delta = end_time - start_time
             response_time = '%s.%s' % (delta.seconds, delta.microseconds)
             base_tags = geonode_make_tags(url)
@@ -259,7 +259,7 @@ def sniff_test_resource(config, resource_type, url):
         message = msg
         success = False
 
-    end_time = datetime.utcnow()
+    end_time = datetime.now(timezone.utc)
 
     delta = end_time - start_time
     response_time = '%s.%s' % (delta.seconds, delta.microseconds)
@@ -311,10 +311,10 @@ def geonode_make_tags(base_url):
 
 if __name__ == '__main__':
     print('START - Running health check tests on %s'
-          % datetime.utcnow().isoformat())
+          % datetime.now(timezone.utc).isoformat())
     run_resources()
     print('END - Running health check tests on %s'
-          % datetime.utcnow().isoformat())
+          % datetime.now(timezone.utc).isoformat())
     # from init import App
     # if len(sys.argv) < 3:
     #     print('Usage: %s <resource_type> <url>' % sys.argv[0])

GeoHealthCheck/models.py

@@ -31,7 +31,7 @@
 import json
 import logging
 from flask_babel import gettext as _
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
 from sqlalchemy import func, and_
 
@@ -58,7 +58,7 @@ def flush_runs():
     all_runs = Run.query.all()
     run_count = 0
     for run in all_runs:
-        days_old = (datetime.utcnow() - run.checked_datetime).days
+        days_old = (datetime.now(timezone.utc) - run.checked_datetime).days
         if days_old > retention_days:
             run_count += 1
             DB.session.delete(run)
@@ -87,7 +87,7 @@ class Run(DB.Model):
     report = deferred(DB.Column(DB.Text, default={}))
 
     def __init__(self, resource, result,
-                 checked_datetime=datetime.utcnow()):
+                 checked_datetime=datetime.now(timezone.utc)):
         self.resource = resource
         self.success = result.success
         self.response_time = result.response_time_str
@@ -680,13 +680,13 @@ def __init__(self, resource, owner, interval_mins):
         self.init_datetimes(interval_mins)
 
     def init_datetimes(self, interval_mins):
-        self.start_time = datetime.utcnow()
+        self.start_time = datetime.now(timezone.utc)
         # Subtract some space from end-time to allow obtain at scheduled time
         minutes = interval_mins - 1
         self.end_time = self.start_time + timedelta(minutes=minutes)
 
     def has_expired(self):
-        now = datetime.utcnow()
+        now = datetime.now(timezone.utc)
         return now > self.end_time
 
     def obtain(self, owner, frequency):
@@ -723,7 +723,7 @@ def __init__(self, username, password, email, role='user'):
         self.set_password(password)
         self.email = email
         self.role = role
-        self.registered_on = datetime.utcnow()
+        self.registered_on = datetime.now(timezone.utc)
 
     def authenticate(self, password):
         return util.verify_hash(password, self.password)

GeoHealthCheck/probe.py

@@ -128,7 +128,7 @@ def get_metadata_cached(self, resource, version='any'):
         metadata = None
         if key in Probe.METADATA_CACHE:
             entry = Probe.METADATA_CACHE[key]
-            delta = datetime.datetime.utcnow() - entry['time']
+            delta = datetime.now(timezone.utc) - entry['time']
             metadata = entry['metadata']
 
             # Don't keep cache forever, refresh every N mins
@@ -144,7 +144,7 @@ def get_metadata_cached(self, resource, version='any'):
                 # Store entry with time, for expiry later
                 entry = {
                     "metadata": metadata,
-                    "time": datetime.datetime.utcnow()
+                    "time": datetime.now(timezone.utc)
                 }
                 Probe.METADATA_CACHE[key] = entry
 

GeoHealthCheck/result.py

@@ -1,4 +1,4 @@
-import datetime
+from datetime import datetime, timezone
 
 
 class Result(object):
@@ -36,10 +36,10 @@ def set(self, success, message):
         self.message = message
 
     def start(self):
-        self.start_time = datetime.datetime.utcnow()
+        self.start_time = datetime.now(timezone.utc)
 
     def stop(self):
-        self.end_time = datetime.datetime.utcnow()
+        self.end_time = datetime.now(timezone.utc)
 
         delta = self.end_time - self.start_time
         self.response_time_secs = delta.seconds