use Python 3.11 features: StrEnum and datetime.UTC

Manifest.Operation now derives from enum.StrEnum, so its members are
real str instances; drop the .value indirection in the feature-flag
lookups.

Replace timezone.utc with the datetime.UTC alias (3.11) across the
non-test modules and drop the now-unused timezone imports.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/borg/archiver/__init__.py

@@ -24,7 +24,7 @@ try:
     import os
     import shlex
     import signal
-    from datetime import datetime, timezone
+    from datetime import UTC, datetime
 
     from ..logger import create_logger, setup_logging
 
@@ -386,7 +386,7 @@ class Archiver(
         # thus we have to initialize replace_placeholders here and process all args that need placeholder replacement.
         if getattr(args, "timestamp", None):
             replace_placeholders.override("now", DatetimeWrapper(args.timestamp))
-            replace_placeholders.override("utcnow", DatetimeWrapper(args.timestamp.astimezone(timezone.utc)))
+            replace_placeholders.override("utcnow", DatetimeWrapper(args.timestamp.astimezone(UTC)))
             args.location = args.location.with_timestamp(args.timestamp)
         for name in "name", "other_name", "newname", "comment":
             value = getattr(args, name, None)

src/borg/archiver/prune_cmd.py

@@ -1,5 +1,5 @@
 from collections import OrderedDict
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 import logging
 from operator import attrgetter
 import os
@@ -18,7 +18,7 @@ logger = create_logger()
 
 
 def prune_within(archives, seconds, kept_because):
-    target = datetime.now(timezone.utc) - timedelta(seconds=seconds)
+    target = datetime.now(UTC) - timedelta(seconds=seconds)
     kept_counter = 0
     result = []
     for a in archives:

src/borg/cache.py

@@ -5,7 +5,7 @@ import os
 import shutil
 import stat
 from collections import namedtuple
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from time import perf_counter
 
@@ -428,7 +428,7 @@ class FilesCacheMixin:
                         entries += 1
             integrity_data = fd.integrity_data
         files_cache_logger.debug(f"FILES-CACHE-KILL: removed {age_discarded} entries with age >= TTL [{ttl}]")
-        t_str = datetime.fromtimestamp(discard_after / 1e9, timezone.utc).isoformat()
+        t_str = datetime.fromtimestamp(discard_after / 1e9, UTC).isoformat()
         files_cache_logger.debug(f"FILES-CACHE-KILL: removed {race_discarded} entries with ctime/mtime >= {t_str}")
         files_cache_logger.debug(f"FILES-CACHE-SAVE: finished, {entries} remaining entries saved.")
         return integrity_data
@@ -670,9 +670,9 @@ class ChunksMixin:
 
     def __init__(self):
         self._chunks = None
-        self.last_refresh_dt = datetime.now(timezone.utc)
+        self.last_refresh_dt = datetime.now(UTC)
         self.refresh_td = timedelta(seconds=60)
-        self.chunks_cache_last_write = datetime.now(timezone.utc)
+        self.chunks_cache_last_write = datetime.now(UTC)
         self.chunks_cache_write_td = timedelta(seconds=600)
 
     @property
@@ -719,7 +719,7 @@ class ChunksMixin:
                 size = len(data)  # data is still uncompressed
             else:
                 raise ValueError("when giving compressed data for a chunk, the uncompressed size must be given also")
-        now = datetime.now(timezone.utc)
+        now = datetime.now(UTC)
         self._maybe_write_chunks_cache(now)
         exists = self.seen_chunk(id, size)
         if exists:
@@ -845,7 +845,7 @@ class AdHocWithFilesCache(FilesCacheMixin, ChunksMixin):
                 logger.debug(f"Chunks index stats: {key}: {value}")
             pi.output("Saving chunks cache")
             # note: cache/chunks.* in repo has a different integrity mechanism
-            now = datetime.now(timezone.utc)
+            now = datetime.now(UTC)
             self._maybe_write_chunks_cache(now, force=True, clear=True)
             self._chunks = None  # nothing there (cleared!)
         pi.output("Saving cache config")

src/borg/helpers/parseformat.py

@@ -12,7 +12,7 @@ import uuid
 from pathlib import Path
 from typing import ClassVar, Any, TYPE_CHECKING, Literal
 from collections import OrderedDict
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from functools import partial
 from hashlib import sha256
 from string import Formatter
@@ -364,7 +364,7 @@ def _replace_placeholders(text, overrides={}):
     """Replace placeholders in text with their values."""
     from ..platform import fqdn, hostname, getosusername
 
-    current_time = datetime.now(timezone.utc)
+    current_time = datetime.now(UTC)
     data = {
         "pid": os.getpid(),
         "fqdn": fqdn,
@@ -697,10 +697,7 @@ class Location:
         # note: this only affects the repository URL/path, not the archive name!
         return Location(
             self.raw,
-            overrides={
-                "now": DatetimeWrapper(timestamp),
-                "utcnow": DatetimeWrapper(timestamp.astimezone(timezone.utc)),
-            },
+            overrides={"now": DatetimeWrapper(timestamp), "utcnow": DatetimeWrapper(timestamp.astimezone(UTC))},
         )
 
 

src/borg/helpers/time.py

@@ -1,9 +1,9 @@
 import os
 import re
-from datetime import datetime, timezone, timedelta
+from datetime import UTC, datetime, timedelta
 
 
-def parse_timestamp(timestamp, tzinfo=timezone.utc):
+def parse_timestamp(timestamp, tzinfo=UTC):
     """Parse an ISO 8601 timestamp string.
 
     For naive/unaware datetime objects, assume they are in the tzinfo timezone (default: UTC).
@@ -26,7 +26,7 @@ def parse_local_timestamp(timestamp, tzinfo=None):
     return dt
 
 
-_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
+_EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
 
 
 def utcfromtimestampns(ts_ns: int) -> datetime:
@@ -196,4 +196,4 @@ class OutputTimestamp:
 
 def archive_ts_now():
     """return tz-aware datetime obj for current time for usage as archive timestamp"""
-    return datetime.now(timezone.utc)  # utc time / utc timezone
+    return datetime.now(UTC)  # utc time / utc timezone

src/borg/manifest.py

@@ -1,7 +1,7 @@
 import enum
 import re
 from collections import namedtuple
-from datetime import datetime, timedelta, timezone
+from datetime import UTC, datetime, timedelta
 from operator import attrgetter
 from collections.abc import Iterator, Sequence
 from typing import Protocol, runtime_checkable
@@ -430,7 +430,7 @@ class Archives:
 
 class Manifest:
     @enum.unique
-    class Operation(enum.Enum):
+    class Operation(enum.StrEnum):
         # The comments here only roughly describe the scope of each feature. In the end, additions need to be
         # based on potential problems older clients could produce when accessing newer repositories and the
         # trade-offs of locking version out or still allowing access. As all older versions and their exact
@@ -514,9 +514,9 @@ class Manifest:
             feature_flags = self.config.get("feature_flags", None)
             if feature_flags is None:
                 return
-            if operation.value not in feature_flags:
+            if operation not in feature_flags:
                 continue
-            requirements = feature_flags[operation.value]
+            requirements = feature_flags[operation]
             if "mandatory" in requirements:
                 unsupported = set(requirements["mandatory"]) - self.SUPPORTED_REPO_FEATURES
                 if unsupported:
@@ -538,10 +538,10 @@ class Manifest:
 
         # self.timestamp needs to be strictly monotonically increasing. Clocks often are not set correctly
         if self.timestamp is None:
-            self.timestamp = datetime.now(tz=timezone.utc).isoformat(timespec="microseconds")
+            self.timestamp = datetime.now(tz=UTC).isoformat(timespec="microseconds")
         else:
             incremented_ts = self.last_timestamp + timedelta(microseconds=1)
-            now_ts = datetime.now(tz=timezone.utc)
+            now_ts = datetime.now(tz=UTC)
             max_ts = max(incremented_ts, now_ts)
             self.timestamp = max_ts.isoformat(timespec="microseconds")
         # include checks for limits as enforced by limited unpacker (used by load())

src/borg/storelocking.py

@@ -96,7 +96,7 @@ class Lock:
 
     def _create_lock(self, *, exclusive=None, update_last_refresh=False):
         assert exclusive is not None
-        now = datetime.datetime.now(datetime.timezone.utc)
+        now = datetime.datetime.now(datetime.UTC)
         timestamp = now.isoformat(timespec="milliseconds")
         lock = dict(exclusive=exclusive, hostid=self.id[0], processid=self.id[1], threadid=self.id[2], time=timestamp)
         value = json.dumps(lock).encode("utf-8")
@@ -123,7 +123,7 @@ class Lock:
         return self.id == (lock["hostid"], lock["processid"], lock["threadid"])
 
     def _is_stale_lock(self, lock):
-        now = datetime.datetime.now(datetime.timezone.utc)
+        now = datetime.datetime.now(datetime.UTC)
         if now > lock["dt"] + self.stale_td:
             logger.debug(f"LOCK-STALE: lock is too old, it was not refreshed. lock: {lock}.")
             return True
@@ -247,7 +247,7 @@ class Lock:
 
     def refresh(self):
         """Refreshes the lock; call this frequently, but not later than every <stale> seconds."""
-        now = datetime.datetime.now(datetime.timezone.utc)
+        now = datetime.datetime.now(datetime.UTC)
         if self.last_refresh_dt is not None and now > self.last_refresh_dt + self.refresh_td:
             old_locks = self._find_locks(only_mine=True)
             if len(old_locks) == 0: