RF: Centralize to have a helper utcnow()

yarikoptic · yarikoptic · commit fce346eb4abc · 2024-09-13T19:54:37.000-04:00
diff --git a/nipype/interfaces/base/support.py b/nipype/interfaces/base/support.py
@@ -11,11 +11,12 @@
 from copy import deepcopy
 from textwrap import wrap
 import re
-from datetime import datetime as dt, UTC
+from datetime import datetime as dt
 from dateutil.parser import parse as parseutc
 import platform
 
 from ... import logging, config
+from ...utils.datetime import utcnow
 from ...utils.misc import is_container, rgetcwd
 from ...utils.filemanip import md5, hash_infile
 
@@ -72,15 +73,15 @@ def __enter__(self):
         if self._runtime.redirect_x:
             self._runtime.environ["DISPLAY"] = config.get_display()
 
-        self._runtime.startTime = dt.isoformat(dt.now(UTC))
+        self._runtime.startTime = dt.isoformat(utcnow())
         self._resmon.start()
         # TODO: Perhaps clean-up path and ensure it exists?
         os.chdir(self._runtime.cwd)
         return self._runtime
 
     def __exit__(self, exc_type, exc_value, exc_tb):
         """Tear-down interface execution."""
-        self._runtime.endTime = dt.isoformat(dt.now(UTC))
+        self._runtime.endTime = dt.isoformat(utcnow())
         timediff = parseutc(self._runtime.endTime) - parseutc(self._runtime.startTime)
         self._runtime.duration = (
             timediff.days * 86400 + timediff.seconds + timediff.microseconds / 1e6
diff --git a/nipype/pipeline/engine/workflows.py b/nipype/pipeline/engine/workflows.py
@@ -8,14 +8,14 @@
 import os
 import os.path as op
 import sys
-from datetime import datetime, UTC
 from copy import deepcopy
 import pickle
 import shutil
 
 import numpy as np
 
 from ... import config, logging
+from ...utils.datetime import utcnow
 from ...utils.misc import str2bool
 from ...utils.functions import getsource, create_function_from_source
 
@@ -628,7 +628,7 @@ def run(self, plugin=None, plugin_args=None, updatehash=False):
         if str2bool(self.config["execution"]["create_report"]):
             self._write_report_info(self.base_dir, self.name, execgraph)
         runner.run(execgraph, updatehash=updatehash, config=self.config)
-        datestr = datetime.now(UTC).strftime("%Y%m%dT%H%M%S")
+        datestr = utcnow().strftime("%Y%m%dT%H%M%S")
         if str2bool(self.config["execution"]["write_provenance"]):
             prov_base = op.join(self.base_dir, "workflow_provenance_%s" % datestr)
             logger.info("Provenance file prefix: %s" % prov_base)
diff --git a/nipype/utils/datetime.py b/nipype/utils/datetime.py
@@ -0,0 +1,19 @@
+# emacs: -*- mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
+# vi: set ft=python sts=4 ts=4 sw=4 et:
+"""
+Utilities for dates and time
+"""
+
+from datetime import datetime as dt
+import sys
+
+if sys.version_info >= (3, 11):
+    from datetime import UTC
+
+    def utcnow():
+        """Adapter since 3.12 prior utcnow is deprecated,
+        but not EOLed 3.8 does not have datetime.UTC"""
+        return dt.now(UTC)
+
+else:
+    utcnow = dt.utcnow
