matrix_goal_bot/db.py

   1 import logging
   2 import os
   3 from datetime import datetime as dt
   4
   5 from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
   6
   7
   8 DATE_FORMAT = '%Y-%m-%d'
   9
  10 def today():
  11     return dt.now().strftime(DATE_FORMAT)
  12
  13
  14 logger = logging.getLogger(__name__)
  15 logging.basicConfig(filename='goals.log', encoding='utf-8', level=logging.DEBUG)
  16
  17
  18 class Vision(SQLModel, table=True):
  19     id: int | None = Field(default=None, primary_key=True)
  20     name: str = Field(default=None, unique=True)
  21     complete_date: dt | None = Field(default=None)
  22     goals: list["Goal"] = Relationship(back_populates="vision")
  23     def __str__(self):
  24         return f'{self.name} (Vision {self.id})'
  25
  26
  27 class PreferGoal(SQLModel, table=True):
  28     goal_id: int | None = Field(default=None, foreign_key="goal.id", primary_key=True)
  29
  30
  31 class Goal(SQLModel, table=True):
  32     id: int | None = Field(default=None, primary_key=True)
  33     name: str = Field(default=None, unique=True)
  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 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 def pretty_print(goal, tasks):
  57     results = []
  58     results.append(f' ## {goal}')
  59     if not tasks:
  60         results.append('-- Nothing to list --')
  61
  62     for task in tasks:
  63         if(task.complete_date):
  64             results.append(f' - [x] {task}')
  65         else:
  66             results.append(f' - [ ] {task}')
  67     return results
  68
  69
  70 class Data():
  71     def __init__(self, db_file=None):
  72         if not db_file:
  73             db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), 'goals.db')
  74
  75         self.file = db_file
  76         self.engine = self.get_engine()
  77
  78
  79     def get_engine(self):
  80         return create_engine(f"sqlite:///{self.file}")
  81
  82     def setup(self):
  83         SQLModel.metadata.create_all(self.engine)
  84         return 'Success'
  85
  86
  87     def _set_pref_goal_id(self, goal_id):
  88         with Session(self.engine) as session:
  89             prefs = session.exec(select(PreferGoal)).all()
  90             for pref in prefs:
  91                 session.delete(pref)
  92             pref = PreferGoal()
  93             pref.goal_id = goal_id
  94             session.add(pref)
  95             session.commit()
  96
  97     def _get_pref_goal_id(self):
  98         with Session(self.engine) as session:
  99             pref = session.exec(select(PreferGoal)).one()
 100             return pref.goal_id
 101
 102
 103
 104     def get_all_visions(self):
 105         with Session(self.engine) as session:
 106             cmd = select(Vision).where(Vision.complete_date == None)
 107             return session.exec(cmd).all()
 108
 109     def get_all_goals(self):
 110         '''List all goals'''
 111         with Session(self.engine) as session:
 112             cmd = select(Goal).where(Goal.complete_date == None)
 113             return session.exec(cmd).all()
 114
 115
 116     def get_done_goals(self):
 117         '''List completed goals'''
 118         with Session(self.engine) as session:
 119             cmd = select(Goal).where(Goal.complete_date != None)
 120             return session.exec(cmd).all()
 121
 122
 123     def get_goal(self, goal_id):
 124         self._set_pref_goal_id(goal_id)
 125         with Session(self.engine) as session:
 126             cmd = select(Goal).where(Goal.id == goal_id)
 127             goal = session.exec(cmd).first()
 128             return (goal, goal.tasks)
 129
 130     def get_vision(self, id):
 131         with Session(self.engine) as session:
 132             cmd = select(Vision).where(Vision.id == id)
 133             vision = session.exec(cmd).first()
 134             if vision:
 135                 return (vision, vision.goals)
 136             else:
 137                 return ('Not found', [])
 138
 139
 140     def complete_vision(self, id):
 141         with Session(self.engine) as session:
 142             cmd = select(Vision).where(Vision.id == id)
 143             vision = session.exec(cmd).first()
 144             vision.complete_date = dt.today()
 145             session.commit()
 146
 147
 148     def complete_goal(self, goal_id):
 149         self._set_pref_goal_id(goal_id)
 150         vision_id = -1
 151         with Session(self.engine) as session:
 152             cmd = select(Goal).where(Goal.id == goal_id)
 153             goal = session.exec(cmd).first()
 154             goal.complete_date = dt.today()
 155             vision_id = goal.vision_id
 156             session.commit()
 157         return vision_id
 158
 159     def rename_goal(self, goal_id, name):
 160         self._set_pref_goal_id(goal_id)
 161         vision_id = -1
 162         with Session(self.engine) as session:
 163             cmd = select(Goal).where(Goal.id == goal_id)
 164             goal = session.exec(cmd).first()
 165             goal.name = name
 166             vision_id = goal.vision_id
 167             session.commit()
 168         return vision_id
 169
 170
 171     def complete_task(self, id):
 172         with Session(self.engine) as session:
 173             cmd = select(Task).where(Task.id == id)
 174             task = session.exec(cmd).first()
 175             task.complete_date = dt.today()
 176             goal_id = task.goal_id
 177             session.commit()
 178         self._set_pref_goal_id(goal_id)
 179         return goal_id
 180
 181     def add_with_name(self, cls, name):
 182         id = -1
 183         with Session(self.engine) as session:
 184             new = cls(name=name)
 185             session.add(new)
 186             session.commit()
 187             id = new.id
 188         return id
 189
 190     def add_vision(self, name):
 191         return self.add_with_name(Vision, name)
 192
 193     def add_goal(self, name):
 194         goal_id = self.add_with_name(Goal, name)
 195         self._set_pref_goal_id(goal_id)
 196         return goal_id
 197
 198     def quick_add_task(self, name):
 199         goal_id = self._get_pref_goal_id()
 200         return self.add_task(goal_id, name)
 201
 202     def add_task(self, goal_id, name):
 203         result = self.add_with_name(Task, name)
 204         with Session(self.engine) as session:
 205             cmd = select(Task).where(Task.name == name)
 206             new = session.exec(cmd).first()
 207             new.goal_id = goal_id
 208             session.commit()
 209         return result
 210
 211     def assign_vision(self, goal_id, vision_id):
 212         '''Goals may move between visions'''
 213         self._set_pref_goal_id(goal_id)
 214         with Session(self.engine) as session:
 215             cmd = select(Goal).where(Goal.id == goal_id)
 216             goal = session.exec(cmd).first()
 217             goal.vision_id = vision_id
 218             session.commit()
 219
 220     def visionless(self):
 221         '''Get goals with no vision'''
 222         with Session(self.engine) as session:
 223             cmd = select(Goal).where(Goal.vision_id  == None)
 224             return session.exec(cmd).all()
 225
 226
 227     def add_note(self, note, goal):
 228         '''Notes print with the goal.'''
 229         pass
 230
 231     # Someday: Notes on each
 232