matrix_fridge_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
   7
   8 ## Todo List
   9 # - [ ] add to fridge
  10 # - [ ] reminder to eat recent food
  11 # - [ ] reminder to toss old food
  12
  13 DATE_FORMAT = '%A, %Y-%m-%d'
  14 MODULE='fridge'
  15 STALE_DAYS = 2
  16 TOSS_DAYS = 4
  17
  18 def today():
  19     return dt.now().strftime(DATE_FORMAT)
  20
  21 logger = logging.getLogger(__name__)
  22 logging.basicConfig(filename=f'matrix-{MODULE}-bot.log', encoding='utf-8', level=logging.DEBUG)
  23
  24 class Food(SQLModel, table=True):
  25     id: int | None = Field(default=None, primary_key=True)
  26     name: str = Field(default=None, unique=True)
  27     added_date: dt = Field(default_factory=lambda: dt.now())
  28
  29     def __str__(self):
  30         added = self.added_date.strftime(DATE_FORMAT)
  31         return f'{self.name} ({added})'
  32
  33 class Data():
  34     def __init__(self, db_file):
  35
  36         self.file = db_file
  37         self.engine = self.get_engine()
  38
  39     def get_db_file(self):
  40         return self.file
  41
  42     def get_engine(self):
  43         return create_engine(f"sqlite:///{self.file}")
  44
  45     def setup(self):
  46         SQLModel.metadata.create_all(self.engine)
  47
  48     @staticmethod
  49     def get_stale():
  50         stale = dt.now() - timedelta(STALE_DAYS)
  51         logger.debug(f'Stale date is {stale}')
  52         return stale
  53
  54     def list(self):
  55         results = []
  56         with Session(self.engine) as session:
  57             stale_dt = Data.get_stale()
  58             cmd = select(Food).where(Food.added_date > stale_dt)
  59             results = session.exec(cmd).all()
  60             logger.debug(f'List restuls are {str(results)}')
  61             if(not results):
  62                 results.append('Nothing Found')
  63
  64         return results
  65
  66     def to_toss(self):
  67         results = []
  68         with Session(self.engine) as session:
  69             stale_dt = Data.get_stale()
  70             cmd = select(Food).where(Food.added_date <= stale_dt)
  71             results = session.exec(cmd).all()
  72             if(not results):
  73                 results.append('Nothing Found')
  74
  75         return results
  76
  77
  78     def filter(self, term):
  79         results = []
  80         for item in self.list():
  81             if term in item:
  82                 results.append(item)
  83
  84         return results
  85
  86     def add(self, name):
  87         try:
  88             with Session(self.engine) as session:
  89                 new = Food(name=name)
  90                 session.add(new)
  91                 session.commit()
  92             return f'Added {name}.'
  93         except:
  94             return f'Unable to add {name}. May be duplicate.'
  95
  96     @staticmethod
  97     def get_food(session, food):
  98         cmd = select(Food).filter(Food.name.ilike(f'%{food}%'))
  99         return session.exec(cmd).one()
 100
 101     def remove(self, name):
 102         # TODO: Tag as tossed or eaten
 103         with Session(self.engine) as session:
 104             item  = Data.get_food(session, name)
 105             if(not item):
 106                 return f'Not found: {name}'
 107
 108             try:
 109                 name = item.name
 110                 session.remove(item)
 111                 session.commit()
 112                 return f'Removed item {name}'
 113             except Exception as ex:
 114                 return f'Unable to remove item mark chore as done: {ex}'
 115
 116