matrix_goal_bot/db.py

   1 import logging
   2 import os
   3 from datetime import datetime as dt
   4 from datetime import timedelta
   5 from typing import List
   6
   7 from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
   8
   9 logger = logging.getLogger(__name__)
  10
  11 DATE_FORMAT = '%Y-%m-%d'
  12
  13 def today():
  14     return dt.now().strftime(DATE_FORMAT)
  15
  16
  17 class Vision(SQLModel, table=True):
  18     id: int | None = Field(default=None, primary_key=True)
  19     name: str = Field(default=None, unique=True)
  20     complete_date: dt | None = Field(default=None)
  21     goals: list["Goal"] = Relationship(back_populates="vision")
  22     def __str__(self):
  23         return f'{self.name} (Vision {self.id})'
  24
  25
  26 class PreferGoal(SQLModel, table=True):
  27     goal_id: int | None = Field(default=None, foreign_key="goal.id", primary_key=True)
  28
  29
  30 class Goal(SQLModel, table=True):
  31     id: int | None = Field(default=None, primary_key=True)
  32     name: str = Field(default=None, unique=True)
  33     start_date: dt | None = Field(default=dt.today())
  34     complete_date: dt | None = Field(default=None)
  35     vision_id: int | None = Field(default=None, foreign_key="vision.id")
  36     notes: str | None = Field(default=None)
  37     tasks: list["Task"] = Relationship(back_populates="goal")
  38     vision: Vision = Relationship(back_populates="goals")
  39     def __str__(self):
  40         if not self.complete_date:
  41             return f'{self.name} (Goal {self.id})'
  42         else:
  43             completed = self.complete_date.strftime(DATE_FORMAT)
  44             return f'(Done {completed}) {self.name} (Goal {self.id})'
  45
  46
  47 class Task(SQLModel, table=True):
  48     id: int | None = Field(default=None, primary_key=True)
  49     name: str = Field(default=None, unique=True)
  50     goal_id: int | None = Field(default=None, foreign_key="goal.id")
  51     complete_date: dt | None = Field(default=None)
  52     goal: Goal = Relationship(back_populates="tasks")
  53     def __str__(self):
  54         return f'{self.name} ({self.id})'
  55
  56
  57 class Win(SQLModel, table=True):
  58     id: int | None = Field(default=None, primary_key=True)
  59     goal: str = Field(default=None, unique=False)
  60     vision: str | None = Field(default=None, unique=False)
  61     complete_date: dt | None = Field(default=dt.today())
  62     def __str__(self):
  63         completed = self.complete_date.strftime(DATE_FORMAT)
  64         return f'{self.goal} ({completed} {self.vision})'
  65
  66 def pretty_print(goal, tasks):
  67     results = []
  68     results.append(f' ## {goal}')
  69     if not tasks:
  70         results.append('-- Nothing to list --')
  71
  72     for task in tasks:
  73         if(task.complete_date):
  74             results.append(f' - [x] {task}')
  75         else:
  76             results.append(f' - [ ] {task}')
  77     return results
  78
  79
  80 def _get_with_id(cls, item_id):
  81     return select(cls).where(cls.id == item_id)
  82
  83 def _get_with_name(cls, name):
  84     return select(cls).where(func.lower(cls.name) == func.lower(name))
  85
  86
  87 def _get_goals_by_vision(goals) -> List[str]:
  88     results = []
  89     visions = {}
  90
  91     for goal in goals:
  92         vision = str(goal.vision)
  93         if not vision in visions:
  94             visions[vision] = []
  95         visions[vision].append(str(goal))
  96     for vision in visions:
  97         results.append('') # Make some space
  98         results.append(f'## {vision}')
  99         for item in visions[vision]:
 100             results.append(' + ' + item)
 101     return results
 102
 103
 104 class Data():
 105     def __init__(self, db_file=None):
 106         if not db_file:
 107             db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), 'goal.db')
 108
 109         self.file = db_file
 110         self.engine = self.get_engine()
 111
 112
 113     def get_engine(self):
 114         return create_engine(f"sqlite:///{self.file}")
 115
 116     def setup(self):
 117         SQLModel.metadata.create_all(self.engine)
 118         return 'Success'
 119
 120     def migrate(self):
 121         with Session(self.engine) as session:
 122             session.exec(text(''' alter table goal add start_date DATETIME; '''))
 123             session.commit()
 124         return 'Success'
 125
 126
 127     def _set_pref_goal_id(self, goal_id):
 128         with Session(self.engine) as session:
 129             prefs = session.exec(select(PreferGoal)).all()
 130             for pref in prefs:
 131                 session.delete(pref)
 132             session.commit()
 133             pref = PreferGoal()
 134             pref.goal_id = goal_id
 135             session.add(pref)
 136             session.commit()
 137
 138     def _get_pref_goal_id(self):
 139         with Session(self.engine) as session:
 140             pref = session.exec(select(PreferGoal)).one()
 141             return pref.goal_id
 142
 143
 144     def clear(self, goal_id, clear_all=False):
 145         with Session(self.engine) as session:
 146             goal = session.exec(_get_with_id(Goal, goal_id)).first()
 147             for item in goal.tasks:
 148                 if clear_all or item.complete_date:
 149                     session.delete(item)
 150             session.commit()
 151
 152     def get_all_visions(self):
 153         with Session(self.engine) as session:
 154             cmd = select(Vision).where(Vision.complete_date == None)
 155             return session.exec(cmd).all()
 156
 157     def get_all_goals(self) -> List[str]:
 158         '''List all goals'''
 159         results = []
 160         with Session(self.engine) as session:
 161             cmd = select(Goal).where(Goal.complete_date == None)
 162             goals = session.exec(cmd).all()
 163             results.append(_get_goals_by_vision(goals))
 164         return results
 165
 166
 167     def get_current_goals(self):
 168         '''List current goals'''
 169         with Session(self.engine) as session:
 170             cmd = select(Goal).where(Goal.complete_date == None).where(Goal.start_date <= dt.today())
 171             goals = session.exec(cmd).all()
 172             logger.debug(f'Current goals: {goals}')
 173             return _get_goals_by_vision(goals)
 174
 175     def get_done_goals(self):
 176         '''List completed goals'''
 177         with Session(self.engine) as session:
 178             cmd = select(Goal).where(Goal.complete_date != None)
 179             goals = session.exec(cmd).all()
 180             return _get_goals_by_vision(goals)
 181
 182     def get_goal(self, goal_id):
 183         self._set_pref_goal_id(goal_id)
 184         with Session(self.engine) as session:
 185             cmd = select(Goal).where(Goal.id == goal_id)
 186             goal = session.exec(cmd).first()
 187             return (goal, goal.tasks)
 188
 189     def get_vision(self, id):
 190         with Session(self.engine) as session:
 191             cmd = select(Vision).where(Vision.id == id)
 192             vision = session.exec(cmd).first()
 193             if vision:
 194                 return (vision, vision.goals)
 195             else:
 196                 return ('Not found', [])
 197
 198
 199     def complete_vision(self, id):
 200         with Session(self.engine) as session:
 201             cmd = select(Vision).where(Vision.id == id)
 202             vision = session.exec(cmd).first()
 203             vision.complete_date = dt.today()
 204             session.commit()
 205
 206
 207     def complete_goal(self, goal_id):
 208         vision_id = -1
 209         with Session(self.engine) as session:
 210             cmd = select(Goal).where(Goal.id == goal_id)
 211             goal = session.exec(cmd).first()
 212             goal.complete_date = dt.today()
 213             vision_id = goal.vision_id
 214
 215             cmd = select(Vision).where(Vision.id == vision_id)
 216             win = Win()
 217             win.goal = goal.name
 218
 219             vision = session.exec(cmd).first()
 220             if vision:
 221                 win.vision = vision.name
 222
 223             session.add(win)
 224             session.commit()
 225         return vision_id
 226
 227
 228     def put_off_vision(self, vision_id):
 229         with Session(self.engine) as session:
 230             # vision = session.exec(cmd).first()
 231             cmd = select(Goal).where(Goal.vision_id == vision_id)
 232             goals = session.exec(cmd).all()
 233             for goal in goals:
 234                 goal.start_date = dt.today() + timedelta(days=14)
 235             session.commit()
 236         return True
 237
 238     def start_goal(self, goal_id):
 239         vision = None
 240         with Session(self.engine) as session:
 241             cmd = select(Goal).where(Goal.id == goal_id)
 242             goal = session.exec(cmd).first()
 243             goal.start_date = dt.today()
 244             vision_id = goal.vision_id
 245             session.commit()
 246         if vision:
 247             return (vision, vision.goals)
 248         else:
 249             return ('No Vision Assigned', [])
 250
 251
 252     def rename_goal(self, goal_id, name):
 253         self._set_pref_goal_id(goal_id)
 254         vision_id = -1
 255         with Session(self.engine) as session:
 256             cmd = select(Goal).where(Goal.id == goal_id)
 257             goal = session.exec(cmd).first()
 258             goal.name = name
 259             vision_id = goal.vision_id
 260             session.commit()
 261         return vision_id
 262
 263
 264     def complete_task(self, id):
 265         with Session(self.engine) as session:
 266             cmd = select(Task).where(Task.id == id)
 267             task = session.exec(cmd).first()
 268             task.complete_date = dt.today()
 269             goal_id = task.goal_id
 270             session.commit()
 271         self._set_pref_goal_id(goal_id)
 272         return goal_id
 273
 274     def add_with_name(self, cls, name):
 275         id = -1
 276         with Session(self.engine) as session:
 277             new = cls(name=name)
 278             session.add(new)
 279             session.commit()
 280             id = new.id
 281         return id
 282
 283     def add_vision(self, name):
 284         return self.add_with_name(Vision, name)
 285
 286     def add_goal(self, name):
 287         goal_id = self.add_with_name(Goal, name)
 288         self._set_pref_goal_id(goal_id)
 289         return goal_id
 290
 291     def quick_add_task(self, name):
 292         goal_id = self._get_pref_goal_id()
 293         task_id = self.add_with_name(Task, name)
 294         with Session(self.engine) as session:
 295             cmd = select(Task).where(Task.id == task_id)
 296             new = session.exec(cmd).first()
 297             new.goal_id = goal_id
 298             session.commit()
 299         return goal_id
 300
 301     def remove_task(self, task_id):
 302         task_name = ''
 303         logger.debug(f'Removing task with id {task_id}')
 304         with Session(self.engine) as session:
 305             cmd = select(Task).where(Task.id  == int(task_id))
 306             to_remove = session.exec(cmd).first()
 307             if not to_remove:
 308                 return f'Unable to find task with id {task_id}'
 309
 310             logger.debug(f'Removing task with name {to_remove.name}')
 311             task_name = to_remove.name
 312             session.delete(to_remove)
 313             session.commit()
 314         return f'Removed task "{task_name}"'
 315
 316     def assign_vision(self, goal_id, vision_id):
 317         '''Goals may move between visions'''
 318         self._set_pref_goal_id(goal_id)
 319         with Session(self.engine) as session:
 320             cmd = select(Goal).where(Goal.id == goal_id)
 321             goal = session.exec(cmd).first()
 322             goal.vision_id = vision_id
 323             session.commit()
 324         return True
 325
 326     def visionless(self):
 327         '''Get goals with no vision'''
 328         with Session(self.engine) as session:
 329             cmd = select(Goal).where(Goal.vision_id  == None).where(Goal.complete_date == None)
 330             return session.exec(cmd).all()
 331
 332
 333     def add_note(self, note, goal):
 334         '''Notes print with the goal.'''
 335         pass
 336
 337
 338     def wins(self):
 339         results = []
 340         with Session(self.engine) as session:
 341             results = session.exec(select(Win)).all()
 342         return results
 343
 344     # Someday: Notes on each
 345