Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def fromVEvents(cls, events_list, ref=None, **kwargs):
"""
:type events: list
"""
assert isinstance(events_list, list)
vevents = dict()
for event in events_list:
if 'RECURRENCE-ID' in event:
if invalid_timezone(event['RECURRENCE-ID']):
default_timezone = kwargs['locale']['default_timezone']
recur_id = default_timezone.localize(event['RECURRENCE-ID'].dt)
ident = str(to_unix_time(recur_id))
else:
ident = str(to_unix_time(event['RECURRENCE-ID'].dt))
vevents[ident] = event
else:
vevents['PROTO'] = event
if ref is None:
ref = 'PROTO' if ref in vevents.keys() else list(vevents.keys())[0]
try:
if type(vevents[ref]['DTSTART'].dt) != type(vevents[ref]['DTEND'].dt): # noqa: E721
raise ValueError('DTSTART and DTEND should be of the same type (datetime or date)')
except KeyError:
pass
if kwargs.get('start'):
instcls = cls._get_type_from_date(kwargs.get('start'))
else:
instcls = cls._get_type_from_vDDD(vevents[ref]['DTSTART'])
thisandfuture = (rrange == THISANDFUTURE)
if thisandfuture:
start_shift, duration = calc_shift_deltas(vevent)
start_shift_seconds = start_shift.days * 3600 * 24 + start_shift.seconds
duration_seconds = duration.days * 3600 * 24 + duration.seconds
dtstartend = expand_vevent(vevent, href)
if not dtstartend:
# Does this event even have dates? Technically it is possible for
# events to be empty/non-existent by deleting all their recurrences
# through EXDATE.
return
for dtstart, dtend in dtstartend:
if dtype == EventType.DATE:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
else:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
if rec_id is not None:
ref = rec_inst = str(utils.to_unix_time(rec_id.dt))
else:
rec_inst = str(dbstart)
ref = PROTO
if thisandfuture:
recs_sql_s = (
'UPDATE {0} SET dtstart = rec_inst + ?, dtend = rec_inst + ?, ref = ? '
'WHERE rec_inst >= ? AND href = ? AND calendar = ?;'.format(recs_table))
stuple_f = (
def get_floating_calendars(self, start: dt.datetime, end: dt.datetime) -> Iterable[str]:
assert start.tzinfo is None
assert end.tzinfo is None
start_u = utils.to_unix_time(start)
end_u = utils.to_unix_time(end)
sql_s = (
'SELECT events.calendar FROM '
'recs_float JOIN events ON '
'recs_float.href = events.href AND '
'recs_float.calendar = events.calendar WHERE '
'(dtstart >= ? AND dtstart < ? OR '
'dtend > ? AND dtend <= ? OR '
'dtstart <= ? AND dtend > ? ) AND events.calendar in ({0}) '
'ORDER BY dtstart')
stuple = tuple(
[start_u, end_u, start_u, end_u, start_u, end_u] + list(self.calendars)) # type: ignore
result = self.sql_ex(sql_s.format(','.join(["?"] * len(self.calendars))), stuple)
for calendar in result:
yield calendar[0]
def get_floating(self, start, end) \
-> Iterable[Tuple[str, str, dt.datetime, dt.datetime, str, str, str]]:
"""return floating events between `start` and `end`
:type start: datetime.datetime
:type end: datetime.datetime
"""
assert start.tzinfo is None
assert end.tzinfo is None
start_u = utils.to_unix_time(start)
end_u = utils.to_unix_time(end)
sql_s = (
'SELECT item, recs_float.href, dtstart, dtend, ref, etag, dtype, events.calendar '
'FROM recs_float JOIN events ON '
'recs_float.href = events.href AND '
'recs_float.calendar = events.calendar WHERE '
'(dtstart >= ? AND dtstart < ? OR '
'dtend > ? AND dtend <= ? OR '
'dtstart <= ? AND dtend > ? ) AND events.calendar in ({0}) '
'ORDER BY dtstart')
stuple = tuple(
[start_u, end_u, start_u, end_u, start_u, end_u] + list(self.calendars)) # type: ignore
result = self.sql_ex(sql_s.format(','.join(["?"] * len(self.calendars))), stuple)
for item, href, start, end, ref, etag, dtype, calendar in result:
start = dt.datetime.utcfromtimestamp(start)
end = dt.datetime.utcfromtimestamp(end)
if dtype == EventType.DATE:
def get_localized(self, start, end) \
-> Iterable[Tuple[str, str, dt.datetime, dt.datetime, str, str, str]]:
"""returns
:type start: datetime.datetime
:type end: datetime.datetime
:param minimal: if set, we do not return an event but a minimal stand in
:type minimal: bool
"""
assert start.tzinfo is not None
assert end.tzinfo is not None
start = utils.to_unix_time(start)
end = utils.to_unix_time(end)
sql_s = (
'SELECT item, recs_loc.href, dtstart, dtend, ref, etag, dtype, events.calendar '
'FROM recs_loc JOIN events ON '
'recs_loc.href = events.href AND '
'recs_loc.calendar = events.calendar WHERE '
'(dtstart >= ? AND dtstart <= ? OR '
'dtend > ? AND dtend <= ? OR '
'dtstart <= ? AND dtend >= ?) AND events.calendar in ({0}) '
'ORDER BY dtstart')
stuple = tuple([start, end, start, end, start, end] + list(self.calendars))
result = self.sql_ex(sql_s.format(','.join(["?"] * len(self.calendars))), stuple)
for item, href, start, end, ref, etag, dtype, calendar in result:
start = pytz.UTC.localize(dt.datetime.utcfromtimestamp(start))
end = pytz.UTC.localize(dt.datetime.utcfromtimestamp(end))
yield item, href, start, end, ref, etag, calendar
def get_localized(self, start, end) \
-> Iterable[Tuple[str, str, dt.datetime, dt.datetime, str, str, str]]:
"""returns
:type start: datetime.datetime
:type end: datetime.datetime
:param minimal: if set, we do not return an event but a minimal stand in
:type minimal: bool
"""
assert start.tzinfo is not None
assert end.tzinfo is not None
start = utils.to_unix_time(start)
end = utils.to_unix_time(end)
sql_s = (
'SELECT item, recs_loc.href, dtstart, dtend, ref, etag, dtype, events.calendar '
'FROM recs_loc JOIN events ON '
'recs_loc.href = events.href AND '
'recs_loc.calendar = events.calendar WHERE '
'(dtstart >= ? AND dtstart <= ? OR '
'dtend > ? AND dtend <= ? OR '
'dtstart <= ? AND dtend >= ?) AND events.calendar in ({0}) '
'ORDER BY dtstart')
stuple = tuple([start, end, start, end, start, end] + list(self.calendars))
result = self.sql_ex(sql_s.format(','.join(["?"] * len(self.calendars))), stuple)
for item, href, start, end, ref, etag, dtype, calendar in result:
start = pytz.UTC.localize(dt.datetime.utcfromtimestamp(start))
end = pytz.UTC.localize(dt.datetime.utcfromtimestamp(end))
yield item, href, start, end, ref, etag, calendar
duration_seconds = duration.days * 3600 * 24 + duration.seconds
dtstartend = expand_vevent(vevent, href)
if not dtstartend:
# Does this event even have dates? Technically it is possible for
# events to be empty/non-existent by deleting all their recurrences
# through EXDATE.
return
for dtstart, dtend in dtstartend:
if dtype == EventType.DATE:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
else:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
if rec_id is not None:
ref = rec_inst = str(utils.to_unix_time(rec_id.dt))
else:
rec_inst = str(dbstart)
ref = PROTO
if thisandfuture:
recs_sql_s = (
'UPDATE {0} SET dtstart = rec_inst + ?, dtend = rec_inst + ?, ref = ? '
'WHERE rec_inst >= ? AND href = ? AND calendar = ?;'.format(recs_table))
stuple_f = (
start_shift_seconds, start_shift_seconds + duration_seconds,
ref, rec_inst, href, calendar,
)
self.sql_ex(recs_sql_s, stuple_f)
def fromVEvents(cls, events_list, ref=None, **kwargs):
"""
:type events: list
"""
assert isinstance(events_list, list)
vevents = dict()
for event in events_list:
if 'RECURRENCE-ID' in event:
if invalid_timezone(event['RECURRENCE-ID']):
default_timezone = kwargs['locale']['default_timezone']
recur_id = default_timezone.localize(event['RECURRENCE-ID'].dt)
ident = str(to_unix_time(recur_id))
else:
ident = str(to_unix_time(event['RECURRENCE-ID'].dt))
vevents[ident] = event
else:
vevents['PROTO'] = event
if ref is None:
ref = 'PROTO' if ref in vevents.keys() else list(vevents.keys())[0]
try:
if type(vevents[ref]['DTSTART'].dt) != type(vevents[ref]['DTEND'].dt): # noqa: E721
raise ValueError('DTSTART and DTEND should be of the same type (datetime or date)')
except KeyError:
pass
if kwargs.get('start'):
instcls = cls._get_type_from_date(kwargs.get('start'))
def get_localized_calendars(self, start: dt.datetime, end: dt.datetime) -> Iterable[str]:
assert start.tzinfo is not None
assert end.tzinfo is not None
start_u = utils.to_unix_time(start)
end_u = utils.to_unix_time(end)
sql_s = (
'SELECT events.calendar FROM '
'recs_loc JOIN events ON '
'recs_loc.href = events.href AND '
'recs_loc.calendar = events.calendar WHERE '
'(dtstart >= ? AND dtstart <= ? OR '
'dtend > ? AND dtend <= ? OR '
'dtstart <= ? AND dtend >= ?) AND events.calendar in ({0}) '
'ORDER BY dtstart')
stuple = tuple(
[start_u, end_u, start_u, end_u, start_u, end_u] + list(self.calendars)) # type: ignore
result = self.sql_ex(sql_s.format(','.join(["?"] * len(self.calendars))), stuple)
for calendar in result:
yield calendar[0] # result is always an iterable, even if getting only one item
if not dtstartend:
# Does this event even have dates? Technically it is possible for
# events to be empty/non-existent by deleting all their recurrences
# through EXDATE.
return
for dtstart, dtend in dtstartend:
if dtype == EventType.DATE:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
else:
dbstart = utils.to_unix_time(dtstart)
dbend = utils.to_unix_time(dtend)
if rec_id is not None:
ref = rec_inst = str(utils.to_unix_time(rec_id.dt))
else:
rec_inst = str(dbstart)
ref = PROTO
if thisandfuture:
recs_sql_s = (
'UPDATE {0} SET dtstart = rec_inst + ?, dtend = rec_inst + ?, ref = ? '
'WHERE rec_inst >= ? AND href = ? AND calendar = ?;'.format(recs_table))
stuple_f = (
start_shift_seconds, start_shift_seconds + duration_seconds,
ref, rec_inst, href, calendar,
)
self.sql_ex(recs_sql_s, stuple_f)
else:
recs_sql_s = (
'INSERT OR REPLACE INTO {0} '