prettify rtm module

my/rtm.py

@@ -1,27 +1,24 @@
-#!/usr/bin/env python3
+"""
+[[https://rememberthemilk.com][Remember The Milk]] tasks and notes
+"""
+
 # pip3 install icalendar
-import logging
-import os
 import re
-from collections import deque
 from pathlib import Path
-from sys import argv
+from typing import Dict, List, Optional, Iterator
-from typing import Dict, List, Optional, TypeVar
 from datetime import datetime
 
-from kython.klogging import LazyLogger
+from .common import LazyLogger, get_files, group_by_key, cproperty
-from kython import group_by_key, cproperty
+from .kython.kompress import open as kopen
-from kython import kompress
+
+from mycfg import rtm as config
+
 
 import icalendar # type: ignore
 from icalendar.cal import Todo # type: ignore
 
 
-logger = LazyLogger('rtm-provider')
+logger = LazyLogger('my.rtm')
-
-
-def get_last_backup():
-    return max(Path('***REMOVED***').glob('*.ical*'))
 
 
 # TODO extract in a module to parse RTM's ical?
@@ -82,51 +79,43 @@ class MyTodo:
         return (mtodo.revision, mtodo.get_time())
 
 
-class RtmBackup:
+class DAL:
     def __init__(self, data: bytes, revision=None) -> None:
         self.cal = icalendar.Calendar.from_ical(data)
         self.revision = revision
 
-    @staticmethod
+    def all_todos(self) -> Iterator[MyTodo]:
-    def from_path(path: Path) -> 'RtmBackup':
+        for t in self.cal.walk('VTODO'):
-        with kompress.open(path, 'rb') as fo:
+            yield MyTodo(t, self.revision)
-            data = fo.read()
-            revision = 'TODO FIXME' # extract_backup_date(path)
-            return RtmBackup(data, revision)
-
-    def get_all_todos(self) -> List[MyTodo]:
-        return [MyTodo(t, self.revision) for t in self.cal.walk('VTODO')]
 
     def get_todos_by_uid(self) -> Dict[str, MyTodo]:
-        todos = self.get_all_todos()
+        todos = self.all_todos()
+        # TODO use make_dict?
         res = {todo.uid: todo for todo in todos}
-        assert len(res) == len(todos) # hope uid is unique, but just in case
         return res
 
     def get_todos_by_title(self) -> Dict[str, List[MyTodo]]:
-        todos = self.get_all_todos()
+        todos = self.all_todos()
         return group_by_key(todos, lambda todo: todo.title)
 
 
-def get_all_tasks():
+def dal():
-    b = RtmBackup.from_path(get_last_backup())
+    last = get_files(config.export_path, glob='*.ical.xz')[-1]
-    return b.get_all_todos()
+    with kopen(last, 'rb') as fo:
+        data = fo.read()
+    return DAL(data=data, revision='TODO')
 
 
-def get_active_tasks():
+def all_tasks() -> Iterator[MyTodo]:
-    return [t for t in get_all_tasks() if not t.is_completed()]
+    yield from dal().all_todos()
 
 
-def test():
+def active_tasks() -> Iterator[MyTodo]:
-    tasks = get_all_tasks()
+    for t in all_tasks():
-    assert len([t for t in tasks if 'gluons' in t.title]) > 0
+        if not t.is_completed():
+            yield t
 
 
-def main():
+def print_all_todos():
-    backup = RtmBackup.from_path(get_last_backup())
+    for t in all_tasks():
-    for t in backup.get_all_todos():
         print(t)
-
-
-if __name__ == '__main__':
-    main()

tests/rtm.py

@@ -0,0 +1,6 @@
+from my.rtm import all_tasks
+
+
+def test():
+    tasks = all_tasks()
+    assert len([t for t in tasks if 'gluons' in t.title]) > 0