github/plexpy
1
0
Fork 0
mirror of https://github.com/Tautulli/Tautulli.git synced 2025-07-07 13:41:15 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Update tempora-4.1.2

Browse source
This commit is contained in:
JonnyWong16 2021-10-14 21:13:30 -07:00
parent edd2f21ce1
commit a94edb4644
No known key found for this signature in database
GPG key ID: B1F1F9807184697A
6 changed files with 982 additions and 695 deletions
Download patch file Download diff file Expand all files Collapse all files

974
lib/tempora/__init__.py
View file

File diff suppressed because it is too large Load diff

51
lib/tempora/schedule.py
View file

@ -1,10 +1,17 @@
# -*- coding: utf-8 -*-
"""
Classes for calling functions a schedule.
""" """
Classes for calling functions a schedule. Has time zone support.
from __future__ import absolute_import For example, to run a job at 08:00 every morning in 'Asia/Calcutta':
>>> job = lambda: print("time is now", datetime.datetime())
>>> time = datetime.time(8, tzinfo=pytz.timezone('Asia/Calcutta'))
>>> cmd = PeriodicCommandFixedDelay.daily_at(time, job)
>>> sched = InvokeScheduler()
>>> sched.add(cmd)
>>> while True: # doctest: +SKIP
... sched.run_pending()
... time.sleep(.1)
"""
import datetime import datetime
import numbers import numbers
@ -13,8 +20,6 @@ import bisect
import pytz import pytz
__metaclass__ = type
def now(): def now():
""" """
@ -44,8 +49,13 @@ class DelayedCommand(datetime.datetime):
@classmethod @classmethod
def from_datetime(cls, other): def from_datetime(cls, other):
return cls( return cls(
other.year, other.month, other.day, other.hour, other.year,
other.minute, other.second, other.microsecond, other.month,
other.day,
other.hour,
other.minute,
other.second,
other.microsecond,
other.tzinfo, other.tzinfo,
) )
@ -91,6 +101,7 @@ class PeriodicCommand(DelayedCommand):
Like a delayed command, but expect this command to run every delay Like a delayed command, but expect this command to run every delay
seconds. seconds.
""" """
def _next_time(self): def _next_time(self):
""" """
Add delay to self, localized Add delay to self, localized
@ -117,8 +128,7 @@ class PeriodicCommand(DelayedCommand):
def __setattr__(self, key, value): def __setattr__(self, key, value):
if key == 'delay' and not value > datetime.timedelta(): if key == 'delay' and not value > datetime.timedelta():
raise ValueError( raise ValueError(
"A PeriodicCommand must have a positive, " "A PeriodicCommand must have a positive, " "non-zero delay."
"non-zero delay."
) )
super(PeriodicCommand, self).__setattr__(key, value) super(PeriodicCommand, self).__setattr__(key, value)
@ -132,6 +142,11 @@ class PeriodicCommandFixedDelay(PeriodicCommand):
@classmethod @classmethod
def at_time(cls, at, delay, target): def at_time(cls, at, delay, target):
"""
>>> cmd = PeriodicCommandFixedDelay.at_time(0, 30, None)
>>> cmd.delay.total_seconds()
30.0
"""
at = cls._from_timestamp(at) at = cls._from_timestamp(at)
cmd = cls.from_datetime(at) cmd = cls.from_datetime(at)
if isinstance(delay, numbers.Number): if isinstance(delay, numbers.Number):
@ -144,11 +159,18 @@ class PeriodicCommandFixedDelay(PeriodicCommand):
def daily_at(cls, at, target): def daily_at(cls, at, target):
""" """
Schedule a command to run at a specific time each day. Schedule a command to run at a specific time each day.
>>> from tempora import utc
>>> noon = utc.time(12, 0)
>>> cmd = PeriodicCommandFixedDelay.daily_at(noon, None)
>>> cmd.delay.total_seconds()
86400.0
""" """
daily = datetime.timedelta(days=1) daily = datetime.timedelta(days=1)
# convert when to the next datetime matching this time # convert when to the next datetime matching this time
when = datetime.datetime.combine(datetime.date.today(), at) when = datetime.datetime.combine(datetime.date.today(), at)
if when < now(): when -= daily
while when < now():
when += daily when += daily
return cls.at_time(cls._localize(when), daily, target) return cls.at_time(cls._localize(when), daily, target)
@ -158,6 +180,7 @@ class Scheduler:
A rudimentary abstract scheduler accepting DelayedCommands A rudimentary abstract scheduler accepting DelayedCommands
and dispatching them on schedule. and dispatching them on schedule.
""" """
def __init__(self): def __init__(self):
self.queue = [] self.queue = []
@ -186,6 +209,7 @@ class InvokeScheduler(Scheduler):
""" """
Command targets are functions to be invoked on schedule. Command targets are functions to be invoked on schedule.
""" """
def run(self, command): def run(self, command):
command.target() command.target()
@ -194,8 +218,9 @@ class CallbackScheduler(Scheduler):
""" """
Command targets are passed to a dispatch callable on schedule. Command targets are passed to a dispatch callable on schedule.
""" """
def __init__(self, dispatch): def __init__(self, dispatch):
super(CallbackScheduler, self).__init__() super().__init__()
self.dispatch = dispatch self.dispatch = dispatch
def run(self, command): def run(self, command):

203
lib/tempora/tests/test_schedule.py
View file

@ -1,6 +1,7 @@
import time import time
import random import random
import datetime import datetime
from unittest import mock
import pytest import pytest
import pytz import pytz
@ -8,111 +9,141 @@ import freezegun
from tempora import schedule from tempora import schedule
__metaclass__ = type
@pytest.fixture
def naive_times(monkeypatch):
monkeypatch.setattr(
'irc.schedule.from_timestamp',
datetime.datetime.fromtimestamp)
monkeypatch.setattr('irc.schedule.now', datetime.datetime.now)
do_nothing = type(None) do_nothing = type(None)
try:
do_nothing()
except TypeError:
# Python 2 compat
def do_nothing():
return None
def test_delayed_command_order(): def test_delayed_command_order():
""" """
delayed commands should be sorted by delay time delayed commands should be sorted by delay time
""" """
delays = [random.randint(0, 99) for x in range(5)] delays = [random.randint(0, 99) for x in range(5)]
cmds = sorted([ cmds = sorted(
schedule.DelayedCommand.after(delay, do_nothing) [schedule.DelayedCommand.after(delay, do_nothing) for delay in delays]
for delay in delays )
]) assert [c.delay.seconds for c in cmds] == sorted(delays)
assert [c.delay.seconds for c in cmds] == sorted(delays)
def test_periodic_command_delay(): def test_periodic_command_delay():
"A PeriodicCommand must have a positive, non-zero delay." "A PeriodicCommand must have a positive, non-zero delay."
with pytest.raises(ValueError) as exc_info: with pytest.raises(ValueError) as exc_info:
schedule.PeriodicCommand.after(0, None) schedule.PeriodicCommand.after(0, None)
assert str(exc_info.value) == test_periodic_command_delay.__doc__ assert str(exc_info.value) == test_periodic_command_delay.__doc__
def test_periodic_command_fixed_delay(): def test_periodic_command_fixed_delay():
""" """
Test that we can construct a periodic command with a fixed initial Test that we can construct a periodic command with a fixed initial
delay. delay.
""" """
fd = schedule.PeriodicCommandFixedDelay.at_time( fd = schedule.PeriodicCommandFixedDelay.at_time(
at=schedule.now(), at=schedule.now(), delay=datetime.timedelta(seconds=2), target=lambda: None
delay=datetime.timedelta(seconds=2), )
target=lambda: None, assert fd.due() is True
) assert fd.next().due() is False
assert fd.due() is True
assert fd.next().due() is False
class TestCommands: class TestCommands:
def test_delayed_command_from_timestamp(self): def test_delayed_command_from_timestamp(self):
""" """
Ensure a delayed command can be constructed from a timestamp. Ensure a delayed command can be constructed from a timestamp.
""" """
t = time.time() t = time.time()
schedule.DelayedCommand.at_time(t, do_nothing) schedule.DelayedCommand.at_time(t, do_nothing)
def test_command_at_noon(self): def test_command_at_noon(self):
""" """
Create a periodic command that's run at noon every day. Create a periodic command that's run at noon every day.
""" """
when = datetime.time(12, 0, tzinfo=pytz.utc) when = datetime.time(12, 0, tzinfo=pytz.utc)
cmd = schedule.PeriodicCommandFixedDelay.daily_at(when, target=None) cmd = schedule.PeriodicCommandFixedDelay.daily_at(when, target=None)
assert cmd.due() is False assert cmd.due() is False
next_cmd = cmd.next() next_cmd = cmd.next()
daily = datetime.timedelta(days=1) daily = datetime.timedelta(days=1)
day_from_now = schedule.now() + daily day_from_now = schedule.now() + daily
two_days_from_now = day_from_now + daily two_days_from_now = day_from_now + daily
assert day_from_now < next_cmd < two_days_from_now assert day_from_now < next_cmd < two_days_from_now
@pytest.mark.parametrize("hour", range(10, 14))
@pytest.mark.parametrize("tz_offset", (14, -14))
def test_command_at_noon_distant_local(self, hour, tz_offset):
"""
Run test_command_at_noon, but with the local timezone
more than 12 hours away from UTC.
"""
with freezegun.freeze_time(f"2020-01-10 {hour:02}:01", tz_offset=tz_offset):
self.test_command_at_noon()
class TestTimezones: class TestTimezones:
def test_alternate_timezone_west(self): def test_alternate_timezone_west(self):
target_tz = pytz.timezone('US/Pacific') target_tz = pytz.timezone('US/Pacific')
target = schedule.now().astimezone(target_tz) target = schedule.now().astimezone(target_tz)
cmd = schedule.DelayedCommand.at_time(target, target=None) cmd = schedule.DelayedCommand.at_time(target, target=None)
assert cmd.due() assert cmd.due()
def test_alternate_timezone_east(self): def test_alternate_timezone_east(self):
target_tz = pytz.timezone('Europe/Amsterdam') target_tz = pytz.timezone('Europe/Amsterdam')
target = schedule.now().astimezone(target_tz) target = schedule.now().astimezone(target_tz)
cmd = schedule.DelayedCommand.at_time(target, target=None) cmd = schedule.DelayedCommand.at_time(target, target=None)
assert cmd.due() assert cmd.due()
def test_daylight_savings(self): def test_daylight_savings(self):
""" """
A command at 9am should always be 9am regardless of A command at 9am should always be 9am regardless of
a DST boundary. a DST boundary.
""" """
with freezegun.freeze_time('2018-03-10 08:00:00'): with freezegun.freeze_time('2018-03-10 08:00:00'):
target_tz = pytz.timezone('US/Eastern') target_tz = pytz.timezone('US/Eastern')
target_time = datetime.time(9, tzinfo=target_tz) target_time = datetime.time(9, tzinfo=target_tz)
cmd = schedule.PeriodicCommandFixedDelay.daily_at( cmd = schedule.PeriodicCommandFixedDelay.daily_at(
target_time, target_time, target=lambda: None
target=lambda: None, )
)
def naive(dt): def naive(dt):
return dt.replace(tzinfo=None) return dt.replace(tzinfo=None)
assert naive(cmd) == datetime.datetime(2018, 3, 10, 9, 0, 0) assert naive(cmd) == datetime.datetime(2018, 3, 10, 9, 0, 0)
next_ = cmd.next() next_ = cmd.next()
assert naive(next_) == datetime.datetime(2018, 3, 11, 9, 0, 0) assert naive(next_) == datetime.datetime(2018, 3, 11, 9, 0, 0)
assert next_ - cmd == datetime.timedelta(hours=23) assert next_ - cmd == datetime.timedelta(hours=23)
class TestScheduler:
def test_invoke_scheduler(self):
sched = schedule.InvokeScheduler()
target = mock.MagicMock()
cmd = schedule.DelayedCommand.after(0, target)
sched.add(cmd)
sched.run_pending()
target.assert_called_once()
assert not sched.queue
def test_callback_scheduler(self):
callback = mock.MagicMock()
sched = schedule.CallbackScheduler(callback)
target = mock.MagicMock()
cmd = schedule.DelayedCommand.after(0, target)
sched.add(cmd)
sched.run_pending()
callback.assert_called_once_with(target)
def test_periodic_command(self):
sched = schedule.InvokeScheduler()
target = mock.MagicMock()
before = datetime.datetime.utcnow()
cmd = schedule.PeriodicCommand.after(10, target)
sched.add(cmd)
sched.run_pending()
target.assert_not_called()
with freezegun.freeze_time(before + datetime.timedelta(seconds=15)):
sched.run_pending()
assert sched.queue
target.assert_called_once()
with freezegun.freeze_time(before + datetime.timedelta(seconds=25)):
sched.run_pending()
assert target.call_count == 2

50
lib/tempora/tests/test_timing.py Normal file
View file

@ -0,0 +1,50 @@
import datetime
import time
import contextlib
import os
from unittest import mock
import pytest
from tempora import timing
def test_IntervalGovernor():
"""
IntervalGovernor should prevent a function from being called more than
once per interval.
"""
func_under_test = mock.MagicMock()
# to look like a function, it needs a __name__ attribute
func_under_test.__name__ = 'func_under_test'
interval = datetime.timedelta(seconds=1)
governed = timing.IntervalGovernor(interval)(func_under_test)
governed('a')
governed('b')
governed(3, 'sir')
func_under_test.assert_called_once_with('a')
@pytest.fixture
def alt_tz(monkeypatch):
hasattr(time, 'tzset') or pytest.skip("tzset not available")
@contextlib.contextmanager
def change():
val = 'AEST-10AEDT-11,M10.5.0,M3.5.0'
with monkeypatch.context() as ctx:
ctx.setitem(os.environ, 'TZ', val)
time.tzset()
yield
time.tzset()
return change()
def test_Stopwatch_timezone_change(alt_tz):
"""
The stopwatch should provide a consistent duration even
if the timezone changes.
"""
watch = timing.Stopwatch()
with alt_tz:
assert abs(watch.split().total_seconds()) < 0.1

394
lib/tempora/timing.py
View file

@ -1,258 +1,266 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals, absolute_import
import datetime import datetime
import functools import functools
import numbers import numbers
import time import time
import collections.abc
import six import contextlib
import jaraco.functools import jaraco.functools
__metaclass__ = type
class Stopwatch: class Stopwatch:
""" """
A simple stopwatch which starts automatically. A simple stopwatch which starts automatically.
>>> w = Stopwatch() >>> w = Stopwatch()
>>> _1_sec = datetime.timedelta(seconds=1) >>> _1_sec = datetime.timedelta(seconds=1)
>>> w.split() < _1_sec >>> w.split() < _1_sec
True True
>>> import time >>> import time
>>> time.sleep(1.0) >>> time.sleep(1.0)
>>> w.split() >= _1_sec >>> w.split() >= _1_sec
True True
>>> w.stop() >= _1_sec >>> w.stop() >= _1_sec
True True
>>> w.reset() >>> w.reset()
>>> w.start() >>> w.start()
>>> w.split() < _1_sec >>> w.split() < _1_sec
True True
It should be possible to launch the Stopwatch in a context: It should be possible to launch the Stopwatch in a context:
>>> with Stopwatch() as watch: >>> with Stopwatch() as watch:
... assert isinstance(watch.split(), datetime.timedelta) ... assert isinstance(watch.split(), datetime.timedelta)
In that case, the watch is stopped when the context is exited, In that case, the watch is stopped when the context is exited,
so to read the elapsed time:: so to read the elapsed time:
>>> watch.elapsed >>> watch.elapsed
datetime.timedelta(...) datetime.timedelta(...)
>>> watch.elapsed.seconds >>> watch.elapsed.seconds
0 0
""" """
def __init__(self):
self.reset()
self.start()
def reset(self): def __init__(self):
self.elapsed = datetime.timedelta(0) self.reset()
if hasattr(self, 'start_time'): self.start()
del self.start_time
def start(self): def reset(self):
self.start_time = datetime.datetime.utcnow() self.elapsed = datetime.timedelta(0)
with contextlib.suppress(AttributeError):
del self.start_time
def stop(self): def start(self):
stop_time = datetime.datetime.utcnow() self.start_time = datetime.datetime.utcnow()
self.elapsed += stop_time - self.start_time
del self.start_time
return self.elapsed
def split(self): def stop(self):
local_duration = datetime.datetime.utcnow() - self.start_time stop_time = datetime.datetime.utcnow()
return self.elapsed + local_duration self.elapsed += stop_time - self.start_time
del self.start_time
return self.elapsed
# context manager support def split(self):
def __enter__(self): local_duration = datetime.datetime.utcnow() - self.start_time
self.start() return self.elapsed + local_duration
return self
def __exit__(self, exc_type, exc_value, traceback): # context manager support
self.stop() def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.stop()
class IntervalGovernor: class IntervalGovernor:
""" """
Decorate a function to only allow it to be called once per Decorate a function to only allow it to be called once per
min_interval. Otherwise, it returns None. min_interval. Otherwise, it returns None.
"""
def __init__(self, min_interval):
if isinstance(min_interval, numbers.Number):
min_interval = datetime.timedelta(seconds=min_interval)
self.min_interval = min_interval
self.last_call = None
def decorate(self, func): >>> gov = IntervalGovernor(30)
@functools.wraps(func) >>> gov.min_interval.total_seconds()
def wrapper(*args, **kwargs): 30.0
allow = ( """
not self.last_call
or self.last_call.split() > self.min_interval
)
if allow:
self.last_call = Stopwatch()
return func(*args, **kwargs)
return wrapper
__call__ = decorate def __init__(self, min_interval):
if isinstance(min_interval, numbers.Number):
min_interval = datetime.timedelta(seconds=min_interval)
self.min_interval = min_interval
self.last_call = None
def decorate(self, func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
allow = not self.last_call or self.last_call.split() > self.min_interval
if allow:
self.last_call = Stopwatch()
return func(*args, **kwargs)
return wrapper
__call__ = decorate
class Timer(Stopwatch): class Timer(Stopwatch):
""" """
Watch for a target elapsed time. Watch for a target elapsed time.
>>> t = Timer(0.1) >>> t = Timer(0.1)
>>> t.expired() >>> t.expired()
False False
>>> __import__('time').sleep(0.15) >>> __import__('time').sleep(0.15)
>>> t.expired() >>> t.expired()
True True
""" """
def __init__(self, target=float('Inf')):
self.target = self._accept(target)
super(Timer, self).__init__()
def _accept(self, target): def __init__(self, target=float('Inf')):
"Accept None or ∞ or datetime or numeric for target" self.target = self._accept(target)
if isinstance(target, datetime.timedelta): super(Timer, self).__init__()
target = target.total_seconds()
if target is None: @staticmethod
# treat None as infinite target def _accept(target):
target = float('Inf') """
Accept None or ∞ or datetime or numeric for target
return target >>> Timer._accept(datetime.timedelta(seconds=30))
30.0
>>> Timer._accept(None)
inf
"""
if isinstance(target, datetime.timedelta):
target = target.total_seconds()
def expired(self): if target is None:
return self.split().total_seconds() > self.target # treat None as infinite target
target = float('Inf')
return target
def expired(self):
return self.split().total_seconds() > self.target
class BackoffDelay(six.Iterator): class BackoffDelay(collections.abc.Iterator):
""" """
Exponential backoff delay. Exponential backoff delay.
Useful for defining delays between retries. Consider for use Useful for defining delays between retries. Consider for use
with ``jaraco.functools.retry_call`` as the cleanup. with ``jaraco.functools.retry_call`` as the cleanup.
Default behavior has no effect; a delay or jitter must Default behavior has no effect; a delay or jitter must
be supplied for the call to be non-degenerate. be supplied for the call to be non-degenerate.
>>> bd = BackoffDelay() >>> bd = BackoffDelay()
>>> bd() >>> bd()
>>> bd() >>> bd()
The following instance will delay 10ms for the first call, The following instance will delay 10ms for the first call,
20ms for the second, etc. 20ms for the second, etc.
>>> bd = BackoffDelay(delay=0.01, factor=2) >>> bd = BackoffDelay(delay=0.01, factor=2)
>>> bd() >>> bd()
>>> bd() >>> bd()
Inspect and adjust the state of the delay anytime. Inspect and adjust the state of the delay anytime.
>>> bd.delay >>> bd.delay
0.04 0.04
>>> bd.delay = 0.01 >>> bd.delay = 0.01
Set limit to prevent the delay from exceeding bounds. Set limit to prevent the delay from exceeding bounds.
>>> bd = BackoffDelay(delay=0.01, factor=2, limit=0.015) >>> bd = BackoffDelay(delay=0.01, factor=2, limit=0.015)
>>> bd() >>> bd()
>>> bd.delay >>> bd.delay
0.015 0.015
To reset the backoff, simply call ``.reset()``: To reset the backoff, simply call ``.reset()``:
>>> bd.reset() >>> bd.reset()
>>> bd.delay >>> bd.delay
0.01 0.01
Iterate on the object to retrieve/advance the delay values. Iterate on the object to retrieve/advance the delay values.
>>> next(bd) >>> next(bd)
0.01 0.01
>>> next(bd) >>> next(bd)
0.015 0.015
>>> import itertools >>> import itertools
>>> tuple(itertools.islice(bd, 3)) >>> tuple(itertools.islice(bd, 3))
(0.015, 0.015, 0.015) (0.015, 0.015, 0.015)
Limit may be a callable taking a number and returning Limit may be a callable taking a number and returning
the limited number. the limited number.
>>> at_least_one = lambda n: max(n, 1) >>> at_least_one = lambda n: max(n, 1)
>>> bd = BackoffDelay(delay=0.01, factor=2, limit=at_least_one) >>> bd = BackoffDelay(delay=0.01, factor=2, limit=at_least_one)
>>> next(bd) >>> next(bd)
0.01 0.01
>>> next(bd) >>> next(bd)
1 1
Pass a jitter to add or subtract seconds to the delay. Pass a jitter to add or subtract seconds to the delay.
>>> bd = BackoffDelay(jitter=0.01) >>> bd = BackoffDelay(jitter=0.01)
>>> next(bd) >>> next(bd)
0 0
>>> next(bd) >>> next(bd)
0.01 0.01
Jitter may be a callable. To supply a non-deterministic jitter Jitter may be a callable. To supply a non-deterministic jitter
between -0.5 and 0.5, consider: between -0.5 and 0.5, consider:
>>> import random >>> import random
>>> jitter=functools.partial(random.uniform, -0.5, 0.5) >>> jitter=functools.partial(random.uniform, -0.5, 0.5)
>>> bd = BackoffDelay(jitter=jitter) >>> bd = BackoffDelay(jitter=jitter)
>>> next(bd) >>> next(bd)
0 0
>>> 0 <= next(bd) <= 0.5 >>> 0 <= next(bd) <= 0.5
True True
""" """
delay = 0 delay = 0
factor = 1 factor = 1
"Multiplier applied to delay" "Multiplier applied to delay"
jitter = 0 jitter = 0
"Number or callable returning extra seconds to add to delay" "Number or callable returning extra seconds to add to delay"
@jaraco.functools.save_method_args @jaraco.functools.save_method_args
def __init__(self, delay=0, factor=1, limit=float('inf'), jitter=0): def __init__(self, delay=0, factor=1, limit=float('inf'), jitter=0):
self.delay = delay self.delay = delay
self.factor = factor self.factor = factor
if isinstance(limit, numbers.Number): if isinstance(limit, numbers.Number):
limit_ = limit limit_ = limit
def limit(n): def limit(n):
return max(0, min(limit_, n)) return max(0, min(limit_, n))
self.limit = limit
if isinstance(jitter, numbers.Number):
jitter_ = jitter
def jitter(): self.limit = limit
return jitter_ if isinstance(jitter, numbers.Number):
self.jitter = jitter jitter_ = jitter
def __call__(self): def jitter():
time.sleep(next(self)) return jitter_
def __next__(self): self.jitter = jitter
delay = self.delay
self.bump()
return delay
def __iter__(self): def __call__(self):
return self time.sleep(next(self))
def bump(self): def __next__(self):
self.delay = self.limit(self.delay * self.factor + self.jitter()) delay = self.delay
self.bump()
return delay
def reset(self): def __iter__(self):
saved = self._saved___init__ return self
self.__init__(*saved.args, **saved.kwargs)
def bump(self):
self.delay = self.limit(self.delay * self.factor + self.jitter())
def reset(self):
saved = self._saved___init__
self.__init__(*saved.args, **saved.kwargs)

5
lib/tempora/utc.py
View file

@ -31,9 +31,6 @@ __all__ = ['now', 'fromtimestamp', 'datetime', 'time']
now = functools.partial(std.datetime.now, std.timezone.utc) now = functools.partial(std.datetime.now, std.timezone.utc)
fromtimestamp = functools.partial( fromtimestamp = functools.partial(std.datetime.fromtimestamp, tz=std.timezone.utc)
std.datetime.fromtimestamp,
tz=std.timezone.utc,
)
datetime = functools.partial(std.datetime, tzinfo=std.timezone.utc) datetime = functools.partial(std.datetime, tzinfo=std.timezone.utc)
time = functools.partial(std.time, tzinfo=std.timezone.utc) time = functools.partial(std.time, tzinfo=std.timezone.utc)