def refresh_user_token(user_social):
"""
Utility function to refresh the access token if is (almost) expired
Args:
user_social (UserSocialAuth): a user social auth instance
"""
try:
last_update = datetime.fromtimestamp(user_social.extra_data.get('updated_at'), tz=pytz.UTC)
expires_in = timedelta(seconds=user_social.extra_data.get('expires_in'))
except TypeError:
_send_refresh_request(user_social)
return
# small error margin of 5 minutes to be safe
error_margin = timedelta(minutes=5)
if now_in_utc() - last_update >= expires_in - error_margin:
_send_refresh_request(user_social)
python类UTC的实例源码
def parse_datetime(dt_string):
"""
Attempts to parse a datetime string with any one of the datetime formats that we
expect from Pearson
Args:
dt_string (str): datetime string to be parsed
Returns:
datetime.datetime: parsed datetime
Raises:
UnparsableRowException:
Thrown if the datetime string cannot be parsed with any of the accepted formats
"""
for dt_format in PEARSON_DATETIME_FORMATS:
try:
return datetime.strptime(dt_string, dt_format).replace(tzinfo=pytz.UTC)
except ValueError:
pass
raise UnparsableRowException('Unparsable datetime: {}'.format(dt_string))
def __init__(self, regr, key, meta=None):
self.key = key
self.regr = regr
self.meta = self.Meta(
# pyrfc3339 drops microseconds, make sure __eq__ is sane
creation_dt=datetime.datetime.now(
tz=pytz.UTC).replace(microsecond=0),
creation_host=socket.getfqdn()) if meta is None else meta
self.id = hashlib.md5(
self.key.key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo)
).hexdigest()
# Implementation note: Email? Multiple accounts can have the
# same email address. Registration URI? Assigned by the
# server, not guaranteed to be stable over time, nor
# canonical URI can be generated. ACME protocol doesn't allow
# account key (and thus its fingerprint) to be updated...
def add_time_interval(base_time, interval, textparser=parsedatetime.Calendar()):
"""Parse the time specified time interval, and add it to the base_time
The interval can be in the English-language format understood by
parsedatetime, e.g., '10 days', '3 weeks', '6 months', '9 hours', or
a sequence of such intervals like '6 months 1 week' or '3 days 12
hours'. If an integer is found with no associated unit, it is
interpreted by default as a number of days.
:param datetime.datetime base_time: The time to be added with the interval.
:param str interval: The time interval to parse.
:returns: The base_time plus the interpretation of the time interval.
:rtype: :class:`datetime.datetime`"""
if interval.strip().isdigit():
interval += " days"
# try to use the same timezone, but fallback to UTC
tzinfo = base_time.tzinfo or pytz.UTC
return textparser.parseDT(interval, base_time, tzinfo=tzinfo)[0]
def test_to_local_timezone(self):
i18n.get_i18n().set_timezone('US/Eastern')
format = '%Y-%m-%d %H:%M:%S %Z%z'
# Test datetime with timezone set
base = datetime.datetime(2002, 10, 27, 6, 0, 0, tzinfo=pytz.UTC)
localtime = i18n.to_local_timezone(base)
result = localtime.strftime(format)
self.assertEqual(result, '2002-10-27 01:00:00 EST-0500')
# Test naive datetime - no timezone set
base = datetime.datetime(2002, 10, 27, 6, 0, 0)
localtime = i18n.to_local_timezone(base)
result = localtime.strftime(format)
self.assertEqual(result, '2002-10-27 01:00:00 EST-0500')
def test_to_utc(self):
i18n.get_i18n().set_timezone('US/Eastern')
format = '%Y-%m-%d %H:%M:%S'
# Test datetime with timezone set
base = datetime.datetime(2002, 10, 27, 6, 0, 0, tzinfo=pytz.UTC)
localtime = i18n.to_utc(base)
result = localtime.strftime(format)
self.assertEqual(result, '2002-10-27 06:00:00')
# Test naive datetime - no timezone set
base = datetime.datetime(2002, 10, 27, 6, 0, 0)
localtime = i18n.to_utc(base)
result = localtime.strftime(format)
self.assertEqual(result, '2002-10-27 11:00:00')
def format_datetime(dttm):
# 1. Convert to timezone-aware
# 2. Convert to UTC
# 3. Format in ISO format
# 4. Add subsecond value if non-zero
# 5. Add "Z"
if dttm.tzinfo is None or dttm.tzinfo.utcoffset(dttm) is None:
# dttm is timezone-naive; assume UTC
zoned = pytz.utc.localize(dttm)
else:
zoned = dttm.astimezone(pytz.utc)
ts = zoned.strftime("%Y-%m-%dT%H:%M:%S")
if zoned.microsecond > 0:
ms = zoned.strftime("%f")
ts = ts + '.' + ms.rstrip("0")
return ts + "Z"
def __init__(self, test_id, status, message, timestamp=None):
"""
Parametrized constructor for the TestProgress model
:param test_id: The value of the 'test_id' field of TestProgress model
:type test_id: int
:param status: The value of the 'status' field of TestProgress model
:type status: TestStatus
:param message: The value of the 'message' field of TestProgress model
:type message: str
:param timestamp: The value of the 'timestamp' field of TestProgress
model (None by default)
:type timestamp: datetime
"""
self.test_id = test_id
self.status = status
tz = get_localzone()
if timestamp is None:
timestamp = tz.localize(datetime.datetime.now(), is_dst=None)
timestamp = timestamp.astimezone(pytz.UTC)
if timestamp.tzinfo is None:
timestamp = pytz.utc.localize(timestamp, is_dst=None)
self.timestamp = timestamp
self.message = message
def get_gsod_data(self, station, year):
filename_format = '/pub/data/gsod/{year}/{station}-{year}.op.gz'
lines = self._retreive_file_lines(filename_format, station, year)
dates = pd.date_range("{}-01-01 00:00".format(year),
"{}-12-31 00:00".format(year),
freq='D', tz=pytz.UTC)
series = pd.Series(None, index=dates, dtype=float)
for line in lines[1:]:
columns = line.split()
date_str = columns[2].decode('utf-8')
temp_F = float(columns[3])
temp_C = (5. / 9.) * (temp_F - 32.)
dt = pytz.UTC.localize(datetime.strptime(date_str, "%Y%m%d"))
series[dt] = temp_C
return series
def get_isd_data(self, station, year):
filename_format = '/pub/data/noaa/{year}/{station}-{year}.gz'
lines = self._retreive_file_lines(filename_format, station, year)
dates = pd.date_range("{}-01-01 00:00".format(year),
"{}-12-31 23:00".format(int(year) + 1),
freq='H', tz=pytz.UTC)
series = pd.Series(None, index=dates, dtype=float)
for line in lines:
if line[87:92].decode('utf-8') == "+9999":
temp_C = float("nan")
else:
temp_C = float(line[87:92]) / 10.
date_str = line[15:27].decode('utf-8')
# there can be multiple readings per hour, so set all to minute 0
dt = pytz.UTC.localize(datetime.strptime(date_str, "%Y%m%d%H%M")).replace(minute=0)
# only set the temp if it's the first encountered in the hour.
if pd.isnull(series.ix[dt]):
series[dt] = temp_C
return series
def to_records(self, df):
records = []
if df.shape[0] > 0:
records.append({
"end": pytz.UTC.localize(df.index[0].to_pydatetime()),
"value": np.nan,
"estimated": False,
})
for e, v, est in zip(df.index[1:], df.value, df.estimated):
records.append({
"end": pytz.UTC.localize(e.to_pydatetime()),
"value": v,
"estimated": bool(est),
})
return records
def test_basic_usage():
start_date = datetime(2000, 1, 1, tzinfo=pytz.UTC)
end_date = datetime(2001, 1, 1, tzinfo=pytz.UTC)
interventions = [Intervention(start_date, end_date)]
mps = get_modeling_period_set(interventions)
groups = list(mps.iter_modeling_period_groups())
assert len(groups) == 1
labels, periods = groups[0]
assert labels == ('baseline', 'reporting')
assert periods[0].start_date is None
assert periods[0].end_date is start_date
assert periods[1].start_date is end_date
assert periods[1].end_date is None
modeling_periods = list(mps.iter_modeling_periods())
assert len(modeling_periods) == 2
def test_multiple_records_no_gap(serializer):
records = [
{
"start": datetime(2000, 1, 1, tzinfo=pytz.UTC),
"end": datetime(2000, 1, 2, tzinfo=pytz.UTC),
"value": 1,
},
{
"start": datetime(2000, 1, 2, tzinfo=pytz.UTC),
"end": datetime(2000, 1, 3, tzinfo=pytz.UTC),
"value": 2,
},
]
df = serializer.to_dataframe(records)
assert df.value[datetime(2000, 1, 1, tzinfo=pytz.UTC)] == 1
assert not df.estimated[datetime(2000, 1, 1, tzinfo=pytz.UTC)]
assert df.value[datetime(2000, 1, 2, tzinfo=pytz.UTC)] == 2
assert not df.estimated[datetime(2000, 1, 2, tzinfo=pytz.UTC)]
assert pd.isnull(df.value[datetime(2000, 1, 3, tzinfo=pytz.UTC)])
assert not df.estimated[datetime(2000, 1, 3, tzinfo=pytz.UTC)]
def test_multiple_records_with_gap(serializer):
records = [
{
"start": datetime(2000, 1, 1, tzinfo=pytz.UTC),
"end": datetime(2000, 1, 2, tzinfo=pytz.UTC),
"value": 1,
},
{
"start": datetime(2000, 1, 3, tzinfo=pytz.UTC),
"end": datetime(2000, 1, 4, tzinfo=pytz.UTC),
"value": 2,
},
]
df = serializer.to_dataframe(records)
assert df.value[datetime(2000, 1, 1, tzinfo=pytz.UTC)] == 1
assert not df.estimated[datetime(2000, 1, 1, tzinfo=pytz.UTC)]
assert pd.isnull(df.value[datetime(2000, 1, 2, tzinfo=pytz.UTC)])
assert not df.estimated[datetime(2000, 1, 2, tzinfo=pytz.UTC)]
assert df.value[datetime(2000, 1, 3, tzinfo=pytz.UTC)] == 2
assert not df.estimated[datetime(2000, 1, 3, tzinfo=pytz.UTC)]
assert pd.isnull(df.value[datetime(2000, 1, 4, tzinfo=pytz.UTC)])
assert not df.estimated[datetime(2000, 1, 4, tzinfo=pytz.UTC)]
def test_multiple_records(serializer):
records = [
{
"start": datetime(2000, 1, 1, tzinfo=pytz.UTC),
"value": 1,
},
{
"start": datetime(2000, 1, 2, tzinfo=pytz.UTC),
"value": 2,
},
]
df = serializer.to_dataframe(records)
assert df.value[datetime(2000, 1, 1, tzinfo=pytz.UTC)] == 1
assert not df.estimated[datetime(2000, 1, 1, tzinfo=pytz.UTC)]
assert pd.isnull(df.value[datetime(2000, 1, 2, tzinfo=pytz.UTC)])
assert not df.estimated[datetime(2000, 1, 2, tzinfo=pytz.UTC)]
def test_to_records(serializer):
data = {"value": [1, np.nan], "estimated": [True, False]}
columns = ["value", "estimated"]
index = pd.date_range('2000-01-01', periods=2, freq='D')
df = pd.DataFrame(data, index=index, columns=columns)
records = serializer.to_records(df)
assert len(records) == 2
assert records[0]["start"] == datetime(2000, 1, 1, tzinfo=pytz.UTC)
assert records[0]["value"] == 1
assert records[0]["estimated"]
assert records[1]["start"] == datetime(2000, 1, 2, tzinfo=pytz.UTC)
assert pd.isnull(records[1]["value"])
assert not records[1]["estimated"]
def test_sales_to_user_in_period(self):
res = views._sales_to_user_in_period(
self.alan.username,
timezone.datetime(2017, 2, 1, 0, 0, tzinfo=pytz.UTC),
timezone.datetime(2017, 2, 17, 0, 0, tzinfo=pytz.UTC),
[self.flan.id, self.flanmad.id],
{self.flan.name: 0, self.flanmad.name: 0},
)
self.assertEqual(2, res[self.flan.name])
self.assertEqual(1, res[self.flanmad.name])
def test_sales_to_user_no_results_out_of_period(self):
res = views._sales_to_user_in_period(
self.bob.username,
timezone.datetime(2017, 2, 1, 0, 0, tzinfo=pytz.UTC),
timezone.datetime(2017, 2, 17, 0, 0, tzinfo=pytz.UTC),
[self.flan.id, self.flanmad.id],
{self.flan.name: 0, self.flanmad.name: 0},
)
self.assertEqual(0, res[self.flan.name])
self.assertEqual(0, res[self.flanmad.name])
