matrix_goal_bot/db.py

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