99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
import datetime
|
|
import logging
|
|
import re
|
|
import pytz
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Iso8601Serializer:
|
|
marker = 'ISO8601'
|
|
separator = ':'
|
|
|
|
_re = re.compile(r'^(?P<date>'
|
|
r'(?P<year>\d{4})'
|
|
r'-(?P<mon>(0[1-9])|(1[012]))'
|
|
r'-(?P<day>(0[1-9])|([12][0-9])|(3[01]))'
|
|
r')?(?P<sep>.)?'
|
|
r'(?P<time>'
|
|
r'(?P<hour>([01][0-9])|(2[0123]))'
|
|
r':(?P<min>[0-5][0-9])'
|
|
r':(?P<sec>[0-5][0-9])'
|
|
r'([,.]\d{1,10})?'
|
|
r')?'
|
|
r'(?P<offset>(?P<offdir>[\-+])(?P<offhours>([01][0-9])|(2[0123]))(:(?P<offmins>[0-5][0-9]))?)?$')
|
|
|
|
def __init__(self, value):
|
|
if isinstance(value, datetime.datetime) \
|
|
or isinstance(value, datetime.date) \
|
|
or isinstance(value, datetime.time):
|
|
dt_obj = value
|
|
else:
|
|
dt_obj = Iso8601Serializer.deserialize(value)
|
|
self._serialized = Iso8601Serializer.serialize(dt_obj)
|
|
|
|
def __str__(self):
|
|
return self._serialized
|
|
|
|
@classmethod
|
|
def serialize(cls, value, ignore_unsupported_input=False):
|
|
strval = None
|
|
if isinstance(value, datetime.datetime):
|
|
strval = value.isoformat()
|
|
elif isinstance(value, datetime.date):
|
|
strval = value.isoformat()
|
|
elif isinstance(value, datetime.time):
|
|
strval = value.isoformat()
|
|
elif not ignore_unsupported_input:
|
|
raise ValueError('Expected datetime.datetime, datetime.date or datetime.time,'
|
|
' not {}'.format(value.__class__.__name__))
|
|
|
|
if strval is not None:
|
|
value = '{marker}{sep}{str}'.format(marker=cls.marker, sep=cls.separator, str=strval)
|
|
|
|
return value
|
|
|
|
@classmethod
|
|
def deserialize(cls, value, ignore_unsupported_input=False):
|
|
prefix = '{marker}{sep}'.format(marker=cls.marker, sep=cls.separator)
|
|
|
|
if not isinstance(value, str):
|
|
if ignore_unsupported_input:
|
|
return value
|
|
raise TypeError('Expected string type, not {}'.format(value.__class__.__name__))
|
|
|
|
if not value.startswith(prefix):
|
|
if ignore_unsupported_input:
|
|
return value
|
|
raise ValueError('String must begin with \'{prefix}\''.format(prefix=prefix))
|
|
|
|
haystack = value[len(prefix):]
|
|
match = cls._re.match(haystack)
|
|
|
|
if match is None:
|
|
if ignore_unsupported_input:
|
|
return value
|
|
raise ValueError('Format not recognized \'{str}\''.format(str=haystack))
|
|
|
|
gd = match.groupdict()
|
|
if gd['hour'] is None:
|
|
value = datetime.date(int(gd['year']), int(gd['mon']), int(gd['day']))
|
|
elif gd['year'] is None:
|
|
value = datetime.time(hour=int(gd['hour']), minute=int(gd['min']), second=int(gd['sec']))
|
|
else:
|
|
value = datetime.datetime(int(gd['year']), int(gd['mon']), int(gd['day']),
|
|
hour=int(gd['hour']), minute=int(gd['min']), second=int(gd['sec']))
|
|
if gd['offset'] is not None:
|
|
offset = datetime.timedelta(hours=int(gd['offhours']))
|
|
if gd['offmins'] is not None:
|
|
offset += datetime.timedelta(minutes=int(gd['offmins']))
|
|
if gd['offdir'] == '+':
|
|
value -= offset
|
|
elif gd['offdir'] == '-':
|
|
value += offset
|
|
else: # pragma no cover
|
|
raise ValueError('Offset format not recognized \'{str}\''.format(str=gd['offset']))
|
|
value = value.replace(tzinfo=pytz.UTC)
|
|
|
|
return value
|