matrix_chore_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
   6
   7 ## Todo List
   8 # - [ ] add Vision FK
   9 # - [ ] add notes to each chore
  10
  11 DATE_FORMAT = '%Y-%m-%d'
  12
  13 def today():
  14     return dt.now().strftime(DATE_FORMAT)
  15
  16 logger = logging.getLogger(__name__)
  17 logging.basicConfig(filename='chore.log', encoding='utf-8', level=logging.DEBUG)
  18
  19 class Chore(SQLModel, table=True):
  20     id: int | None = Field(default=None, primary_key=True)
  21     name: str = Field(default=None, unique=True)
  22
  23     def __str__(self):
  24         return f'{self.name} ({self.id})'
  25
  26
  27 class Record(SQLModel, table=True):
  28     id: int | None = Field(default=None, primary_key=True)
  29     chore_id: int | None = Field(default=None, foreign_key="chore.id")
  30     complete_date: dt = Field(default_factory=lambda: dt.now())
  31
  32
  33 def get_chore(session, chore_id):
  34     cmd = select(Chore).where(Chore.id == chore_id )
  35     return session.exec(cmd).first()
  36
  37 class Data():
  38     def __init__(self, db_file):
  39
  40         self.file = db_file
  41         self.engine = self.get_engine()
  42
  43     def get_db_file(self):
  44         return self.file
  45
  46     def get_engine(self):
  47         return create_engine(f"sqlite:///{self.file}")
  48
  49     def setup(self):
  50         SQLModel.metadata.create_all(self.engine)
  51
  52
  53     @staticmethod
  54     def check_chore(session, chore_id) -> str:
  55         chore = get_chore(session, chore_id)
  56         if(not chore):
  57             return f'Chore not found: {chore_id}'
  58
  59         sql = text('''
  60         SELECT * FROM Record
  61         WHERE chore_id = :chore_id
  62         AND complete_date >= :today
  63         ORDER BY complete_date DESC
  64         LIMIT 1
  65        ''')
  66         params = {}
  67         params['chore_id'] = chore_id
  68         params['today'] = today()
  69         results = session.execute(sql, params).first()
  70
  71         logger.debug(results)
  72         if(results):
  73             return f'- [x] {chore}'
  74         else:
  75             return f'- [ ] {chore}'
  76
  77
  78     def check(self, chore_id):
  79         with Session(self.engine) as session:
  80             return Data.check_chore(session, chore_id)
  81
  82     def list(self):
  83         results = []
  84         with Session(self.engine) as session:
  85             cmd = select(Chore)
  86             chores = session.exec(cmd).all()
  87             if(not chores):
  88                 results.append('None Found')
  89
  90             for chore in chores:
  91                 results.append(Data.check_chore(session, chore.id))
  92         return results
  93
  94     def filter(self, term):
  95         results = []
  96         for item in self.list():
  97             if term in item:
  98                 results.append(item)
  99
 100         return results
 101
 102
 103
 104
 105     def add(self, name):
 106         try:
 107             with Session(self.engine) as session:
 108                 new_chore = Chore(name=name)
 109                 session.add(new_chore)
 110                 session.commit()
 111             return f'Created chore {name}.'
 112         except:
 113             return f'Unable to create {name}. May be duplicate.'
 114
 115
 116
 117     def done(self, chore_id):
 118         with Session(self.engine) as session:
 119             chore = get_chore(session, chore_id)
 120             if(not chore):
 121                 return f'Chore not found: {chore_id}'
 122
 123             try:
 124                 session.add(Record(chore_id=chore.id))
 125                 session.commit()
 126                 return f'Marked chore as done: {chore.name}.'
 127             except Exception as ex:
 128                 return f'Unable to mark chore as done: {ex}'
 129
 130
 131     def undo(self, chore_id):
 132         with Session(self.engine) as session:
 133             cmd = select(Chore).where(Chore.id == chore_id)
 134             chore = session.exec(cmd).first()
 135
 136             if(not chore):
 137                 return f'Not found: {chore_id}'
 138
 139             try:
 140                 cmd = select(Record).where(Record.chore_id == chore_id).where(Record.complete_date >= today())
 141                 results = session.exec(cmd).all()
 142                 for result in results:
 143                     session.delete(result)
 144                 session.commit()
 145                 return f'Marked chore as *not* done: {chore.name}.'
 146             except Exception as ex:
 147                 return f'Unable to mark chore as done: {ex}'
 148
 149     def remove(self, chore_id):
 150         try:
 151             with Session(self.engine) as session:
 152                 chore = get_chore(session, chore_id)
 153                 if(not chore):
 154                     return f'Chore not found: {chore_id}'
 155
 156                 sql = text('''
 157                 DELETE FROM Record
 158                 WHERE chore_id = :chore_id
 159                ''')
 160                 params = {}
 161                 params['chore_id'] = chore_id
 162                 results = session.execute(sql, params)
 163                 logger.debug(results)
 164                 sql = text('''
 165                 DELETE FROM Chore
 166                 WHERE id = :chore_id
 167                ''')
 168                 results = session.execute(sql, params)
 169                 logger.debug(results)
 170                 session.commit()
 171                 return f'Removed chore {chore_id}'
 172         except Exception as ex:
 173             return f'Unable to remove chore: {ex}'