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 print_ophan_tasks(self):
 110         pass
 111
 112     def get_all_goals(self):
 113         '''List all goals'''
 114         with Session(self.engine) as session:
 115             cmd = select(Goal).where(Goal.complete_date == None)
 116             return session.exec(cmd).all()
 117
 118
 119     def get_done_goals(self):
 120         '''List completed goals'''
 121         with Session(self.engine) as session:
 122             cmd = select(Goal).where(Goal.complete_date != None)
 123             return session.exec(cmd).all()
 124
 125
 126     def get_goal(self, goal_id):
 127         self._set_pref_goal_id(goal_id)
 128         with Session(self.engine) as session:
 129             cmd = select(Goal).where(Goal.id == goal_id)
 130             goal = session.exec(cmd).first()
 131             return (goal, goal.tasks)
 132
 133     def get_vision(self, id):
 134         with Session(self.engine) as session:
 135             cmd = select(Vision).where(Vision.id == id)
 136             vision = session.exec(cmd).first()
 137             if vision:
 138                 return (vision, vision.goals)
 139             else:
 140                 return ('Not found', [])
 141
 142
 143     def complete_vision(self, id):
 144         with Session(self.engine) as session:
 145             cmd = select(Vision).where(Vision.id == id)
 146             vision = session.exec(cmd).first()
 147             vision.complete_date = dt.today()
 148             session.commit()
 149
 150
 151     def complete_goal(self, goal_id):
 152         self._set_pref_goal_id(goal_id)
 153         vision_id = -1
 154         with Session(self.engine) as session:
 155             cmd = select(Goal).where(Goal.id == goal_id)
 156             goal = session.exec(cmd).first()
 157             goal.complete_date = dt.today()
 158             vision_id = goal.vision_id
 159             session.commit()
 160         return vision_id
 161
 162     def rename_goal(self, goal_id, name):
 163         self._set_pref_goal_id(goal_id)
 164         vision_id = -1
 165         with Session(self.engine) as session:
 166             cmd = select(Goal).where(Goal.id == goal_id)
 167             goal = session.exec(cmd).first()
 168             goal.name = name
 169             vision_id = goal.vision_id
 170             session.commit()
 171         return vision_id
 172
 173
 174     def complete_task(self, id):
 175         with Session(self.engine) as session:
 176             cmd = select(Task).where(Task.id == id)
 177             task = session.exec(cmd).first()
 178             task.complete_date = dt.today()
 179             goal_id = task.goal_id
 180             session.commit()
 181         self._set_pref_goal_id(goal_id)
 182         return goal_id
 183
 184     def add_with_name(self, cls, name):
 185         id = -1
 186         with Session(self.engine) as session:
 187             new = cls(name=name)
 188             session.add(new)
 189             session.commit()
 190             id = new.id
 191         return id
 192
 193     def add_vision(self, name):
 194         return self.add_with_name(Vision, name)
 195
 196     def add_goal(self, name):
 197         goal_id = self.add_with_name(Goal, name)
 198         self._set_pref_goal_id(goal_id)
 199         return goal_id
 200
 201     def quick_add_task(self, name):
 202         goal_id = self._get_pref_goal_id()
 203         return self.add_task(goal_id, name)
 204
 205     def add_task(self, goal_id, name):
 206         self._set_pref_goal_id(goal_id)
 207         result = self.add_with_name(Task, name)
 208         with Session(self.engine) as session:
 209             cmd = select(Task).where(Task.name == name)
 210             new = session.exec(cmd).first()
 211             new.goal_id = goal_id
 212             session.commit()
 213         return result
 214
 215     def assign_vision(self, goal_id, vision_id):
 216         '''Goals may move between visions'''
 217         self._set_pref_goal_id(goal_id)
 218         with Session(self.engine) as session:
 219             cmd = select(Goal).where(Goal.id == goal_id)
 220             goal = session.exec(cmd).first()
 221             goal.vision_id = vision_id
 222             session.commit()
 223
 224     def visionless(self):
 225         '''Get goals with no vision'''
 226         with Session(self.engine) as session:
 227             cmd = select(Goal).where(Goal.vision_id  == None)
 228             return session.exec(cmd).all()
 229
 230
 231     def add_note(self, note, goal):
 232         '''Notes print with the goal.'''
 233         pass
 234
 235     # Someday: Notes on each
 236