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