Reading Ability of Events from String Content

Reading events from pure string content is added. It becomes a need when the calendar content is not served from a URL or a file.
pull/15/head
Cuneyt Mertayak 2018-04-03 22:32:25 -07:00
rodzic 788365c3b3
commit 5831b1ee1f
3 zmienionych plików z 34 dodań i 6 usunięć

Wyświetl plik

@ -73,6 +73,12 @@ class ICalDownload:
return self.decode(content, apple_fix=apple_fix)
def data_from_string(self, string_content, apple_fix=False):
if not string_content or len(string_content) == 0:
raise IOError("String content is not readable or is empty!")
return self.decode(string_content, apple_fix=apple_fix)
def decode(self, content, apple_fix=False):
"""
Decode content using the set charset.

Wyświetl plik

@ -12,12 +12,13 @@ event_store = {}
threads = {}
def events(url=None, file=None, start=None, end=None, fix_apple=False):
def events(url=None, file=None, string_content=None, start=None, end=None, fix_apple=False):
"""
Get all events form the given iCal URL occurring in the given time range.
:param url: iCal URL
:param file: iCal file path
:param string_content: iCal content as string
:param start: start date (see dateutils.date)
:param end: end date (see dateutils.date)
:param fix_apple: fix known Apple iCal issues
@ -33,18 +34,22 @@ def events(url=None, file=None, start=None, end=None, fix_apple=False):
if not content and file:
content = ICalDownload().data_from_file(file, apple_fix=fix_apple)
if not content and string_content:
content = ICalDownload().data_from_string(string_content, apple_fix=fix_apple)
found_events += parse_events(content, start=start, end=end)
return found_events
def request_data(key, url, file, start, end, fix_apple):
def request_data(key, url, file, string_content, start, end, fix_apple):
"""
Request data, update local data cache and remove this Thread form queue.
:param key: key for data source to get result later
:param url: iCal URL
:param file: iCal file path
:param string_content: iCal content as string
:param start: start date
:param end: end date
:param fix_apple: fix known Apple iCal issues
@ -52,24 +57,25 @@ def request_data(key, url, file, start, end, fix_apple):
data = []
try:
data += events(url=url, file=file, start=start, end=end, fix_apple=fix_apple)
data += events(url=url, file=file, string_content=string_content, start=start, end=end, fix_apple=fix_apple)
finally:
update_events(key, data)
request_finished(key)
def events_async(key, url=None, file=None, start=None, end=None, fix_apple=False):
def events_async(key, url=None, file=None, start=None, string_content=None, end=None, fix_apple=False):
"""
Trigger an asynchronous data request.
:param key: key for data source to get result later
:param url: iCal URL
:param file: iCal file path
:param string_content: iCal content as string
:param start: start date
:param end: end date
:param fix_apple: fix known Apple iCal issues
"""
t = Thread(target=request_data, args=(key, url, file, start, end, fix_apple))
t = Thread(target=request_data, args=(key, url, file, string_content, start, end, fix_apple))
with event_lock:
if key not in threads:

Wyświetl plik

@ -74,7 +74,23 @@ class ICalEventsTests(unittest.TestCase):
end = date(2017, 5, 19)
key = "basic"
icalevents.request_data(key, url=None, file=ical, start=start, end=end, fix_apple=False)
icalevents.request_data(key, url=None, file=ical, string_content=None, start=start, end=end, fix_apple=False)
self.assertTrue(icalevents.all_done(key), "request is finished")
self.assertEqual(len(icalevents.latest_events(key)), 2, "two events are found")
def test_string_data(self):
ical = "test/test_data/basic.ics"
with open(ical, mode='rb') as f:
string_content = f.read()
start = date(2017, 5, 18)
end = date(2017, 5, 19)
key = "basic"
icalevents.request_data(key, url=None, file=None, string_content=string_content, start=start, end=end,
fix_apple=False)
self.assertTrue(icalevents.all_done(key), "request is finished")
self.assertEqual(len(icalevents.latest_events(key)), 2, "two events are found")