import logging
import os
from datetime import datetime as dt
+from decimal import Decimal
+from typing import List
from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
logger = logging.getLogger(__name__)
logging.basicConfig(filename=f'{MODULE}.log', encoding='utf-8', level=logging.DEBUG)
+class ItemStoreLink(SQLModel, table=True):
+ item_id: int | None = Field(default=None, foreign_key="item.id", primary_key=True)
+ store_id: int | None = Field(default=None, foreign_key="store.id", primary_key=True)
+
+
class Item(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
price: Decimal | None = Field(decimal_places=2)
- stores: list["Store"] = Relationship(back_populates="shopping_list")
+ stores: list["Store"] = Relationship(back_populates="shopping_list", link_model=ItemStoreLink)
purchase_date: dt | None = Field(default=None)
def __str__(self):
- return f'- [ ] {self.name} (#{self.id})'
+ return f'{self.name} (#{self.id})'
class Store(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
- shopping_list: list["Item"] = Relationship(back_populates="stores")
- purchase_history: list["PurchaseRecord"] = Relationship(back_populates="store")
+ shopping_list: list["Item"] = Relationship(back_populates="stores", link_model=ItemStoreLink)
+ # purchase_history: list["PurchaseRecord"] = Relationship(back_populates="store")
def __str__(self):
return f'{self.name} (#{self.id})'
-# May get used later...
-class PurchaseRecord(SQLModel, table=True):
- id: int | None = Field(default=None, primary_key=True)
- name: str = Field(default=None, unique=True)
- price: Decimal | None = Field(decimal_places=2)
- store: Store = Relationship(back_populates="purchase_history")
- purchase_date: dt | None = Field(default=None)
+# May get used later...
+#class PurchaseRecord(SQLModel, table=True):
+# id: int | None = Field(default=None, primary_key=True)
+# name: str = Field(default=None, unique=True)
+# price: Decimal | None = Field(decimal_places=2)
+# store: Store = Relationship(back_populates="purchase_history")
+# purchase_date: dt | None = Field(default=None)
+
+def _get_with_id(cls, item_id):
+ return select(cls).where(cls.id == item_id)
+
class Data():
def __init__(self, db_file=None):
if not db_file:
- db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), 'goals.db')
+ db_file = os.path.join(os.environ.get('MATRIX_DATA_DIR', ''), f'{MODULE}.db')
self.file = db_file
self.engine = self.get_engine()
id = new.id
return id
- def _get_with_id(self, cls, item_id):
- with Session(self.engine) as session:
- cmd = select(cls).where(cls.id == item_id)
- return session.exec(cmd).first()
-
def add_store(self, name):
- return self.add_with_name(Store, name)
+ return self._add_with_name(Store, name)
def add_item(self, name):
- return self.add_with_name(Item, name)
+ return self._add_with_name(Item, name)
def assign_store(self, store_id, item_id):
with Session(self.engine) as session:
- store = self.get_with_id(Store, store_id)
- item = self.get_with_id(Item, item_id)
+ store = session.exec(_get_with_id(Store, store_id)).first()
+ item = session.exec(_get_with_id(Item, item_id)).first()
item.stores.append(store)
session.commit()
def purchase_item(self, item_id):
with Session(self.engine) as session:
- item = self.get_with_id(Item, item_id)
+ item = session.exec(_get_with_id(Item, item_id)).first()
item.purchase_date = dt.today()
session.commit()
def clear_list(self, store_id, clear_all=False):
with Session(self.engine) as session:
- store = self.get_with_id(Store, store_id)
- for item in store.items:
- if clear_all or item.purchased_date:
- item.remove()
+ store = session.exec(_get_with_id(Store, store_id)).first()
+ for item in store.shopping_list:
+ if clear_all or item.purchase_date:
+ session.delete(item)
session.commit()
+ def get_stores(self):
+ with Session(self.engine) as session:
+ cmd = select(Store)
+ return session.exec(cmd).all()
+
def get_unsorted_list(self):
results = []
with Session(self.engine) as session:
- cmd = select(Item).where(Item.store == None)
+ cmd = select(Item).where(Item.stores == None)
items = session.exec(cmd).all()
if not items:
return [f'No unsorted items!']
return items
- def get_shopping_list(self, store_id) -> List(str):
+ def get_shopping_list(self, store_id) -> List[str]:
results = []
with Session(self.engine) as session:
cmd = select(Store).where(Store.id == store_id)
items = store.shopping_list
if not items:
- results.append('-- Nothing to shop for at {store.name} --')
-
- for item in items:
- if(item.purchase_date):
- results.append(f' - [x] {item}')
- else:
- results.append(f' - [ ] {item}')
+ results.append(f'-- Nothing to shop for at {store.name} --')
+ else:
+ for item in items:
+ if(item.purchase_date):
+ results.append(f' - [x] {item}')
+ else:
+ results.append(f' - [ ] {item}')
return results
+ def remove_item(self, item_id):
+ with Session(self.engine) as session:
+ item = session.exec(_get_with_id(Item, item_id)).first()
+ session.delete(item)
+ session.commit()
+
+ def remove_store(self, store_id):
+ with Session(self.engine) as session:
+ store = session.exec(_get_with_id(Store, store_id)).first()
+ session.delete(store)
+ session.commit()
+
def get_engine(self):
return create_engine(f"sqlite:///{self.file}")
projects / matrix-shop-bot.git / blobdiff