use py37+ datetime.isoformat / .fromisoformat

since python 3.7, .isoformat() is usable IF timespec != "auto"
is given ("auto" [default] would be as evil as before, sometimes
formatting with, sometimes without microseconds).

also since python 3.7, there is now .fromisoformat().
This commit is contained in:
Thomas Waldmann 2022-08-11 21:18:56 +02:00
parent b392a60c08
commit bab68a8d25
6 changed files with 29 additions and 39 deletions

View File

@ -644,8 +644,8 @@ Duration: {0.duration}
"cmdline": sys.argv,
"hostname": hostname,
"username": getuser(),
"time": start.strftime(ISO_FORMAT),
"time_end": end.strftime(ISO_FORMAT),
"time": start.isoformat(timespec="microseconds"),
"time_end": end.isoformat(timespec="microseconds"),
"chunker_params": self.chunker_params,
}
if stats is not None:

View File

@ -102,9 +102,6 @@ EXIT_WARNING = 1 # reached normal end of operation, but there were issues
EXIT_ERROR = 2 # terminated abruptly, did not reach end of operation
EXIT_SIGNAL_BASE = 128 # terminated due to signal, rc = 128 + sig_no
# never use datetime.isoformat(), it is evil. always use one of these:
# datetime.strftime(ISO_FORMAT) # output always includes .microseconds
# datetime.strftime(ISO_FORMAT_NO_USECS) # output never includes microseconds
ISO_FORMAT_NO_USECS = "%Y-%m-%dT%H:%M:%S"
ISO_FORMAT = ISO_FORMAT_NO_USECS + ".%f"

View File

@ -65,7 +65,7 @@ class Archives(abc.MutableMapping):
id, ts = info
assert isinstance(id, bytes)
if isinstance(ts, datetime):
ts = ts.replace(tzinfo=None).strftime(ISO_FORMAT)
ts = ts.replace(tzinfo=None).isoformat(timespec="microseconds")
assert isinstance(ts, str)
self._archives[name] = {"id": id, "time": ts}
@ -254,11 +254,11 @@ class Manifest:
self.config["tam_required"] = True
# self.timestamp needs to be strictly monotonically increasing. Clocks often are not set correctly
if self.timestamp is None:
self.timestamp = datetime.utcnow().strftime(ISO_FORMAT)
self.timestamp = datetime.utcnow().isoformat(timespec="microseconds")
else:
prev_ts = self.last_timestamp
incremented = (prev_ts + timedelta(microseconds=1)).strftime(ISO_FORMAT)
self.timestamp = max(incremented, datetime.utcnow().strftime(ISO_FORMAT))
incremented = (prev_ts + timedelta(microseconds=1)).isoformat(timespec="microseconds")
self.timestamp = max(incremented, datetime.utcnow().isoformat(timespec="microseconds"))
# include checks for limits as enforced by limited unpacker (used by load())
assert len(self.archives) <= MAX_ARCHIVES
assert all(len(name) <= 255 for name in self.archives)

View File

@ -2,8 +2,6 @@ import os
import time
from datetime import datetime, timezone
from ..constants import ISO_FORMAT, ISO_FORMAT_NO_USECS
def to_localtime(ts):
"""Convert datetime object from UTC to local time zone"""
@ -12,8 +10,7 @@ def to_localtime(ts):
def parse_timestamp(timestamp, tzinfo=timezone.utc):
"""Parse a ISO 8601 timestamp string"""
fmt = ISO_FORMAT if "." in timestamp else ISO_FORMAT_NO_USECS
dt = datetime.strptime(timestamp, fmt)
dt = datetime.fromisoformat(timestamp)
if tzinfo is not None:
dt = dt.replace(tzinfo=tzinfo)
return dt
@ -26,22 +23,11 @@ def timestamp(s):
ts = safe_s(os.stat(s).st_mtime)
return datetime.fromtimestamp(ts, tz=timezone.utc)
except OSError:
# didn't work, try parsing as timestamp. UTC, no TZ, no microsecs support.
for format in (
"%Y-%m-%dT%H:%M:%SZ",
"%Y-%m-%dT%H:%M:%S+00:00",
"%Y-%m-%dT%H:%M:%S",
"%Y-%m-%d %H:%M:%S",
"%Y-%m-%dT%H:%M",
"%Y-%m-%d %H:%M",
"%Y-%m-%d",
"%Y-%j",
):
try:
return datetime.strptime(s, format).replace(tzinfo=timezone.utc)
except ValueError:
continue
raise ValueError
# didn't work, try parsing as a ISO timestamp. if no TZ is given, we assume UTC.
dt = datetime.fromisoformat(s)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
# Not too rarely, we get crappy timestamps from the fs, that overflow some computations.
@ -106,7 +92,7 @@ def isoformat_time(ts: datetime):
Format *ts* according to ISO 8601.
"""
# note: first make all datetime objects tz aware before adding %z here.
return ts.strftime(ISO_FORMAT)
return ts.isoformat(timespec="microseconds")
def format_timedelta(td):

View File

@ -656,7 +656,9 @@ class Repository:
if self.append_only:
with open(os.path.join(self.path, "transactions"), "a") as log:
print(
"transaction %d, UTC time %s" % (transaction_id, datetime.utcnow().strftime(ISO_FORMAT)), file=log
"transaction %d, UTC time %s"
% (transaction_id, datetime.utcnow().isoformat(timespec="microseconds")),
file=log,
)
# Write hints file

View File

@ -249,6 +249,11 @@ def test_disk_full(cmd):
assert rc == EXIT_SUCCESS
def checkts(ts):
# check if the timestamp is in the expected format
assert datetime.strptime(ts, ISO_FORMAT) # must not raise
class ArchiverTestCaseBase(BaseTestCase):
EXE: str = None # python source based
FORK_DEFAULT = False
@ -1682,7 +1687,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
repository = info_repo["repository"]
assert len(repository["id"]) == 64
assert "last_modified" in repository
assert datetime.strptime(repository["last_modified"], ISO_FORMAT) # must not raise
checkts(repository["last_modified"])
assert info_repo["encryption"]["mode"] == RK_ENCRYPTION[13:]
assert "keyfile" not in info_repo["encryption"]
cache = info_repo["cache"]
@ -1701,8 +1706,8 @@ class ArchiverTestCase(ArchiverTestCaseBase):
assert isinstance(archive["duration"], float)
assert len(archive["id"]) == 64
assert "stats" in archive
assert datetime.strptime(archive["start"], ISO_FORMAT)
assert datetime.strptime(archive["end"], ISO_FORMAT)
checkts(archive["start"])
checkts(archive["end"])
def test_info_json_of_empty_archive(self):
"""See https://github.com/borgbackup/borg/issues/6120"""
@ -2579,11 +2584,11 @@ class ArchiverTestCase(ArchiverTestCaseBase):
list_repo = json.loads(self.cmd(f"--repo={self.repository_location}", "rlist", "--json"))
repository = list_repo["repository"]
assert len(repository["id"]) == 64
assert datetime.strptime(repository["last_modified"], ISO_FORMAT) # must not raise
checkts(repository["last_modified"])
assert list_repo["encryption"]["mode"] == RK_ENCRYPTION[13:]
assert "keyfile" not in list_repo["encryption"]
archive0 = list_repo["archives"][0]
assert datetime.strptime(archive0["time"], ISO_FORMAT) # must not raise
checkts(archive0["time"])
list_archive = self.cmd(f"--repo={self.repository_location}", "list", "test", "--json-lines")
items = [json.loads(s) for s in list_archive.splitlines()]
@ -2591,7 +2596,7 @@ class ArchiverTestCase(ArchiverTestCaseBase):
file1 = items[1]
assert file1["path"] == "input/file1"
assert file1["size"] == 81920
assert datetime.strptime(file1["mtime"], ISO_FORMAT) # must not raise
checkts(file1["mtime"])
list_archive = self.cmd(
f"--repo={self.repository_location}", "list", "test", "--json-lines", "--format={sha256}"
@ -4058,7 +4063,7 @@ class ManifestAuthenticationTest(ArchiverTestCaseBase):
"version": 1,
"archives": {},
"config": {},
"timestamp": (datetime.utcnow() + timedelta(days=1)).strftime(ISO_FORMAT),
"timestamp": (datetime.utcnow() + timedelta(days=1)).isoformat(timespec="microseconds"),
}
),
),
@ -4078,7 +4083,7 @@ class ManifestAuthenticationTest(ArchiverTestCaseBase):
{
"version": 1,
"archives": {},
"timestamp": (datetime.utcnow() + timedelta(days=1)).strftime(ISO_FORMAT),
"timestamp": (datetime.utcnow() + timedelta(days=1)).isoformat(timespec="microseconds"),
}
),
),