goals/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
   6
   7 ## Todo List
   8 # - [ ] goal / vision / task
   9
  10 DB_FILE = os.path.join(os.environ.get('DATA_DIR'), 'goals.db')
  11
  12 DATE_FORMAT = '%Y-%m-%d'
  13
  14 def today():
  15     return dt.now().strftime(DATE_FORMAT)
  16
  17
  18 logger = logging.getLogger(__name__)
  19 logging.basicConfig(filename='chore.log', encoding='utf-8', level=logging.DEBUG)
  20
  21
  22 class Vision(SQLModel, table=True):
  23     id: int | None = Field(default=None, primary_key=True)
  24     name: str = Field(default=None, unique=True)
  25     complete_date: dt = Field(default=None)
  26     def __str__(self):
  27         return f'{self.name} ({self.id})'
  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     complete_date: dt = Field(default=None)
  34     vision_id: int | None = Field(default=None, foreign_key="vision.id")
  35     notes: str = Field(default=None)
  36     def __str__(self):
  37         return f'{self.name} ({self.id})'
  38
  39 class Task(SQLModel, table=True):
  40     id: int | None = Field(default=None, primary_key=True)
  41     name: str = Field(default=None, unique=True)
  42     goal_id: int | None = Field(default=None, foreign_key="goal.id")
  43     complete_date: dt = Field(default=None)
  44
  45 class Data():
  46     def __init__(self, db_file = DB_FILE):
  47         self.file = db_file
  48         self.engine = self.get_engine()
  49
  50     def get_engine(self):
  51         return create_engine(f"sqlite:///{self.file}")
  52
  53     def setup(self):
  54         SQLModel.metadata.create_all(self.engine)
  55
  56
  57     def print_all_visions(self):
  58         pass
  59
  60     def print_all_goals(self):
  61         pass
  62
  63     def print_ophan_tasks(self):
  64         pass
  65
  66
  67     def print_goal(self, goal, include_finished=False):
  68         '''Includes tasks'''
  69         pass
  70
  71     def complete_vision(self, id):
  72         pass
  73
  74     def complete_goal(self, id):
  75         pass
  76
  77     def complete_task(self, id):
  78         pass
  79
  80     def add_with_name(self, cls, name):
  81         try:
  82             with Session(self.engine) as session:
  83                 new = cls(name=name)
  84                 session.add(chore)
  85                 session.commit()
  86             return f'Created {name} [{str(cls)}].'
  87         except:
  88             return f'Unable to create {name} [{str(cls)}]. May be duplicate.'
  89
  90
  91     def add_vision(self, name):
  92         return self.add_with_name(Vision, name)
  93
  94     def add_goal(self, name):
  95         return self.add_with_name(Goal, name)
  96
  97     def add_task(self, name, goal_id):
  98         result = self.add_with_name(Task, name)
  99         with Session(self.engine) as session:
 100             new = select(Task).where(Task.name == name).first()
 101             new.goal_id = goal_id
 102             session.commit()
 103         return result
 104
 105     def assign_vision(self, goal_id, vision_id):
 106         '''Goals may move between visions'''
 107         with Session(self.engine) as session:
 108             new = select(Goal).where(Goal.id == goal_id).first()
 109             new.vision_id = vision_id
 110             session.commit()
 111
 112     def add_note(self, note, goal):
 113         '''Notes print with the goal.'''
 114         pass
 115
 116     # Someday: Notes on each
 117
 118     def check(self, chore_id):
 119         with Session(self.engine) as session:
 120             return Data.check_chore(session, chore_id)
 121
 122     def list(self):
 123         results = []
 124         with Session(self.engine) as session:
 125             cmd = select(Chore)
 126             chores = session.exec(cmd).all()
 127             if(not chores):
 128                 retuls.append('None Found')
 129
 130             for chore in chores:
 131                 results.append(Data.check_chore(session, chore.id))
 132         return results
 133
 134     def add(self, name):
 135         try:
 136             with Session(self.engine) as session:
 137                 new_chore = Chore(name=name)
 138                 session.add(new_chore)
 139                 session.commit()
 140             return f'Created chore {name}.'
 141         except:
 142             return f'Unable to create {name}. May be duplicate.'
 143
 144
 145
 146     def done(self, chore_id):
 147         with Session(self.engine) as session:
 148             chore = get_chore(session, chore_id)
 149             if(not chore):
 150                 return f'Chore not found: {chore_id}'
 151
 152             try:
 153                 session.add(Record(chore_id=chore.id))
 154                 session.commit()
 155                 return f'Marked chore as done: {chore.name}.'
 156             except Exception as ex:
 157                 return f'Unable to mark chore as done: {ex}'
 158
 159
 160     def undo(self, chore_id):
 161         with Session(self.engine) as session:
 162             cmd = select(Chore).where(Chore.id == chore_id)
 163             chore = session.exec(cmd).first()
 164
 165             if(not chore):
 166                 return f'Not found: {chore_id}'
 167
 168             try:
 169                 cmd = select(Record).where(Record.chore_id == chore_id).where(Record.complete_date >= today())
 170                 results = session.exec(cmd).all()
 171                 for result in results:
 172                     session.delete(result)
 173                 session.commit()
 174                 return f'Marked chore as *not* done: {chore.name}.'
 175             except Exception as ex:
 176                 return f'Unable to mark chore as done: {ex}'
 177
 178     def remove(self, chore_id):
 179         try:
 180             with Session(self.engine) as session:
 181                 chore = get_chore(session, chore_id)
 182                 if(not chore):
 183                     return f'Chore not found: {chore_id}'
 184
 185                 sql = text('''
 186                 DELETE FROM Record
 187                 WHERE chore_id = :chore_id
 188                ''')
 189                 params = {}
 190                 params['chore_id'] = chore_id
 191                 results = session.execute(sql, params)
 192                 logger.debug(results)
 193                 sql = text('''
 194                 DELETE FROM Chore
 195                 WHERE id = :chore_id
 196                ''')
 197                 results = session.execute(sql, params)
 198                 logger.debug(results)
 199                 session.commit()
 200                 return f'Removed chore {chore_id}'
 201         except Exception as ex:
 202             return f'Unable to remove chore: {ex}'