OpenSource/python-telegram-bot
OpenSource/python-telegram-bot
1
0
Fork 0
mirror of https://github.com/python-telegram-bot/python-telegram-bot.git synced 2025-03-17 04:39:55 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add tzinfo kwarg to from_timestamp() (#1621)

Browse source 
* Add tz kwarg to from_timestamp()

* Correct handling of tzinfo=None

* Small Improvements

* None-tz yields naive dto

* Remove legacey compatibility of UTC stuff

* Update telegram/utils/helpers.py

Co-authored-by: Noam Meltzer <tsnoam@gmail.com>
This commit is contained in:
Bibo-Joshi 2020-05-01 22:55:13 +02:00 • committed by GitHub
parent 8427346a0d
commit 7e231183c4
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 45 additions and 68 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
telegram/ext/jobqueue.py
View file

@ -29,7 +29,7 @@ from threading import Thread, Lock, Event
from telegram.ext.callbackcontext import CallbackContext
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.helpers import to_float_timestamp, _UTC
from telegram.utils.helpers import to_float_timestamp
class Days(object):
@ -436,7 +436,7 @@ class Job(object):
self._days = None
self.days = days
self.tzinfo = tzinfo or _UTC
self.tzinfo = tzinfo or datetime.timezone.utc
self._job_queue = weakref.proxy(job_queue) if job_queue is not None else None
@ -519,7 +519,7 @@ class Job(object):
def _set_next_t(self, next_t):
if isinstance(next_t, datetime.datetime):
# Set timezone to UTC in case datetime is in local timezone.
next_t = next_t.astimezone(_UTC)
next_t = next_t.astimezone(datetime.timezone.utc)
next_t = to_float_timestamp(next_t)
elif not (isinstance(next_t, Number) or next_t is None):
raise TypeError("The 'next_t' argument should be one of the following types: "

57
telegram/utils/helpers.py
View file

@ -75,46 +75,12 @@ def escape_markdown(text, version=1, entity_type=None):
# -------- date/time related helpers --------
# TODO: add generic specification of UTC for naive datetimes to docs
if hasattr(dtm, 'timezone'):
# Python 3.3+
def _datetime_to_float_timestamp(dt_obj):
if dt_obj.tzinfo is None:
dt_obj = dt_obj.replace(tzinfo=_UTC)
return dt_obj.timestamp()
_UtcOffsetTimezone = dtm.timezone
_UTC = dtm.timezone.utc
else:
# Python < 3.3 (incl 2.7)
# hardcoded timezone class (`datetime.timezone` isn't available in py2)
class _UtcOffsetTimezone(dtm.tzinfo):
def __init__(self, offset):
self.offset = offset
def tzname(self, dt):
return 'UTC +{}'.format(self.offset)
def utcoffset(self, dt):
return self.offset
def dst(self, dt):
return dtm.timedelta(0)
_UTC = _UtcOffsetTimezone(dtm.timedelta(0))
__EPOCH_DT = dtm.datetime.fromtimestamp(0, tz=_UTC)
__NAIVE_EPOCH_DT = __EPOCH_DT.replace(tzinfo=None)
# _datetime_to_float_timestamp
# Not using future.backports.datetime here as datetime value might be an input from the user,
# making every isinstace() call more delicate. So we just use our own compat layer.
def _datetime_to_float_timestamp(dt_obj):
epoch_dt = __EPOCH_DT if dt_obj.tzinfo is not None else __NAIVE_EPOCH_DT
return (dt_obj - epoch_dt).total_seconds()
_datetime_to_float_timestamp.__doc__ = \
def _datetime_to_float_timestamp(dt_obj):
"""Converts a datetime object to a float timestamp (with sub-second precision).
If the datetime object is timezone-naive, it is assumed to be in UTC."""
If the datetime object is timezone-naive, it is assumed to be in UTC."""
if dt_obj.tzinfo is None:
dt_obj = dt_obj.replace(tzinfo=dtm.timezone.utc)
return dt_obj.timestamp()
def to_float_timestamp(t, reference_timestamp=None):
@ -196,22 +162,27 @@ def to_timestamp(dt_obj, reference_timestamp=None):
return int(to_float_timestamp(dt_obj, reference_timestamp)) if dt_obj is not None else None
def from_timestamp(unixtime):
def from_timestamp(unixtime, tzinfo=dtm.timezone.utc):
"""
Converts an (integer) unix timestamp to a naive datetime object in UTC.
Converts an (integer) unix timestamp to a timezone aware datetime object.
``None`` s are left alone (i.e. ``from_timestamp(None)`` is ``None``).
Args:
unixtime (int): integer POSIX timestamp
tzinfo (:obj:`datetime.tzinfo`, optional): The timezone, the timestamp is to be converted
to. Defaults to UTC.
Returns:
equivalent :obj:`datetime.datetime` value in naive UTC if ``timestamp`` is not
timezone aware equivalent :obj:`datetime.datetime` value if ``timestamp`` is not
``None``; else ``None``
"""
if unixtime is None:
return None
return dtm.datetime.utcfromtimestamp(unixtime)
if tzinfo is not None:
return dtm.datetime.fromtimestamp(unixtime, tz=tzinfo)
else:
return dtm.datetime.utcfromtimestamp(unixtime)
# -------- end --------

3
tests/conftest.py
View file

@ -31,7 +31,6 @@ from telegram import (Bot, Message, User, Chat, MessageEntity, Update,
InlineQuery, CallbackQuery, ShippingQuery, PreCheckoutQuery,
ChosenInlineResult)
from telegram.ext import Dispatcher, JobQueue, Updater, BaseFilter, Defaults
from telegram.utils.helpers import _UtcOffsetTimezone
from telegram.error import BadRequest
from tests.bots import get_bot
@ -281,7 +280,7 @@ def utc_offset(request):
@pytest.fixture()
def timezone(utc_offset):
return _UtcOffsetTimezone(utc_offset)
return datetime.timezone(utc_offset)
def expect_bad_request(func, message, reason):

18
tests/test_helpers.py
View file

@ -27,14 +27,14 @@ from telegram import User
from telegram import MessageEntity
from telegram.message import Message
from telegram.utils import helpers
from telegram.utils.helpers import _UtcOffsetTimezone, _datetime_to_float_timestamp
from telegram.utils.helpers import _datetime_to_float_timestamp
# sample time specification values categorised into absolute / delta / time-of-day
ABSOLUTE_TIME_SPECS = [dtm.datetime.now(tz=_UtcOffsetTimezone(dtm.timedelta(hours=-7))),
ABSOLUTE_TIME_SPECS = [dtm.datetime.now(tz=dtm.timezone(dtm.timedelta(hours=-7))),
dtm.datetime.utcnow()]
DELTA_TIME_SPECS = [dtm.timedelta(hours=3, seconds=42, milliseconds=2), 30, 7.5]
TIME_OF_DAY_TIME_SPECS = [dtm.time(12, 42, tzinfo=_UtcOffsetTimezone(dtm.timedelta(hours=-7))),
TIME_OF_DAY_TIME_SPECS = [dtm.time(12, 42, tzinfo=dtm.timezone(dtm.timedelta(hours=-7))),
dtm.time(12, 42)]
RELATIVE_TIME_SPECS = DELTA_TIME_SPECS + TIME_OF_DAY_TIME_SPECS
TIME_SPECS = ABSOLUTE_TIME_SPECS + RELATIVE_TIME_SPECS
@ -142,8 +142,16 @@ class TestHelpers(object):
# this 'convenience' behaviour has been left left for backwards compatibility
assert helpers.to_timestamp(None) is None
def test_from_timestamp(self):
assert helpers.from_timestamp(1573431976) == dtm.datetime(2019, 11, 11, 0, 26, 16)
def test_from_timestamp_naive(self):
datetime = dtm.datetime(2019, 11, 11, 0, 26, 16, tzinfo=None)
assert helpers.from_timestamp(1573431976, tzinfo=None) == datetime
def test_from_timestamp_aware(self, timezone):
# we're parametrizing this with two different UTC offsets to exclude the possibility
# of an xpass when the test is run in a timezone with the same UTC offset
datetime = dtm.datetime(2019, 11, 11, 0, 26, 16, 10**5, tzinfo=timezone)
assert (helpers.from_timestamp(1573431976.1 - timezone.utcoffset(None).total_seconds())
== datetime)
def test_create_deep_linked_url(self):
username = 'JamesTheMock'

29
tests/test_jobqueue.py
View file

@ -29,7 +29,6 @@ from flaky import flaky
from telegram.ext import JobQueue, Updater, Job, CallbackContext
from telegram.utils.deprecate import TelegramDeprecationWarning
from telegram.utils.helpers import _UtcOffsetTimezone, _UTC
@pytest.fixture(scope='function')
@ -277,7 +276,7 @@ class TestJobQueue(object):
# must subtract one minute because the UTC offset has to be strictly less than 24h
# thus this test will xpass if run in the interval [00:00, 00:01) UTC time
# (because target time will be 23:59 UTC, so local and target weekday will be the same)
target_tzinfo = _UtcOffsetTimezone(dtm.timedelta(days=1, minutes=-1))
target_tzinfo = dtm.timezone(dtm.timedelta(days=1, minutes=-1))
target_datetime = (utcnow + dtm.timedelta(days=1, minutes=-1, seconds=delta)).replace(
tzinfo=target_tzinfo)
target_time = target_datetime.timetz()
@ -344,7 +343,7 @@ class TestJobQueue(object):
jobs = [job_1, job_2, job_3]
for job in jobs:
assert job.tzinfo == _UTC
assert job.tzinfo == dtm.timezone.utc
def test_job_next_t_property(self, job_queue):
# Testing:
@ -379,9 +378,9 @@ class TestJobQueue(object):
job = job_queue.run_once(self.job_run_once, 0.05)
t = dtm.datetime.now(tz=_UtcOffsetTimezone(dtm.timedelta(hours=12)))
t = dtm.datetime.now(tz=dtm.timezone(dtm.timedelta(hours=12)))
job._set_next_t(t)
job.tzinfo = _UtcOffsetTimezone(dtm.timedelta(hours=5))
job.tzinfo = dtm.timezone(dtm.timedelta(hours=5))
assert job.next_t == t.astimezone(job.tzinfo)
def test_passing_tzinfo_to_job(self, job_queue):
@ -389,21 +388,21 @@ class TestJobQueue(object):
and run_repeating methods"""
when_dt_tz_specific = dtm.datetime.now(
tz=_UtcOffsetTimezone(dtm.timedelta(hours=12))
tz=dtm.timezone(dtm.timedelta(hours=12))
) + dtm.timedelta(seconds=2)
when_dt_tz_utc = dtm.datetime.now() + dtm.timedelta(seconds=2)
job_once1 = job_queue.run_once(self.job_run_once, when_dt_tz_specific)
job_once2 = job_queue.run_once(self.job_run_once, when_dt_tz_utc)
when_time_tz_specific = (dtm.datetime.now(
tz=_UtcOffsetTimezone(dtm.timedelta(hours=12))
tz=dtm.timezone(dtm.timedelta(hours=12))
) + dtm.timedelta(seconds=2)).timetz()
when_time_tz_utc = (dtm.datetime.now() + dtm.timedelta(seconds=2)).timetz()
job_once3 = job_queue.run_once(self.job_run_once, when_time_tz_specific)
job_once4 = job_queue.run_once(self.job_run_once, when_time_tz_utc)
first_dt_tz_specific = dtm.datetime.now(
tz=_UtcOffsetTimezone(dtm.timedelta(hours=12))
tz=dtm.timezone(dtm.timedelta(hours=12))
) + dtm.timedelta(seconds=2)
first_dt_tz_utc = dtm.datetime.now() + dtm.timedelta(seconds=2)
job_repeating1 = job_queue.run_repeating(
@ -412,7 +411,7 @@ class TestJobQueue(object):
self.job_run_once, 2, first=first_dt_tz_utc)
first_time_tz_specific = (dtm.datetime.now(
tz=_UtcOffsetTimezone(dtm.timedelta(hours=12))
tz=dtm.timezone(dtm.timedelta(hours=12))
) + dtm.timedelta(seconds=2)).timetz()
first_time_tz_utc = (dtm.datetime.now() + dtm.timedelta(seconds=2)).timetz()
job_repeating3 = job_queue.run_repeating(
@ -421,19 +420,19 @@ class TestJobQueue(object):
self.job_run_once, 2, first=first_time_tz_utc)
time_tz_specific = (dtm.datetime.now(
tz=_UtcOffsetTimezone(dtm.timedelta(hours=12))
tz=dtm.timezone(dtm.timedelta(hours=12))
) + dtm.timedelta(seconds=2)).timetz()
time_tz_utc = (dtm.datetime.now() + dtm.timedelta(seconds=2)).timetz()
job_daily1 = job_queue.run_daily(self.job_run_once, time_tz_specific)
job_daily2 = job_queue.run_daily(self.job_run_once, time_tz_utc)
assert job_once1.tzinfo == when_dt_tz_specific.tzinfo
assert job_once2.tzinfo == _UTC
assert job_once2.tzinfo == dtm.timezone.utc
assert job_once3.tzinfo == when_time_tz_specific.tzinfo
assert job_once4.tzinfo == _UTC
assert job_once4.tzinfo == dtm.timezone.utc
assert job_repeating1.tzinfo == first_dt_tz_specific.tzinfo
assert job_repeating2.tzinfo == _UTC
assert job_repeating2.tzinfo == dtm.timezone.utc
assert job_repeating3.tzinfo == first_time_tz_specific.tzinfo
assert job_repeating4.tzinfo == _UTC
assert job_repeating4.tzinfo == dtm.timezone.utc
assert job_daily1.tzinfo == time_tz_specific.tzinfo
assert job_daily2.tzinfo == _UTC
assert job_daily2.tzinfo == dtm.timezone.utc