Techdebt: Replace deprecated utcfromtimestamp-methods (#7146)
This commit is contained in:
parent
be0dffcd38
commit
d77acd4456
@ -35,6 +35,7 @@ from moto.core.utils import (
|
|||||||
gzip_decompress,
|
gzip_decompress,
|
||||||
method_names_from_class,
|
method_names_from_class,
|
||||||
params_sort_function,
|
params_sort_function,
|
||||||
|
utcfromtimestamp,
|
||||||
)
|
)
|
||||||
from moto.utilities.utils import load_resource, load_resource_as_bytes
|
from moto.utilities.utils import load_resource, load_resource_as_bytes
|
||||||
|
|
||||||
@ -1076,11 +1077,7 @@ def to_str(value: Any, spec: Dict[str, Any]) -> str:
|
|||||||
elif vtype == "double":
|
elif vtype == "double":
|
||||||
return str(value)
|
return str(value)
|
||||||
elif vtype == "timestamp":
|
elif vtype == "timestamp":
|
||||||
return (
|
return utcfromtimestamp(value).replace(tzinfo=datetime.timezone.utc).isoformat()
|
||||||
datetime.datetime.utcfromtimestamp(value)
|
|
||||||
.replace(tzinfo=datetime.timezone.utc)
|
|
||||||
.isoformat()
|
|
||||||
)
|
|
||||||
elif vtype == "string":
|
elif vtype == "string":
|
||||||
return str(value)
|
return str(value)
|
||||||
elif value is None:
|
elif value is None:
|
||||||
|
@ -182,7 +182,7 @@ def str_to_rfc_1123_datetime(value: str) -> datetime.datetime:
|
|||||||
|
|
||||||
def unix_time(dt: Optional[datetime.datetime] = None) -> float:
|
def unix_time(dt: Optional[datetime.datetime] = None) -> float:
|
||||||
dt = dt or utcnow()
|
dt = dt or utcnow()
|
||||||
epoch = datetime.datetime.utcfromtimestamp(0)
|
epoch = utcfromtimestamp(0)
|
||||||
delta = dt - epoch
|
delta = dt - epoch
|
||||||
return (delta.days * 86400) + (delta.seconds + (delta.microseconds / 1e6))
|
return (delta.days * 86400) + (delta.seconds + (delta.microseconds / 1e6))
|
||||||
|
|
||||||
@ -191,6 +191,24 @@ def unix_time_millis(dt: Optional[datetime.datetime] = None) -> float:
|
|||||||
return unix_time(dt) * 1000.0
|
return unix_time(dt) * 1000.0
|
||||||
|
|
||||||
|
|
||||||
|
def utcfromtimestamp(value: int) -> datetime.datetime:
|
||||||
|
"""
|
||||||
|
Return the UTC datetime corresponding to the POSIX timestamp, with tzinfo None. The resulting object is naive.
|
||||||
|
"""
|
||||||
|
# Python 3.12 starts throwing deprecation warnings for utcfromtimestamp()
|
||||||
|
# The docs recommend to use fromtimestamp(UTC) instead
|
||||||
|
#
|
||||||
|
# fromtimestamp(UTC) creates an aware datetime - but utcfromtimestamp() creates a naive datetime
|
||||||
|
# That's why we have to `replace(tzinfo=None)` to make now(UTC) naive.
|
||||||
|
if PYTHON_311:
|
||||||
|
# Only available from 3.11
|
||||||
|
from datetime import UTC # type: ignore
|
||||||
|
|
||||||
|
return datetime.datetime.fromtimestamp(value, tz=UTC).replace(tzinfo=None)
|
||||||
|
else:
|
||||||
|
return datetime.datetime.utcfromtimestamp(value)
|
||||||
|
|
||||||
|
|
||||||
def utcnow() -> datetime.datetime:
|
def utcnow() -> datetime.datetime:
|
||||||
# Python 3.12 starts throwing deprecation warnings for utcnow()
|
# Python 3.12 starts throwing deprecation warnings for utcnow()
|
||||||
# The docs recommend to use now(UTC) instead
|
# The docs recommend to use now(UTC) instead
|
||||||
@ -198,7 +216,7 @@ def utcnow() -> datetime.datetime:
|
|||||||
# now(UTC) creates an aware datetime - but utcnow() creates a naive datetime
|
# now(UTC) creates an aware datetime - but utcnow() creates a naive datetime
|
||||||
# That's why we have to `replace(tzinfo=None)` to make now(UTC) naive.
|
# That's why we have to `replace(tzinfo=None)` to make now(UTC) naive.
|
||||||
if PYTHON_311:
|
if PYTHON_311:
|
||||||
# Only available in 3.11
|
# Only available from 3.11
|
||||||
from datetime import UTC # type: ignore
|
from datetime import UTC # type: ignore
|
||||||
|
|
||||||
return datetime.datetime.now(UTC).replace(tzinfo=None)
|
return datetime.datetime.now(UTC).replace(tzinfo=None)
|
||||||
|
@ -5,7 +5,6 @@ import re
|
|||||||
import sys
|
import sys
|
||||||
import warnings
|
import warnings
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
from datetime import datetime
|
|
||||||
from enum import Enum, unique
|
from enum import Enum, unique
|
||||||
from json import JSONDecodeError
|
from json import JSONDecodeError
|
||||||
from operator import eq, ge, gt, le, lt
|
from operator import eq, ge, gt, le, lt
|
||||||
@ -20,6 +19,7 @@ from moto.core.utils import (
|
|||||||
iso_8601_datetime_without_milliseconds,
|
iso_8601_datetime_without_milliseconds,
|
||||||
unix_time,
|
unix_time,
|
||||||
unix_time_millis,
|
unix_time_millis,
|
||||||
|
utcfromtimestamp,
|
||||||
)
|
)
|
||||||
from moto.events.exceptions import (
|
from moto.events.exceptions import (
|
||||||
IllegalStatusException,
|
IllegalStatusException,
|
||||||
@ -170,7 +170,7 @@ class Rule(CloudFormationModel):
|
|||||||
|
|
||||||
event_copy = copy.deepcopy(event)
|
event_copy = copy.deepcopy(event)
|
||||||
event_copy["time"] = iso_8601_datetime_without_milliseconds(
|
event_copy["time"] = iso_8601_datetime_without_milliseconds(
|
||||||
datetime.utcfromtimestamp(event_copy["time"])
|
utcfromtimestamp(event_copy["time"])
|
||||||
)
|
)
|
||||||
|
|
||||||
log_stream_name = str(random.uuid4())
|
log_stream_name = str(random.uuid4())
|
||||||
@ -202,7 +202,7 @@ class Rule(CloudFormationModel):
|
|||||||
|
|
||||||
event_copy = copy.deepcopy(event)
|
event_copy = copy.deepcopy(event)
|
||||||
event_copy["time"] = iso_8601_datetime_without_milliseconds(
|
event_copy["time"] = iso_8601_datetime_without_milliseconds(
|
||||||
datetime.utcfromtimestamp(event_copy["time"])
|
utcfromtimestamp(event_copy["time"])
|
||||||
)
|
)
|
||||||
|
|
||||||
if group_id:
|
if group_id:
|
||||||
|
@ -4,7 +4,7 @@ import time
|
|||||||
from typing import Any, Dict, List, Optional, Tuple
|
from typing import Any, Dict, List, Optional, Tuple
|
||||||
|
|
||||||
from moto.core import BackendDict, BaseBackend, BaseModel
|
from moto.core import BackendDict, BaseBackend, BaseModel
|
||||||
from moto.core.utils import utcnow
|
from moto.core.utils import utcfromtimestamp, utcnow
|
||||||
from moto.moto_api._internal import mock_random
|
from moto.moto_api._internal import mock_random
|
||||||
|
|
||||||
from .exceptions import (
|
from .exceptions import (
|
||||||
@ -273,7 +273,7 @@ class SecretsManagerBackend(BaseBackend):
|
|||||||
return identifier in self.secrets
|
return identifier in self.secrets
|
||||||
|
|
||||||
def _unix_time_secs(self, dt: datetime.datetime) -> float:
|
def _unix_time_secs(self, dt: datetime.datetime) -> float:
|
||||||
epoch = datetime.datetime.utcfromtimestamp(0)
|
epoch = utcfromtimestamp(0)
|
||||||
return (dt - epoch).total_seconds()
|
return (dt - epoch).total_seconds()
|
||||||
|
|
||||||
def _client_request_token_validator(self, client_request_token: str) -> None:
|
def _client_request_token_validator(self, client_request_token: str) -> None:
|
||||||
|
@ -12,6 +12,7 @@ from freezegun import freeze_time
|
|||||||
|
|
||||||
from moto import mock_cloudwatch, mock_s3
|
from moto import mock_cloudwatch, mock_s3
|
||||||
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
|
from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
|
||||||
|
from moto.core.utils import utcnow
|
||||||
|
|
||||||
|
|
||||||
@mock_cloudwatch
|
@mock_cloudwatch
|
||||||
@ -1912,7 +1913,7 @@ def test_get_metric_data_queries():
|
|||||||
verify that >= 10 queries can still be parsed
|
verify that >= 10 queries can still be parsed
|
||||||
there was an error with the order of parsing items, leading to IndexError
|
there was an error with the order of parsing items, leading to IndexError
|
||||||
"""
|
"""
|
||||||
now = datetime.utcnow().replace(microsecond=0)
|
now = utcnow().replace(microsecond=0)
|
||||||
start_time = now - timedelta(minutes=10)
|
start_time = now - timedelta(minutes=10)
|
||||||
end_time = now + timedelta(minutes=5)
|
end_time = now + timedelta(minutes=5)
|
||||||
original_query = {
|
original_query = {
|
||||||
|
Loading…
Reference in New Issue
Block a user