WIP

.gitignore

@@ -7,5 +7,3 @@ dist
 test.db
 
 .ruff_cache
-
-.coverage

dibbler/db.py

@@ -1,7 +1,16 @@
+from pathlib import Path
+
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
 from dibbler.conf import config
 
-engine = create_engine(config.get("database", "url"))
+if (url := config.get("database", "url")) is not None:
+  database_url = url
+
+elif (url_file := config.get("database", "url_file")) is not None:
+  with Path(url_file).open() as file:
+    database_url = file.read().strip()
+
+engine = create_engine(database_url)
 Session = sessionmaker(bind=engine)

dibbler/lib/helpers.py

@@ -1,7 +1,79 @@
-import os
 import pwd
-import signal
 import subprocess
+import os
+import signal
+
+from sqlalchemy import or_, and_
+
+from ..models import User, Product
+
+
+def search_user(string, session, ignorethisflag=None):
+    string = string.lower()
+    exact_match = (
+        session.query(User)
+        .filter(or_(User.name == string, User.card == string, User.rfid == string))
+        .first()
+    )
+    if exact_match:
+        return exact_match
+    user_list = (
+        session.query(User)
+        .filter(
+            or_(
+                User.name.ilike(f"%{string}%"),
+                User.card.ilike(f"%{string}%"),
+                User.rfid.ilike(f"%{string}%"),
+            )
+        )
+        .all()
+    )
+    return user_list
+
+
+def search_product(string, session, find_hidden_products=True):
+    if find_hidden_products:
+        exact_match = (
+            session.query(Product)
+            .filter(or_(Product.bar_code == string, Product.name == string))
+            .first()
+        )
+    else:
+        exact_match = (
+            session.query(Product)
+            .filter(
+                or_(
+                    Product.bar_code == string,
+                    and_(Product.name == string, Product.hidden is False),
+                )
+            )
+            .first()
+        )
+    if exact_match:
+        return exact_match
+    if find_hidden_products:
+        product_list = (
+            session.query(Product)
+            .filter(
+                or_(
+                    Product.bar_code.ilike(f"%{string}%"),
+                    Product.name.ilike(f"%{string}%"),
+                )
+            )
+            .all()
+        )
+    else:
+        product_list = (
+            session.query(Product)
+            .filter(
+                or_(
+                    Product.bar_code.ilike(f"%{string}%"),
+                    and_(Product.name.ilike(f"%{string}%"), Product.hidden is False),
+                )
+            )
+            .all()
+        )
+    return product_list
 
 
 def system_user_exists(username):

dibbler/main.py

@@ -1,4 +1,8 @@
 import argparse
+from pathlib import Path
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
 
 from dibbler.conf import config
 
@@ -23,29 +27,62 @@ subparsers.add_parser("slabbedasker", help="Find out who is slabbedasker")
 subparsers.add_parser("seed-data", help="Fill with mock data")
 
 
+def _get_database_url_from_config() -> str:
+    """Get the database URL from the configuration."""
+    url = config.get("database", "url")
+    if url is not None:
+        return url
+
+    url_file = config.get("database", "url_file")
+    if url_file is not None:
+        with Path(url_file).open() as file:
+            return file.read().strip()
+
+    raise ValueError("No database URL found in configuration.")
+
+
+def _connect_to_database(url: str, **engine_args) -> Session:
+    try:
+        engine = create_engine(url, **engine_args)
+        sql_session = Session(engine)
+    except Exception as err:
+        print("Error: could not connect to database.")
+        print(err)
+        exit(1)
+
+    print(f"Debug: Connected to database at '{url}'")
+    return sql_session
+
+
 def main():
     args = parser.parse_args()
     config.read(args.config)
 
+    database_url = _get_database_url_from_config()
+    sql_session = _connect_to_database(
+      database_url,
+      echo=config.getboolean("database", "echo_sql", fallback=False),
+    )
+
     if args.subcommand == "loop":
         import dibbler.subcommands.loop as loop
 
-        loop.main()
+        loop.main(sql_session)
 
     elif args.subcommand == "create-db":
         import dibbler.subcommands.makedb as makedb
 
-        makedb.main()
+        makedb.main(sql_session)
 
     elif args.subcommand == "slabbedasker":
         import dibbler.subcommands.slabbedasker as slabbedasker
 
-        slabbedasker.main()
+        slabbedasker.main(sql_session)
 
     elif args.subcommand == "seed-data":
         import dibbler.subcommands.seed_test_data as seed_test_data
 
-        seed_test_data.main()
+        seed_test_data.main(sql_session)
 
 
 if __name__ == "__main__":

dibbler/menus/main.py

@@ -0,0 +1,176 @@
+import sys
+import signal
+import traceback
+
+from sqlalchemy import (
+    event,
+)
+from sqlalchemy.orm import Session
+
+from libdib.repl import (
+    NumberedCmd,
+    InteractiveItemSelector,
+    prompt_yes_no,
+)
+
+from dibbler.conf import config
+
+class DibblerCli(NumberedCmd):
+    def __init__(self, sql_session: Session):
+        super().__init__()
+        self.sql_session = sql_session
+        self.sql_session_dirty = False
+
+        @event.listens_for(self.sql_session, "after_flush")
+        def mark_session_as_dirty(*_):
+            self.sql_session_dirty = True
+            self.prompt_header = "(unsaved changes)"
+
+        @event.listens_for(self.sql_session, "after_commit")
+        @event.listens_for(self.sql_session, "after_rollback")
+        def mark_session_as_clean(*_):
+            self.sql_session_dirty = False
+            self.prompt_header = None
+
+    # TODO: move to libdib.repl
+    @classmethod
+    def run_with_safe_exit_wrapper(cls, sql_session: Session):
+        tool = cls(sql_session)
+
+        if not config.getboolean("general", "stop_allowed"):
+            signal.signal(signal.SIGQUIT, signal.SIG_IGN)
+
+        if not config.getboolean("general", "stop_allowed"):
+            signal.signal(signal.SIGTSTP, signal.SIG_IGN)
+
+        while True:
+            try:
+                tool.cmdloop()
+            except KeyboardInterrupt:
+                if not tool.sql_session_dirty:
+                    exit(0)
+                try:
+                    print()
+                    if prompt_yes_no(
+                        "Are you sure you want to exit without saving?", default=False
+                    ):
+                        raise KeyboardInterrupt
+                except KeyboardInterrupt:
+                    if tool.sql_session is not None:
+                        tool.sql_session.rollback()
+                    exit(0)
+            except Exception:
+                print("Something went wrong.")
+                print(f"{sys.exc_info()[0]}: {sys.exc_info()[1]}")
+                if config.getboolean("general", "show_tracebacks"):
+                    traceback.print_tb(sys.exc_info()[2])
+
+    def default(self, maybe_barcode: str):
+        raise NotImplementedError(
+            "This command is not implemented yet. Please use the numbered commands instead."
+        )
+
+    def do_buy(self, arg: str):
+        ...
+
+    def do_product_list(self, arg: str):
+        ...
+
+    def do_show_user(self, arg: str):
+        ...
+
+    def do_user_list(self, arg: str):
+        ...
+
+    def do_adjust_credit(self, arg: str):
+        ...
+
+    def do_transfer(self, arg: str):
+        ...
+
+    def do_add_stock(self, arg: str):
+        ...
+
+    def do_add_edit(self, arg: str):
+        ...
+        # AddEditMenu(self.sql_session).cmdloop()
+
+    def do_product_search(self, arg: str):
+        ...
+
+    def do_statistics(self, arg: str):
+        ...
+
+    def do_faq(self, arg: str):
+        ...
+
+    def do_print_label(self, arg: str):
+        ...
+
+
+    def do_exit(self, _: str):
+        if self.sql_session_dirty:
+            if prompt_yes_no("Would you like to save your changes?"):
+                self.sql_session.commit()
+            else:
+                self.sql_session.rollback()
+        exit(0)
+
+    funcs = {
+        0: {
+            "f": default,
+            "doc": "Choose / Add item with its ISBN",
+        },
+        1: {
+            "f": do_buy,
+            "doc": "Buy",
+        },
+        2: {
+            "f": do_product_list,
+            "doc": "Product List",
+        },
+        3: {
+            "f": do_show_user,
+            "doc": "Show User",
+        },
+        4: {
+            "f": do_user_list,
+            "doc": "User List",
+        },
+        5: {
+            "f": do_adjust_credit,
+            "doc": "Adjust Credit",
+        },
+        6: {
+            "f": do_transfer,
+            "doc": "Transfer",
+        },
+        7: {
+            "f": do_add_stock,
+            "doc": "Add Stock",
+        },
+        8: {
+            "f": do_add_edit,
+            "doc": "Add/Edit",
+        },
+        9: {
+            "f": do_product_search,
+            "doc": "Product Search",
+        },
+        10: {
+            "f": do_statistics,
+            "doc": "Statistics",
+        },
+        11: {
+            "f": do_faq,
+            "doc": "FAQ",
+        },
+        12: {
+            "f": do_print_label,
+            "doc": "Print Label",
+        },
+        13: {
+            "f": do_exit,
+            "doc": "Exit",
+        },
+    }

dibbler/models/Base.py

@@ -10,16 +10,12 @@ from sqlalchemy.orm.collections import (
 )
 
 
-def _pascal_case_to_snake_case(name: str) -> str:
-    return "".join(["_" + i.lower() if i.isupper() else i for i in name]).lstrip("_")
-
-
 class Base(DeclarativeBase):
     metadata = MetaData(
         naming_convention={
-            "ix": "ix_%(table_name)s_%(column_0_label)s",
+            "ix": "ix_%(column_0_label)s",
             "uq": "uq_%(table_name)s_%(column_0_name)s",
-            "ck": "ck_%(table_name)s_%(constraint_name)s",
+            "ck": "ck_%(table_name)s_`%(constraint_name)s`",
             "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
             "pk": "pk_%(table_name)s",
         }
@@ -27,7 +23,7 @@ class Base(DeclarativeBase):
 
     @declared_attr.directive
     def __tablename__(cls) -> str:
-        return _pascal_case_to_snake_case(cls.__name__)
+        return cls.__name__
 
     def __repr__(self) -> str:
         columns = ", ".join(

dibbler/models/Product.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
-
+from typing import TYPE_CHECKING
-from typing import Self
 
 from sqlalchemy import (
     Boolean,
@@ -10,44 +9,39 @@ from sqlalchemy import (
 from sqlalchemy.orm import (
     Mapped,
     mapped_column,
+    relationship,
 )
 
 from .Base import Base
 
+if TYPE_CHECKING:
+    from .PurchaseEntry import PurchaseEntry
+    from .UserProducts import UserProducts
+
 
 class Product(Base):
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    __tablename__ = "products"
-    """Internal database ID"""
 
-    bar_code: Mapped[str] = mapped_column(String(13), unique=True)
+    product_id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    """
+    bar_code: Mapped[str] = mapped_column(String(13))
-        The bar code of the product.
+    name: Mapped[str] = mapped_column(String(45))
+    price: Mapped[int] = mapped_column(Integer)
+    stock: Mapped[int] = mapped_column(Integer)
+    hidden: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
 
-        This is a unique identifier for the product, typically a 13-digit
+    purchases: Mapped[set[PurchaseEntry]] = relationship(back_populates="product")
-        EAN-13 code.
+    users: Mapped[set[UserProducts]] = relationship(back_populates="product")
-    """
 
-    name: Mapped[str] = mapped_column(String(45), unique=True)
+    bar_code_re = r"[0-9]+"
-    """
+    name_re = r".+"
-        The name of the product.
+    name_length = 45
 
-        Please don't write fanfics here, this is not a place for that.
+    def __init__(self, bar_code, name, price, stock=0, hidden=False):
-    """
-
-    hidden: Mapped[bool] = mapped_column(Boolean, default=False)
-    """
-        Whether the product is hidden from the user interface.
-
-        Hidden products are not shown in the product list, but can still be
-        used in transactions.
-    """
-
-    def __init__(
-        self: Self,
-        bar_code: str,
-        name: str,
-        hidden: bool = False,
-    ) -> None:
-        self.bar_code = bar_code
         self.name = name
+        self.bar_code = bar_code
+        self.price = price
+        self.stock = stock
         self.hidden = hidden
+
+    def __str__(self):
+        return self.name

dibbler/models/ProductCache.py

@@ -1,15 +0,0 @@
-from datetime import datetime
-
-from sqlalchemy import Integer, DateTime
-from sqlalchemy.orm import Mapped, mapped_column
-
-from dibbler.models import Base
-
-class ProductCache(Base):
-    product_id: Mapped[int] = mapped_column(Integer, primary_key=True)
-
-    price: Mapped[int] = mapped_column(Integer)
-    price_timestamp: Mapped[datetime] = mapped_column(DateTime)
-
-    stock: Mapped[int] = mapped_column(Integer)
-    stock_timestamp: Mapped[datetime] = mapped_column(DateTime)

dibbler/models/Purchase.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from datetime import datetime
+import math
+
+from sqlalchemy import (
+    DateTime,
+    Integer,
+)
+from sqlalchemy.orm import (
+    Mapped,
+    mapped_column,
+    relationship,
+)
+
+from .Base import Base
+from .Transaction import Transaction
+
+if TYPE_CHECKING:
+    from .PurchaseEntry import PurchaseEntry
+
+
+class Purchase(Base):
+    __tablename__ = "purchases"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    time: Mapped[datetime] = mapped_column(DateTime)
+    price: Mapped[int] = mapped_column(Integer)
+
+    transactions: Mapped[set[Transaction]] = relationship(
+        back_populates="purchase", order_by="Transaction.user_name"
+    )
+    entries: Mapped[set[PurchaseEntry]] = relationship(back_populates="purchase")
+
+    def __init__(self):
+        pass
+
+    def is_complete(self):
+        return len(self.transactions) > 0 and len(self.entries) > 0
+
+    def price_per_transaction(self, round_up=True):
+        if round_up:
+            return int(math.ceil(float(self.price) / len(self.transactions)))
+        else:
+            return int(math.floor(float(self.price) / len(self.transactions)))
+
+    def set_price(self, round_up=True):
+        self.price = 0
+        for entry in self.entries:
+            self.price += entry.amount * entry.product.price
+        if len(self.transactions) > 0:
+            for t in self.transactions:
+                t.amount = self.price_per_transaction(round_up=round_up)
+
+    def perform_purchase(self, ignore_penalty=False, round_up=True):
+        self.time = datetime.datetime.now()
+        self.set_price(round_up=round_up)
+        for t in self.transactions:
+            t.perform_transaction(ignore_penalty=ignore_penalty)
+        for entry in self.entries:
+            entry.product.stock -= entry.amount
+
+    def perform_soft_purchase(self, price, round_up=True):
+        self.time = datetime.datetime.now()
+        self.price = price
+        for t in self.transactions:
+            t.amount = self.price_per_transaction(round_up=round_up)
+        for t in self.transactions:
+            t.perform_transaction()

dibbler/models/PurchaseEntry.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from sqlalchemy import (
+    Integer,
+    ForeignKey,
+)
+from sqlalchemy.orm import (
+    Mapped,
+    mapped_column,
+    relationship,
+)
+
+from .Base import Base
+
+if TYPE_CHECKING:
+    from .Product import Product
+    from .Purchase import Purchase
+
+
+class PurchaseEntry(Base):
+    __tablename__ = "purchase_entries"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    amount: Mapped[int] = mapped_column(Integer)
+
+    product_id: Mapped[int] = mapped_column(ForeignKey("products.product_id"))
+    purchase_id: Mapped[int] = mapped_column(ForeignKey("purchases.id"))
+
+    product: Mapped[Product] = relationship(lazy="joined")
+    purchase: Mapped[Purchase] = relationship(lazy="joined")
+
+    def __init__(self, purchase, product, amount):
+        self.product = product
+        self.product_bar_code = product.bar_code
+        self.purchase = purchase
+        self.amount = amount

dibbler/models/Transaction.py

@@ -1,463 +1,52 @@
 from __future__ import annotations
+from typing import TYPE_CHECKING
 
 from datetime import datetime
-from typing import TYPE_CHECKING, Self
 
 from sqlalchemy import (
-    CheckConstraint,
     DateTime,
     ForeignKey,
     Integer,
-    Text,
+    String,
-    and_,
-    column,
-    or_,
 )
 from sqlalchemy.orm import (
     Mapped,
     mapped_column,
     relationship,
 )
-from sqlalchemy.orm.collections import (
-    InstrumentedDict,
-    InstrumentedList,
-    InstrumentedSet,
-)
-from sqlalchemy.sql.schema import Index
 
 from .Base import Base
-from .TransactionType import TransactionType, TransactionTypeSQL
 
 if TYPE_CHECKING:
-    from .Product import Product
     from .User import User
-
+    from .Purchase import Purchase
-# TODO: rename to *_PERCENT
-
-# NOTE: these only matter when there are no adjustments made in the database.
-DEFAULT_INTEREST_RATE_PERCENTAGE = 100
-DEFAULT_PENALTY_THRESHOLD = -100
-DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE = 200
-
-# TODO: allow for joint transactions?
-#       dibbler allows joint transactions (e.g. buying more than one product at once, several people buying the same product, etc.)
-#       instead of having the software split the transactions up, making them hard to reconnect,
-#       maybe we should add some sort of joint transaction id field to allow multiple transactions to be grouped together?
-
-_DYNAMIC_FIELDS: set[str] = {
-    "amount",
-    "interest_rate_percent",
-    "penalty_multiplier_percent",
-    "penalty_threshold",
-    "per_product",
-    "product_count",
-    "product_id",
-    "transfer_user_id",
-}
-
-_EXPECTED_FIELDS: dict[TransactionType, set[str]] = {
-    TransactionType.ADD_PRODUCT: {"amount", "per_product", "product_count", "product_id"},
-    TransactionType.ADJUST_BALANCE: {"amount"},
-    TransactionType.ADJUST_INTEREST: {"interest_rate_percent"},
-    TransactionType.ADJUST_PENALTY: {"penalty_multiplier_percent", "penalty_threshold"},
-    TransactionType.ADJUST_STOCK: {"product_count", "product_id"},
-    TransactionType.BUY_PRODUCT: {"product_count", "product_id"},
-    TransactionType.TRANSFER: {"amount", "transfer_user_id"},
-}
-
-assert all(x <= _DYNAMIC_FIELDS for x in _EXPECTED_FIELDS.values()), (
-    "All expected fields must be part of _DYNAMIC_FIELDS."
-)
-
-
-def _transaction_type_field_constraints(
-    transaction_type: TransactionType,
-    expected_fields: set[str],
-) -> CheckConstraint:
-    unexpected_fields = _DYNAMIC_FIELDS - expected_fields
-
-    return CheckConstraint(
-        or_(
-            column("type") != transaction_type.value,
-            and_(
-                *[column(field) != None for field in expected_fields],
-                *[column(field) == None for field in unexpected_fields],
-            ),
-        ),
-        name=f"trx_type_{transaction_type.value}_expected_fields",
-    )
 
 
 class Transaction(Base):
-    __table_args__ = (
+    __tablename__ = "transactions"
-        *[
-            _transaction_type_field_constraints(transaction_type, expected_fields)
-            for transaction_type, expected_fields in _EXPECTED_FIELDS.items()
-        ],
-        CheckConstraint(
-            or_(
-                column("type") != TransactionType.TRANSFER.value,
-                column("user_id") != column("transfer_user_id"),
-            ),
-            name="trx_type_transfer_no_self_transfers",
-        ),
-        # Speed up product count calculation
-        Index("product_user_time", "product_id", "user_id", "time"),
-        # Speed up product owner calculation
-        Index("user_product_time", "user_id", "product_id", "time"),
-        # Speed up user transaction list / credit calculation
-        Index("user_time", "user_id", "time"),
-    )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True)
-    """
-        A unique identifier for the transaction.
 
-        Not used for anything else than identifying the transaction in the database.
+    time: Mapped[datetime] = mapped_column(DateTime)
-    """
+    amount: Mapped[int] = mapped_column(Integer)
+    penalty: Mapped[int] = mapped_column(Integer)
+    description: Mapped[str | None] = mapped_column(String(50))
 
-    time: Mapped[datetime] = mapped_column(DateTime, unique=True)
+    user_name: Mapped[str] = mapped_column(ForeignKey("users.name"))
-    """
+    purchase_id: Mapped[int | None] = mapped_column(ForeignKey("purchases.id"))
-        The time when the transaction took place.
 
-        This is used to order transactions chronologically, and to calculate
+    user: Mapped[User] = relationship(lazy="joined")
-        all kinds of state.
+    purchase: Mapped[Purchase] = relationship(lazy="joined")
-    """
 
-    message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    def __init__(self, user, amount=0, description=None, purchase=None, penalty=1):
-    """
+        self.user = user
-        A message that can be set by the user to describe the reason
-        behind the transaction (or potentially a place to write som fan fiction).
-
-        This is not used for any calculations, but can be useful for debugging.
-    """
-
-    type_: Mapped[TransactionType] = mapped_column(TransactionTypeSQL, name="type")
-    """
-        Which type of transaction this is.
-
-        The type determines which fields are expected to be set.
-    """
-
-    amount: Mapped[int | None] = mapped_column(Integer)
-    """
-        This field means different things depending on the transaction type:
-
-         - `ADD_PRODUCT`: The real amount spent on the products.
-
-         - `ADJUST_BALANCE`: The amount of credit to add or subtract from the user's balance.
-
-         - `BUY_PRODUCT`: The amount of credit spent on the product.
-                         Note that this includes any penalties and interest that the user
-                         had to pay as well.
-
-         - `TRANSFER`: The amount of balance to transfer to another user.
-    """
-
-    per_product: Mapped[int | None] = mapped_column(Integer)
-    """
-        If adding products, how much is each product worth
-
-        Note that this is distinct from the total amount of the transaction,
-        because this gets rounded up to the nearest integer, while the total amount
-        that the user paid in the store would be stored in the `amount` field.
-    """
-
-    user_id: Mapped[int] = mapped_column(ForeignKey("user.id"))
-    """The user who performs the transaction. See `user` for more details."""
-    user: Mapped[User] = relationship(
-        lazy="joined",
-        foreign_keys=[user_id],
-    )
-    """
-        The user who performs the transaction.
-
-        For some transaction types, like `TRANSFER` and `ADD_PRODUCT`, this is a
-        functional field with "real world consequences" for price calculations.
-
-        For others, like `ADJUST_PENALTY` and `ADJUST_STOCK`, this is just a record of who
-        performed the transaction, and does not affect any state calculations.
-    """
-
-    # Receiving user when moving credit from one user to another
-    transfer_user_id: Mapped[int | None] = mapped_column(ForeignKey("user.id"))
-    """The user who receives money in a `TRANSFER` transaction."""
-    transfer_user: Mapped[User | None] = relationship(
-        lazy="joined",
-        foreign_keys=[transfer_user_id],
-    )
-    """The user who receives money in a `TRANSFER` transaction."""
-
-    # The product that is either being added or bought
-    product_id: Mapped[int | None] = mapped_column(ForeignKey("product.id"))
-    """The product being added or bought."""
-    product: Mapped[Product | None] = relationship(lazy="joined")
-    """The product being added or bought."""
-
-    # The amount of products being added or bought
-    product_count: Mapped[int | None] = mapped_column(Integer)
-    """
-        The amount of products being added or bought.
-    """
-
-    penalty_threshold: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    """
-        On `ADJUST_PENALTY` transactions, this is the threshold in krs for when the user
-        should start getting penalized for low credit.
-
-        See also `penalty_multiplier`.
-    """
-
-    penalty_multiplier_percent: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    """
-        On `ADJUST_PENALTY` transactions, this is the multiplier for the amount of
-        money the user has to pay when they have too low credit.
-
-        The multiplier is a percentage, so `100` means the user has to pay the full
-        price of the product, `200` means they have to pay double, etc.
-
-        See also `penalty_threshold`.
-    """
-
-    # TODO: this should be inferred
-    # Assuming this is a BUY_PRODUCT transaction, was the user penalized for having
-    # too low credit in this transaction?
-    # is_penalized: Mapped[Boolean] = mapped_column(Boolean, default=False)
-
-    interest_rate_percent: Mapped[int | None] = mapped_column(Integer, nullable=True)
-    """
-        On `ADJUST_INTEREST` transactions, this is the interest rate in percent
-        that the user has to pay on their balance.
-
-        The interest rate is a percentage, so `100` means the user has to pay the full
-        price of the product, `200` means they have to pay double, etc.
-    """
-
-    def __init__(
-        self: Self,
-        type_: TransactionType,
-        user_id: int,
-        amount: int | None = None,
-        time: datetime | None = None,
-        message: str | None = None,
-        product_id: int | None = None,
-        transfer_user_id: int | None = None,
-        per_product: int | None = None,
-        product_count: int | None = None,
-        penalty_threshold: int | None = None,
-        penalty_multiplier_percent: int | None = None,
-        interest_rate_percent: int | None = None,
-    ) -> None:
-        """
-        Please do not call this constructor directly, use the factory methods instead.
-        """
-        if time is None:
-            time = datetime.now()
-
-        self.time = time
-        self.message = message
-        self.type_ = type_
         self.amount = amount
-        self.user_id = user_id
+        self.description = description
-        self.product_id = product_id
+        self.purchase = purchase
-        self.transfer_user_id = transfer_user_id
+        self.penalty = penalty
-        self.per_product = per_product
-        self.product_count = product_count
-        self.penalty_threshold = penalty_threshold
-        self.penalty_multiplier_percent = penalty_multiplier_percent
-        self.interest_rate_percent = interest_rate_percent
 
-        self._validate_by_transaction_type()
+    def perform_transaction(self, ignore_penalty=False):
-
+        self.time = datetime.datetime.now()
-    def _validate_by_transaction_type(self: Self) -> None:
+        if not ignore_penalty:
-        """
+            self.amount *= self.penalty
-        Validates the transaction's fields based on its type.
+        self.user.credit -= self.amount
-        Raises `ValueError` if the transaction is invalid.
-        """
-        # TODO: do we allow free products?
-        if self.amount == 0:
-            raise ValueError("Amount must not be zero.")
-
-        for field in _EXPECTED_FIELDS[self.type_]:
-            if getattr(self, field) is None:
-                raise ValueError(f"{field} must not be None for {self.type_.value} transactions.")
-
-        for field in _DYNAMIC_FIELDS - _EXPECTED_FIELDS[self.type_]:
-            if getattr(self, field) is not None:
-                raise ValueError(f"{field} must be None for {self.type_.value} transactions.")
-
-        if self.per_product is not None and self.per_product <= 0:
-            raise ValueError("per_product must be greater than zero.")
-
-        if (
-            self.per_product is not None
-            and self.product_count is not None
-            and self.amount is not None
-            and self.amount > self.per_product * self.product_count
-        ):
-            raise ValueError(
-                "The real amount of the transaction must be less than the total value of the products."
-            )
-
-    # TODO: improve printing further
-
-    def __repr__(self) -> str:
-        sort_order = [
-            "id",
-            "time",
-        ]
-
-        columns = ", ".join(
-            f"{k}={repr(v)}"
-            for k, v in sorted(
-                self.__dict__.items(),
-                key=lambda item: chr(sort_order.index(item[0]))
-                if item[0] in sort_order
-                else item[0],
-            )
-            if not any(
-                [
-                    k == "type_",
-                    (k == "message" and v is None),
-                    k.startswith("_"),
-                    # Ensure that we don't try to print out the entire list of
-                    # relationships, which could create an infinite loop
-                    isinstance(v, Base),
-                    isinstance(v, InstrumentedList),
-                    isinstance(v, InstrumentedSet),
-                    isinstance(v, InstrumentedDict),
-                    *[k in (_DYNAMIC_FIELDS - _EXPECTED_FIELDS[self.type_])],
-                ]
-            )
-        )
-        return f"{self.type_.upper()}({columns})"
-
-    ###################
-    # FACTORY METHODS #
-    ###################
-
-    @classmethod
-    def adjust_balance(
-        cls: type[Self],
-        amount: int,
-        user_id: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.ADJUST_BALANCE,
-            amount=amount,
-            user_id=user_id,
-            message=message,
-        )
-
-    @classmethod
-    def adjust_interest(
-        cls: type[Self],
-        interest_rate_percent: int,
-        user_id: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.ADJUST_INTEREST,
-            interest_rate_percent=interest_rate_percent,
-            user_id=user_id,
-            message=message,
-        )
-
-    @classmethod
-    def adjust_penalty(
-        cls: type[Self],
-        penalty_multiplier_percent: int,
-        penalty_threshold: int,
-        user_id: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.ADJUST_PENALTY,
-            penalty_multiplier_percent=penalty_multiplier_percent,
-            penalty_threshold=penalty_threshold,
-            user_id=user_id,
-            message=message,
-        )
-
-    @classmethod
-    def adjust_stock(
-        cls: type[Self],
-        user_id: int,
-        product_id: int,
-        product_count: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.ADJUST_STOCK,
-            user_id=user_id,
-            product_id=product_id,
-            product_count=product_count,
-            message=message,
-        )
-
-    @classmethod
-    def add_product(
-        cls: type[Self],
-        amount: int,
-        user_id: int,
-        product_id: int,
-        per_product: int,
-        product_count: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.ADD_PRODUCT,
-            amount=amount,
-            user_id=user_id,
-            product_id=product_id,
-            per_product=per_product,
-            product_count=product_count,
-            message=message,
-        )
-
-    @classmethod
-    def buy_product(
-        cls: type[Self],
-        user_id: int,
-        product_id: int,
-        product_count: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.BUY_PRODUCT,
-            user_id=user_id,
-            product_id=product_id,
-            product_count=product_count,
-            message=message,
-        )
-
-    @classmethod
-    def transfer(
-        cls: type[Self],
-        amount: int,
-        user_id: int,
-        transfer_user_id: int,
-        time: datetime | None = None,
-        message: str | None = None,
-    ) -> Transaction:
-        return cls(
-            time=time,
-            type_=TransactionType.TRANSFER,
-            amount=amount,
-            user_id=user_id,
-            transfer_user_id=transfer_user_id,
-            message=message,
-        )

dibbler/models/TransactionType.py

@@ -1,26 +0,0 @@
-from enum import StrEnum, auto
-
-from sqlalchemy import Enum as SQLEnum
-
-
-class TransactionType(StrEnum):
-    """
-    Enum for transaction types.
-    """
-
-    ADD_PRODUCT = auto()
-    ADJUST_BALANCE = auto()
-    ADJUST_INTEREST = auto()
-    ADJUST_PENALTY = auto()
-    ADJUST_STOCK = auto()
-    BUY_PRODUCT = auto()
-    TRANSFER = auto()
-
-
-TransactionTypeSQL = SQLEnum(
-    TransactionType,
-    native_enum=True,
-    create_constraint=True,
-    validate_strings=True,
-    values_callable=lambda x: [i.value for i in x],
-)

dibbler/models/User.py

@@ -1,47 +1,49 @@
 from __future__ import annotations
-
+from typing import TYPE_CHECKING
-from typing import TYPE_CHECKING, Self
 
 from sqlalchemy import (
     Integer,
     String,
-    select,
 )
 from sqlalchemy.orm import (
     Mapped,
-    Session,
     mapped_column,
+    relationship,
 )
 
 from .Base import Base
 
 if TYPE_CHECKING:
+    from .UserProducts import UserProducts
     from .Transaction import Transaction
 
 
 class User(Base):
-    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    __tablename__ = "users"
-    """Internal database ID"""
+    name: Mapped[str] = mapped_column(String(10), primary_key=True)
-
+    credit: Mapped[str] = mapped_column(Integer)
-    name: Mapped[str] = mapped_column(String(20), unique=True)
-    """
-        The PVV username of the user.
-    """
-
     card: Mapped[str | None] = mapped_column(String(20))
     rfid: Mapped[str | None] = mapped_column(String(20))
 
-    # name_re = r"[a-z]+"
+    products: Mapped[set[UserProducts]] = relationship(back_populates="user")
-    # card_re = r"(([Nn][Tt][Nn][Uu])?[0-9]+)?"
+    transactions: Mapped[set[Transaction]] = relationship(back_populates="user")
-    # rfid_re = r"[0-9a-fA-F]*"
 
-    def __init__(self: Self, name: str, card: str | None = None, rfid: str | None = None) -> None:
+    name_re = r"[a-z]+"
+    card_re = r"(([Nn][Tt][Nn][Uu])?[0-9]+)?"
+    rfid_re = r"[0-9a-fA-F]*"
+
+    def __init__(self, name, card, rfid=None, credit=0):
         self.name = name
+        if card == "":
+            card = None
         self.card = card
+        if rfid == "":
+            rfid = None
         self.rfid = rfid
+        self.credit = credit
 
-    # def __str__(self):
+    def __str__(self):
-    #     return self.name
+        return self.name
 
-    # def is_anonymous(self):
+    def is_anonymous(self):
-    #     return self.card == "11122233"
+        return self.card == "11122233"

dibbler/models/UserCache.py

@@ -1,13 +0,0 @@
-from datetime import datetime
-
-from sqlalchemy import Integer, DateTime
-from sqlalchemy.orm import Mapped, mapped_column
-
-from dibbler.models import Base
-
-# More like user balance cash money flow, amirite?
-class UserBalanceCache(Base):
-    user_id: Mapped[int] = mapped_column(Integer, primary_key=True)
-
-    balance: Mapped[int] = mapped_column(Integer)
-    timestamp: Mapped[datetime] = mapped_column(DateTime)

dibbler/models/UserProducts.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+
+from sqlalchemy import (
+    Integer,
+    ForeignKey,
+)
+from sqlalchemy.orm import (
+    Mapped,
+    mapped_column,
+    relationship,
+)
+
+from .Base import Base
+
+if TYPE_CHECKING:
+    from .User import User
+    from .Product import Product
+
+
+class UserProducts(Base):
+    __tablename__ = "user_products"
+
+    user_name: Mapped[str] = mapped_column(ForeignKey("users.name"), primary_key=True)
+    product_id: Mapped[int] = mapped_column(ForeignKey("products.product_id"), primary_key=True)
+
+    count: Mapped[int] = mapped_column(Integer)
+    sign: Mapped[int] = mapped_column(Integer)
+
+    user: Mapped[User] = relationship()
+    product: Mapped[Product] = relationship()

dibbler/models/__init__.py

@@ -1,12 +1,17 @@
 __all__ = [
     "Base",
     "Product",
+    "Purchase",
+    "PurchaseEntry",
     "Transaction",
     "User",
+    "UserProducts",
 ]
 
 from .Base import Base
 from .Product import Product
+from .Purchase import Purchase
+from .PurchaseEntry import PurchaseEntry
 from .Transaction import Transaction
-from .TransactionType import TransactionType
 from .User import User
+from .UserProducts import UserProducts

dibbler/queries/adjust_interest.py

@@ -1,2 +0,0 @@
-# NOTE: this type of transaction should be password protected.
-#       the password can be set as a string literal in the config file.

dibbler/queries/adjust_penalty.py

@@ -1,2 +0,0 @@
-# NOTE: this type of transaction should be password protected.
-#       the password can be set as a string literal in the config file.

dibbler/queries/current_interest.py

@@ -1,19 +0,0 @@
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from dibbler.models import Transaction, TransactionType
-from dibbler.models.Transaction import DEFAULT_INTEREST_RATE_PERCENTAGE
-
-
-def current_interest(sql_session: Session) -> int:
-    result = sql_session.scalars(
-        select(Transaction)
-        .where(Transaction.type_ == TransactionType.ADJUST_INTEREST)
-        .order_by(Transaction.time.desc())
-        .limit(1)
-    ).one_or_none()
-
-    if result is None:
-        return DEFAULT_INTEREST_RATE_PERCENTAGE
-
-    return result.interest_rate_percent

dibbler/queries/current_penalty.py

@@ -1,25 +0,0 @@
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from dibbler.models import Transaction, TransactionType
-from dibbler.models.Transaction import (
-    DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE,
-    DEFAULT_PENALTY_THRESHOLD,
-)
-
-
-def current_penalty(sql_session: Session) -> tuple[int, int]:
-    result = sql_session.scalars(
-        select(Transaction)
-        .where(Transaction.type_ == TransactionType.ADJUST_PENALTY)
-        .order_by(Transaction.time.desc())
-        .limit(1)
-    ).one_or_none()
-
-    if result is None:
-        return DEFAULT_PENALTY_THRESHOLD, DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE
-
-    assert result.penalty_threshold is not None, "Penalty threshold must be set"
-    assert result.penalty_multiplier_percent is not None, "Penalty multiplier percent must be set"
-
-    return result.penalty_threshold, result.penalty_multiplier_percent

dibbler/queries/product_price.py

@@ -1,245 +0,0 @@
-import math
-from dataclasses import dataclass
-from datetime import datetime
-
-from sqlalchemy import (
-    ColumnElement,
-    Integer,
-    SQLColumnExpression,
-    asc,
-    case,
-    cast,
-    func,
-    literal,
-    select,
-)
-from sqlalchemy.orm import Session
-
-from dibbler.models import (
-    Product,
-    Transaction,
-    TransactionType,
-)
-from dibbler.models.Transaction import DEFAULT_INTEREST_RATE_PERCENTAGE
-
-
-def _product_price_query(
-    product_id: int | ColumnElement[int],
-    use_cache: bool = True,
-    until: datetime | SQLColumnExpression[datetime] | None = None,
-    until_including: bool = True,
-    cte_name: str = "rec_cte",
-):
-    """
-    The inner query for calculating the product price.
-    """
-
-    if use_cache:
-        print("WARNING: Using cache for product price query is not implemented yet.")
-
-    initial_element = select(
-        literal(0).label("i"),
-        literal(0).label("time"),
-        literal(None).label("transaction_id"),
-        literal(0).label("price"),
-        literal(0).label("product_count"),
-    )
-
-    recursive_cte = initial_element.cte(name=cte_name, recursive=True)
-
-    # Subset of transactions that we'll want to iterate over.
-    trx_subset = (
-        select(
-            func.row_number().over(order_by=asc(Transaction.time)).label("i"),
-            Transaction.id,
-            Transaction.time,
-            Transaction.type_,
-            Transaction.product_count,
-            Transaction.per_product,
-        )
-        .where(
-            Transaction.type_.in_(
-                [
-                    TransactionType.BUY_PRODUCT,
-                    TransactionType.ADD_PRODUCT,
-                    TransactionType.ADJUST_STOCK,
-                ]
-            ),
-            Transaction.product_id == product_id,
-            case(
-                (literal(until_including), Transaction.time <= until),
-                else_=Transaction.time < until,
-            )
-            if until is not None
-            else literal(True),
-        )
-        .order_by(Transaction.time.asc())
-        .alias("trx_subset")
-    )
-
-    recursive_elements = (
-        select(
-            trx_subset.c.i,
-            trx_subset.c.time,
-            trx_subset.c.id.label("transaction_id"),
-            case(
-                # Someone buys the product -> price remains the same.
-                (trx_subset.c.type_ == TransactionType.BUY_PRODUCT, recursive_cte.c.price),
-                # Someone adds the product -> price is recalculated based on
-                #  product count, previous price, and new price.
-                (
-                    trx_subset.c.type_ == TransactionType.ADD_PRODUCT,
-                    cast(
-                        func.ceil(
-                            (
-                                recursive_cte.c.price * func.max(recursive_cte.c.product_count, 0)
-                                + trx_subset.c.per_product * trx_subset.c.product_count
-                            )
-                            / (
-                                # The running product count can be negative if the accounting is bad.
-                                # This ensures that we never end up with negative prices or zero divisions
-                                # and other disastrous phenomena.
-                                func.max(recursive_cte.c.product_count, 0)
-                                + trx_subset.c.product_count
-                            )
-                        ),
-                        Integer,
-                    ),
-                ),
-                # Someone adjusts the stock -> price remains the same.
-                (trx_subset.c.type_ == TransactionType.ADJUST_STOCK, recursive_cte.c.price),
-                # Should never happen
-                else_=recursive_cte.c.price,
-            ).label("price"),
-            case(
-                # Someone buys the product -> product count is reduced.
-                (
-                    trx_subset.c.type_ == TransactionType.BUY_PRODUCT,
-                    recursive_cte.c.product_count - trx_subset.c.product_count,
-                ),
-                # Someone adds the product -> product count is increased.
-                (
-                    trx_subset.c.type_ == TransactionType.ADD_PRODUCT,
-                    recursive_cte.c.product_count + trx_subset.c.product_count,
-                ),
-                # Someone adjusts the stock -> product count is adjusted.
-                (
-                    trx_subset.c.type_ == TransactionType.ADJUST_STOCK,
-                    recursive_cte.c.product_count + trx_subset.c.product_count,
-                ),
-                # Should never happen
-                else_=recursive_cte.c.product_count,
-            ).label("product_count"),
-        )
-        .select_from(trx_subset)
-        .where(trx_subset.c.i == recursive_cte.c.i + 1)
-    )
-
-    return recursive_cte.union_all(recursive_elements)
-
-
-# TODO: create a function for the log that pretty prints the log entries
-#       for debugging purposes
-
-
-@dataclass
-class ProductPriceLogEntry:
-    transaction: Transaction
-    price: int
-    product_count: int
-
-
-def product_price_log(
-    sql_session: Session,
-    product: Product,
-    use_cache: bool = True,
-    until: Transaction | None = None,
-) -> list[ProductPriceLogEntry]:
-    """
-    Calculates the price of a product and returns a log of the price changes.
-    """
-
-    recursive_cte = _product_price_query(
-        product.id,
-        use_cache=use_cache,
-        until=until.time if until else None,
-    )
-
-    result = sql_session.execute(
-        select(
-            Transaction,
-            recursive_cte.c.price,
-            recursive_cte.c.product_count,
-        )
-        .select_from(recursive_cte)
-        .join(
-            Transaction,
-            onclause=Transaction.id == recursive_cte.c.transaction_id,
-        )
-        .order_by(recursive_cte.c.i.asc())
-    ).all()
-
-    if result is None:
-        # If there are no transactions for this product, the query should return an empty list, not None.
-        raise RuntimeError(
-            f"Something went wrong while calculating the price log for product {product.name} (ID: {product.id})."
-        )
-
-    return [
-        ProductPriceLogEntry(
-            transaction=row[0],
-            price=row.price,
-            product_count=row.product_count,
-        )
-        for row in result
-    ]
-
-
-@staticmethod
-def product_price(
-    sql_session: Session,
-    product: Product,
-    use_cache: bool = True,
-    until: Transaction | None = None,
-    include_interest: bool = False,
-) -> int:
-    """
-    Calculates the price of a product.
-    """
-
-    recursive_cte = _product_price_query(
-        product.id,
-        use_cache=use_cache,
-        until=until.time if until else None,
-    )
-
-    # TODO: optionally verify subresults:
-    #   - product_count should never be negative (but this happens sometimes, so just a warning)
-    #   - price should never be negative
-
-    result = sql_session.scalars(
-        select(recursive_cte.c.price).order_by(recursive_cte.c.i.desc()).limit(1)
-    ).one_or_none()
-
-    if result is None:
-        # If there are no transactions for this product, the query should return 0, not None.
-        raise RuntimeError(
-            f"Something went wrong while calculating the price for product {product.name} (ID: {product.id})."
-        )
-
-    if include_interest:
-        interest_rate = (
-            sql_session.scalar(
-                select(Transaction.interest_rate_percent)
-                .where(
-                    Transaction.type_ == TransactionType.ADJUST_INTEREST,
-                    literal(True) if until is None else Transaction.time <= until.time,
-                )
-                .order_by(Transaction.time.desc())
-                .limit(1)
-            )
-            or DEFAULT_INTEREST_RATE_PERCENTAGE
-        )
-        result = math.ceil(result * interest_rate / 100)
-
-    return result

dibbler/queries/product_stock.py

@@ -1,76 +0,0 @@
-from datetime import datetime
-
-from sqlalchemy import case, func, literal, select
-from sqlalchemy.orm import Session
-
-from dibbler.models import (
-    Product,
-    Transaction,
-    TransactionType,
-)
-
-
-def _product_stock_query(
-    product_id: int,
-    use_cache: bool = True,
-    until: datetime | None = None,
-):
-    """
-    The inner query for calculating the product stock.
-    """
-
-    if use_cache:
-        print("WARNING: Using cache for product stock query is not implemented yet.")
-
-    query = select(
-        func.sum(
-            case(
-                (
-                    Transaction.type_ == TransactionType.ADD_PRODUCT,
-                    Transaction.product_count,
-                ),
-                (
-                    Transaction.type_ == TransactionType.BUY_PRODUCT,
-                    -Transaction.product_count,
-                ),
-                (
-                    Transaction.type_ == TransactionType.ADJUST_STOCK,
-                    Transaction.product_count,
-                ),
-                else_=0,
-            )
-        )
-    ).where(
-        Transaction.type_.in_(
-            [
-                TransactionType.BUY_PRODUCT,
-                TransactionType.ADD_PRODUCT,
-                TransactionType.ADJUST_STOCK,
-            ]
-        ),
-        Transaction.product_id == product_id,
-        Transaction.time <= until if until is not None else literal(True),
-    )
-
-    return query
-
-
-def product_stock(
-    sql_session: Session,
-    product: Product,
-    use_cache: bool = True,
-    until: datetime | None = None,
-) -> int:
-    """
-    Returns the number of products in stock.
-    """
-
-    query = _product_stock_query(
-        product_id=product.id,
-        use_cache=use_cache,
-        until=until,
-    )
-
-    result = sql_session.scalars(query).one_or_none()
-
-    return result or 0

dibbler/queries/search_product.py

@@ -1,39 +0,0 @@
-from sqlalchemy import and_, literal, or_, select
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product
-
-
-def search_product(
-    string: str,
-    sql_session: Session,
-    find_hidden_products=True,
-) -> Product | list[Product]:
-    exact_match = sql_session.scalars(
-        select(Product).where(
-            or_(
-                Product.bar_code == string,
-                and_(
-                    Product.name == string,
-                    literal(True) if find_hidden_products else not Product.hidden,
-                ),
-            )
-        )
-    ).first()
-
-    if exact_match:
-        return exact_match
-
-    product_list = sql_session.scalars(
-        select(Product).where(
-            or_(
-                Product.bar_code.ilike(f"%{string}%"),
-                and_(
-                    Product.name.ilike(f"%{string}%"),
-                    literal(True) if find_hidden_products else not Product.hidden,
-                ),
-            )
-        )
-    ).all()
-
-    return list(product_list)

dibbler/queries/search_user.py

@@ -1,37 +0,0 @@
-from sqlalchemy import or_, select
-from sqlalchemy.orm import Session
-
-from dibbler.models import User
-
-
-def search_user(
-    string: str,
-    sql_session: Session,
-    ignorethisflag=None,
-) -> User | list[User]:
-    string = string.lower()
-
-    exact_match = sql_session.scalars(
-        select(User).where(
-            or_(
-                User.name == string,
-                User.card == string,
-                User.rfid == string,
-            )
-        )
-    ).first()
-
-    if exact_match:
-        return exact_match
-
-    user_list = sql_session.scalars(
-        select(User).where(
-            or_(
-                User.name.ilike(f"%{string}%"),
-                User.card.ilike(f"%{string}%"),
-                User.rfid.ilike(f"%{string}%"),
-            )
-        )
-    ).all()
-
-    return list(user_list)

dibbler/queries/user_balance.py

@@ -1,319 +0,0 @@
-from dataclasses import dataclass
-from datetime import datetime
-
-from sqlalchemy import (
-    Float,
-    Integer,
-    and_,
-    asc,
-    case,
-    cast,
-    column,
-    func,
-    literal,
-    or_,
-    select,
-)
-from sqlalchemy.orm import Session
-
-from dibbler.models import (
-    Transaction,
-    TransactionType,
-    User,
-)
-from dibbler.models.Transaction import (
-    DEFAULT_INTEREST_RATE_PERCENTAGE,
-    DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE,
-    DEFAULT_PENALTY_THRESHOLD,
-)
-from dibbler.queries.product_price import _product_price_query
-
-
-def _user_balance_query(
-    user_id: int,
-    use_cache: bool = True,
-    until: datetime | None = None,
-    until_including: bool = True,
-    cte_name: str = "rec_cte",
-):
-    """
-    The inner query for calculating the user's balance.
-    """
-
-    if use_cache:
-        print("WARNING: Using cache for user balance query is not implemented yet.")
-
-    initial_element = select(
-        literal(0).label("i"),
-        literal(0).label("time"),
-        literal(None).label("transaction_id"),
-        literal(0).label("balance"),
-        literal(DEFAULT_INTEREST_RATE_PERCENTAGE).label("interest_rate_percent"),
-        literal(DEFAULT_PENALTY_THRESHOLD).label("penalty_threshold"),
-        literal(DEFAULT_PENALTY_MULTIPLIER_PERCENTAGE).label("penalty_multiplier_percent"),
-    )
-
-    recursive_cte = initial_element.cte(name=cte_name, recursive=True)
-
-    # Subset of transactions that we'll want to iterate over.
-    trx_subset = (
-        select(
-            func.row_number().over(order_by=asc(Transaction.time)).label("i"),
-            Transaction.amount,
-            Transaction.id,
-            Transaction.interest_rate_percent,
-            Transaction.penalty_multiplier_percent,
-            Transaction.penalty_threshold,
-            Transaction.product_count,
-            Transaction.product_id,
-            Transaction.time,
-            Transaction.transfer_user_id,
-            Transaction.type_,
-        )
-        .where(
-            or_(
-                and_(
-                    Transaction.user_id == user_id,
-                    Transaction.type_.in_(
-                        [
-                            TransactionType.ADD_PRODUCT,
-                            TransactionType.ADJUST_BALANCE,
-                            TransactionType.BUY_PRODUCT,
-                            TransactionType.TRANSFER,
-                        ]
-                    ),
-                ),
-                and_(
-                    Transaction.type_ == TransactionType.TRANSFER,
-                    Transaction.transfer_user_id == user_id,
-                ),
-                Transaction.type_.in_(
-                    [
-                        TransactionType.ADJUST_INTEREST,
-                        TransactionType.ADJUST_PENALTY,
-                    ]
-                ),
-            ),
-            case(
-                (literal(until_including), Transaction.time <= until),
-                else_=Transaction.time < until,
-            )
-            if until is not None
-            else literal(True),
-        )
-        .order_by(Transaction.time.asc())
-        .alias("trx_subset")
-    )
-
-    recursive_elements = (
-        select(
-            trx_subset.c.i,
-            trx_subset.c.time,
-            trx_subset.c.id.label("transaction_id"),
-            case(
-                # Adjusts balance -> balance gets adjusted
-                (
-                    trx_subset.c.type_ == TransactionType.ADJUST_BALANCE,
-                    recursive_cte.c.balance + trx_subset.c.amount,
-                ),
-                # Adds a product -> balance increases
-                (
-                    trx_subset.c.type_ == TransactionType.ADD_PRODUCT,
-                    recursive_cte.c.balance + trx_subset.c.amount,
-                ),
-                # Buys a product -> balance decreases
-                (
-                    trx_subset.c.type_ == TransactionType.BUY_PRODUCT,
-                    recursive_cte.c.balance
-                    - (
-                        trx_subset.c.product_count
-                        # Price of a single product, accounted for penalties and interest.
-                        * cast(
-                            func.ceil(
-                                # TODO: This can get quite expensive real quick, so we should do some caching of the
-                                #       product prices somehow.
-                                # Base price
-                                (
-                                    # FIXME: this always returns 0 for some reason...
-                                    select(cast(column("price"), Float))
-                                    .select_from(
-                                        _product_price_query(
-                                            trx_subset.c.product_id,
-                                            use_cache=use_cache,
-                                            until=trx_subset.c.time,
-                                            until_including=False,
-                                            cte_name="product_price_cte",
-                                        )
-                                    )
-                                    .order_by(column("i").desc())
-                                    .limit(1)
-                                ).scalar_subquery()
-                                # TODO: should interest be applied before or after the penalty multiplier?
-                                #       at the moment of writing, after sound right, but maybe ask someone?
-                                # Interest
-                                * (cast(recursive_cte.c.interest_rate_percent, Float) / 100)
-                                # Penalty
-                                * case(
-                                    (
-                                        # TODO: should this be <= or <?
-                                        recursive_cte.c.balance < recursive_cte.c.penalty_threshold,
-                                        (
-                                            cast(recursive_cte.c.penalty_multiplier_percent, Float)
-                                            / 100
-                                        ),
-                                    ),
-                                    else_=1.0,
-                                )
-                            ),
-                            Integer,
-                        )
-                    ),
-                ),
-                # Transfers money to self ->  balance increases
-                (
-                    and_(
-                        trx_subset.c.type_ == TransactionType.TRANSFER,
-                        trx_subset.c.transfer_user_id == user_id,
-                    ),
-                    recursive_cte.c.balance + trx_subset.c.amount,
-                ),
-                # Transfers money from self ->  balance decreases
-                (
-                    and_(
-                        trx_subset.c.type_ == TransactionType.TRANSFER,
-                        trx_subset.c.transfer_user_id != user_id,
-                    ),
-                    recursive_cte.c.balance - trx_subset.c.amount,
-                ),
-                # Interest adjustment -> balance stays the same
-                # Penalty adjustment -> balance stays the same
-                else_=recursive_cte.c.balance,
-            ).label("balance"),
-            case(
-                (
-                    trx_subset.c.type_ == TransactionType.ADJUST_INTEREST,
-                    trx_subset.c.interest_rate_percent,
-                ),
-                else_=recursive_cte.c.interest_rate_percent,
-            ).label("interest_rate_percent"),
-            case(
-                (
-                    trx_subset.c.type_ == TransactionType.ADJUST_PENALTY,
-                    trx_subset.c.penalty_threshold,
-                ),
-                else_=recursive_cte.c.penalty_threshold,
-            ).label("penalty_threshold"),
-            case(
-                (
-                    trx_subset.c.type_ == TransactionType.ADJUST_PENALTY,
-                    trx_subset.c.penalty_multiplier_percent,
-                ),
-                else_=recursive_cte.c.penalty_multiplier_percent,
-            ).label("penalty_multiplier_percent"),
-        )
-        .select_from(trx_subset)
-        .where(trx_subset.c.i == recursive_cte.c.i + 1)
-    )
-
-    return recursive_cte.union_all(recursive_elements)
-
-
-# TODO: create a function for the log that pretty prints the log entries
-#       for debugging purposes
-
-
-@dataclass
-class UserBalanceLogEntry:
-    transaction: Transaction
-    balance: int
-    interest_rate_percent: int
-    penalty_threshold: int
-    penalty_multiplier_percent: int
-
-    def is_penalized(self) -> bool:
-        """
-        Returns whether this exact transaction is penalized.
-        """
-
-        return False
-
-        # return self.transaction.type_ == TransactionType.BUY_PRODUCT and prev?
-
-
-def user_balance_log(
-    sql_session: Session,
-    user: User,
-    use_cache: bool = True,
-    until: Transaction | None = None,
-) -> list[UserBalanceLogEntry]:
-    """
-    Returns a log of the user's balance over time, including interest and penalty adjustments.
-    """
-
-    recursive_cte = _user_balance_query(
-        user.id,
-        use_cache=use_cache,
-        until=until.time if until else None,
-    )
-
-    result = sql_session.execute(
-        select(
-            Transaction,
-            recursive_cte.c.balance,
-            recursive_cte.c.interest_rate_percent,
-            recursive_cte.c.penalty_threshold,
-            recursive_cte.c.penalty_multiplier_percent,
-        )
-        .select_from(recursive_cte)
-        .join(
-            Transaction,
-            onclause=Transaction.id == recursive_cte.c.transaction_id,
-        )
-        .order_by(recursive_cte.c.i.asc())
-    ).all()
-
-    if result is None:
-        # If there are no transactions for this user, the query should return 0, not None.
-        raise RuntimeError(
-            f"Something went wrong while calculating the balance for user {user.name} (ID: {user.id})."
-        )
-
-    return [
-        UserBalanceLogEntry(
-            transaction=row[0],
-            balance=row.balance,
-            interest_rate_percent=row.interest_rate_percent,
-            penalty_threshold=row.penalty_threshold,
-            penalty_multiplier_percent=row.penalty_multiplier_percent,
-        )
-        for row in result
-    ]
-
-
-def user_balance(
-    sql_session: Session,
-    user: User,
-    use_cache: bool = True,
-    until: Transaction | None = None,
-) -> int:
-    """
-    Calculates the balance of a user.
-    """
-
-    recursive_cte = _user_balance_query(
-        user.id,
-        use_cache=use_cache,
-        until=until.time if until else None,
-    )
-
-    result = sql_session.scalar(
-        select(recursive_cte.c.balance).order_by(recursive_cte.c.i.desc()).limit(1)
-    )
-
-    if result is None:
-        # If there are no transactions for this user, the query should return 0, not None.
-        raise RuntimeError(
-            f"Something went wrong while calculating the balance for user {user.name} (ID: {user.id})."
-        )
-
-    return result

dibbler/queries/user_transactions.py

@@ -1,20 +0,0 @@
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from dibbler.models import Transaction, User
-
-# TODO: allow filtering out 'special transactions' like 'ADJUST_INTEREST' and 'ADJUST_PENALTY'
-
-
-def user_transactions(sql_session: Session, user: User) -> list[Transaction]:
-    """
-    Returns the transactions of the user in chronological order.
-    """
-
-    return list(
-        sql_session.scalars(
-            select(Transaction)
-            .where(Transaction.user_id == user.id)
-            .order_by(Transaction.time.asc())
-        ).all()
-    )

dibbler/subcommands/loop.py

@@ -1,79 +1,12 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-
 import random
-import sys
-import traceback
 
-from ..conf import config
+from sqlalchemy.orm import Session
-from ..lib.helpers import *
-from ..menus import *
 
+from ..menus.main import DibblerCli
+
+def main(sql_session: Session):
     random.seed()
 
+    DibblerCli.run_with_safe_exit_wrapper(sql_session)
 
-def main():
+    exit(0)
-    if not config.getboolean("general", "stop_allowed"):
-        signal.signal(signal.SIGQUIT, signal.SIG_IGN)
-
-    if not config.getboolean("general", "stop_allowed"):
-        signal.signal(signal.SIGTSTP, signal.SIG_IGN)
-
-    main = MainMenu(
-        "Dibbler main menu",
-        items=[
-            BuyMenu(),
-            ProductListMenu(),
-            ShowUserMenu(),
-            UserListMenu(),
-            AdjustCreditMenu(),
-            TransferMenu(),
-            AddStockMenu(),
-            Menu(
-                "Add/edit",
-                items=[
-                    AddUserMenu(),
-                    EditUserMenu(),
-                    AddProductMenu(),
-                    EditProductMenu(),
-                    AdjustStockMenu(),
-                    CleanupStockMenu(),
-                ],
-            ),
-            ProductSearchMenu(),
-            Menu(
-                "Statistics",
-                items=[
-                    ProductPopularityMenu(),
-                    ProductRevenueMenu(),
-                    BalanceMenu(),
-                    LoggedStatisticsMenu(),
-                ],
-            ),
-            FAQMenu(),
-            PrintLabelMenu(),
-        ],
-        exit_msg="happy happy joy joy",
-        exit_confirm_msg="Really quit Dibbler?",
-    )
-    if not config.getboolean("general", "quit_allowed"):
-        main.exit_disallowed_msg = "You can check out any time you like, but you can never leave."
-    while True:
-        # noinspection PyBroadException
-        try:
-            main.execute()
-        except KeyboardInterrupt:
-            print("")
-            print("Interrupted.")
-        except:
-            print("Something went wrong.")
-            print(f"{sys.exc_info()[0]}: {sys.exc_info()[1]}")
-            if config.getboolean("general", "show_tracebacks"):
-                traceback.print_tb(sys.exc_info()[2])
-        else:
-            break
-        print("Restarting main menu.")
-
-
-if __name__ == "__main__":
-    main()

dibbler/subcommands/makedb.py

@@ -1,11 +1,8 @@
-#!/usr/bin/python
+from sqlalchemy.orm import Session
 from dibbler.models import Base
-from dibbler.db import engine
 
 
-def main():
+def main(sql_session: Session):
-    Base.metadata.create_all(engine)
+    if not sql_session.bind:
-
+        raise RuntimeError("SQLAlchemy session is not bound to a database engine.")
-
+    Base.metadata.create_all(sql_session.bind)
-if __name__ == "__main__":
-    main()

dibbler/subcommands/seed_test_data.py

@@ -1,77 +1,45 @@
-from datetime import datetime
+import json
 from pathlib import Path
 
-from dibbler.db import Session
+from sqlalchemy.orm import Session
-from dibbler.models import Product, Transaction, User
+
+from dibbler.models import Product, User
 
 JSON_FILE = Path(__file__).parent.parent.parent / "mock_data.json"
 
 
-# TODO: integrate this as a part of create-db, either asking interactively
+def clear_db(sql_session: Session):
-#       whether to seed test data, or by using command line arguments for
-#       automatating the answer.
-
-
-def clear_db(sql_session):
     sql_session.query(Product).delete()
     sql_session.query(User).delete()
     sql_session.commit()
 
 
-def main():
+def main(sql_session: Session):
-    # TODO: There is some leftover json data in the mock_data.json file.
-    #       It should be dealt with before merging this PR, either by removing
-    #       it or using it here.
-    sql_session = Session()
     clear_db(sql_session)
+    product_items = []
+    user_items = []
 
-    # Add users
+    with open(JSON_FILE) as f:
-    user1 = User("Test User 1")
+        json_obj = json.load(f)
-    user2 = User("Test User 2")
 
-    sql_session.add(user1)
+        for product in json_obj["products"]:
-    sql_session.add(user2)
+            product_item = Product(
-    sql_session.commit()
+                bar_code=product["bar_code"],
-
+                name=product["name"],
-    # Add products
+                price=product["price"],
-    product1 = Product("1234567890123", "Test Product 1")
+                stock=product["stock"],
-    product2 = Product("9876543210987", "Test Product 2")
+            )
-    sql_session.add(product1)
+            product_items.append(product_item)
-    sql_session.add(product2)
+
-    sql_session.commit()
+        for user in json_obj["users"]:
-
+            user_item = User(
-    # Add transactions
+                name=user["name"],
-    transactions = [
+                card=user["card"],
-        Transaction.adjust_balance(
+                rfid=user["rfid"],
-            time=datetime(2023, 10, 1, 10, 0, 0),
+                credit=user["credit"],
-            amount=100,
+            )
-            user_id=user1.id,
+            user_items.append(user_item)
-        ),
+
-        Transaction.adjust_balance(
+        sql_session.add_all(product_items)
-            time=datetime(2023, 10, 1, 10, 0, 1),
+        sql_session.add_all(user_items)
-            amount=50,
-            user_id=user2.id,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 2),
-            amount=-50,
-            user_id=user1.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            amount=27 * 2,
-            per_product=27,
-            product_count=2,
-            user_id=user1.id,
-            product_id=product1.id,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            product_count=1,
-            user_id=user2.id,
-            product_id=product1.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
         sql_session.commit()

dibbler/subcommands/slabbedasker.py

@@ -1,18 +1,12 @@
-#!/usr/bin/python
+from sqlalchemy.orm import Session
 
-from dibbler.db import Session
+# from dibbler.db import Session
 from dibbler.models import User
 
 
-def main():
+def main(sql_session: Session):
-    # Start an SQL session
-    session = Session()
   # Let's find all users with a negative credit
-    slabbedasker = session.query(User).filter(User.credit < 0).all()
+    slabbedasker = sql_session.query(User).filter(User.credit < 0).all()
 
     for slubbert in slabbedasker:
         print(f"{slubbert.name}, {slubbert.credit}")
-
-
-if __name__ == "__main__":
-    main()

flake.lock

@@ -17,13 +17,65 @@
         "type": "indirect"
       }
     },
+    "flake-utils_2": {
+      "inputs": {
+        "systems": "systems_2"
+      },
+      "locked": {
+        "lastModified": 1731533236,
+        "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=",
+        "owner": "numtide",
+        "repo": "flake-utils",
+        "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b",
+        "type": "github"
+      },
+      "original": {
+        "id": "flake-utils",
+        "type": "indirect"
+      }
+    },
+    "libdib": {
+      "inputs": {
+        "flake-utils": "flake-utils_2",
+        "nixpkgs": "nixpkgs"
+      },
+      "locked": {
+        "lastModified": 1749301134,
+        "narHash": "sha256-JHLVV4ug8AgG71xhXEdmXozQfesXut6NUdXbBZNkD3c=",
+        "ref": "refs/heads/main",
+        "rev": "ca26131c22bb2833c81254dbabab6d785b9f37f0",
+        "revCount": 8,
+        "type": "git",
+        "url": "https://git.pvv.ntnu.no/Projects/libdib.git"
+      },
+      "original": {
+        "type": "git",
+        "url": "https://git.pvv.ntnu.no/Projects/libdib.git"
+      }
+    },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1749285348,
+        "lastModified": 1749143949,
-        "narHash": "sha256-frdhQvPbmDYaScPFiCnfdh3B/Vh81Uuoo0w5TkWmmjU=",
+        "narHash": "sha256-QuUtALJpVrPnPeozlUG/y+oIMSLdptHxb3GK6cpSVhA=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "3e3afe5174c561dee0df6f2c2b2236990146329f",
+        "rev": "d3d2d80a2191a73d1e86456a751b83aa13085d7d",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixos-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "nixpkgs_2": {
+      "locked": {
+        "lastModified": 1749143949,
+        "narHash": "sha256-QuUtALJpVrPnPeozlUG/y+oIMSLdptHxb3GK6cpSVhA=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "d3d2d80a2191a73d1e86456a751b83aa13085d7d",
         "type": "github"
       },
       "original": {
@@ -36,7 +88,8 @@
     "root": {
       "inputs": {
         "flake-utils": "flake-utils",
-        "nixpkgs": "nixpkgs"
+        "libdib": "libdib",
+        "nixpkgs": "nixpkgs_2"
       }
     },
     "systems": {
@@ -53,6 +106,21 @@
         "repo": "default",
         "type": "github"
       }
+    },
+    "systems_2": {
+      "locked": {
+        "lastModified": 1681028828,
+        "narHash": "sha256-Vy1rq5AaRuLzOxct8nz4T6wlgyUR7zLU309k9mBC768=",
+        "owner": "nix-systems",
+        "repo": "default",
+        "rev": "da67096a3b9bf56a91d16901293e51ba5b49a27e",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-systems",
+        "repo": "default",
+        "type": "github"
+      }
     }
   },
   "root": "root",

flake.nix

@@ -1,9 +1,13 @@
 {
   description = "Dibbler samspleisebod";
 
-  inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
+  inputs = {
+    nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
 
-  outputs = { self, nixpkgs, flake-utils }: let
+    libdib.url = "git+https://git.pvv.ntnu.no/Projects/libdib.git";
+  };
+
+  outputs = { self, nixpkgs, flake-utils, libdib }: let
     inherit (nixpkgs) lib;
 
     systems = [
@@ -14,7 +18,12 @@
     ];
 
     forAllSystems = f: lib.genAttrs systems (system: let
-      pkgs = nixpkgs.legacyPackages.${system};
+      pkgs = import nixpkgs {
+        inherit system;
+        overlays = [
+          libdib.overlays.default
+        ];
+      };
     in f system pkgs);
   in {
       packages = forAllSystems (system: pkgs: {

nix/dibbler.nix

@@ -1,6 +1,5 @@
 { lib
 , python3Packages
-, fetchFromGitHub
 }:
 python3Packages.buildPythonApplication {
   pname = "dibbler";
@@ -13,17 +12,10 @@ python3Packages.buildPythonApplication {
   # https://github.com/NixOS/nixpkgs/issues/285234
   dontCheckRuntimeDeps = true;
 
-  pythonImportsCheck = [];
-
-  doCheck = true;
-  nativeCheckInputs = with python3Packages; [
-    pytest
-    pytestCheckHook
-  ];
-
   nativeBuildInputs = with python3Packages; [ setuptools ];
   propagatedBuildInputs = with python3Packages; [
     brother-ql
+    libdib
     matplotlib
     psycopg2
     python-barcode

nix/module.nix

@@ -72,30 +72,6 @@ in {
       serviceConfig.Restart = "always"; # restart when session is closed
     };
 
-    services = {
+    services.getty.autologinUser = lib.mkForce "dibbler";
-      openssh = {
-        enable = true;
-        permitRootLogin = "yes";
-      };
-
-      getty.autologinUser = lib.mkForce "dibbler";
-      udisks2.enable = false;
-    };
-
-    networking.firewall.logRefusedConnections = false;
-    console.keyMap = "no";
-    programs.command-not-found.enable = false;
-    i18n.supportedLocales = [ "en_US.UTF-8/UTF-8" ];
-    # environment.noXlibs = true;
-
-    documentation = {
-      info.enable = false;
-      man.enable = false;
-    };
-
-    security = {
-      polkit.enable = lib.mkForce false;
-      audit.enable = false;
-    };
   };
 }

nix/shell.nix

@@ -12,12 +12,10 @@ mkShell {
     (python.withPackages (ps: with ps; [
       brother-ql
       matplotlib
+      libdib
       psycopg2
       python-barcode
       sqlalchemy
-
-      pytest
-      pytest-cov
     ]))
   ];
 }

pyproject.toml

@@ -15,16 +15,14 @@ dependencies = [
     "SQLAlchemy >= 2.0, <2.1",
     "brother-ql",
     "matplotlib",
+    "libdib",
     "psycopg2 >= 2.8, <2.10",
     "python-barcode",
 ]
 dynamic = ["version"]
 
-[project.optional-dependencies]
+[tool.uv.sources]
-dev = [
+libdib = { git = "https://git.pvv.ntnu.no/Projects/libdib.git" }
-    "pytest",
-    "pytest-cov",
-]
 
 [tool.setuptools.packages.find]
 include = ["dibbler*"]

tests/conftest.py

@@ -1,36 +0,0 @@
-import pytest
-
-from sqlalchemy import create_engine, event
-from sqlalchemy.orm import Session
-
-from dibbler.models import Base
-
-
-def pytest_addoption(parser):
-    parser.addoption(
-        "--echo",
-        action="store_true",
-        help="Enable SQLAlchemy echo mode for debugging",
-    )
-
-
-@pytest.fixture(scope="function")
-def sql_session(request):
-    """Create a new SQLAlchemy session for testing."""
-
-    echo = request.config.getoption("--echo")
-
-    engine = create_engine(
-        "sqlite:///:memory:",
-        echo=echo,
-    )
-
-    @event.listens_for(engine, "connect")
-    def set_sqlite_pragma(dbapi_connection, _connection_record):
-        cursor = dbapi_connection.cursor()
-        cursor.execute("PRAGMA foreign_keys=ON")
-        cursor.close()
-
-    Base.metadata.create_all(engine)
-    with Session(engine) as sql_session:
-        yield sql_session

tests/models/test_product.py

@@ -1,32 +0,0 @@
-import pytest
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product
-
-
-def insert_test_data(sql_session: Session) -> Product:
-    product = Product("1234567890123", "Test Product")
-    sql_session.add(product)
-    sql_session.commit()
-    return product
-
-
-def test_product_no_duplicate_barcodes(sql_session: Session):
-    product = insert_test_data(sql_session)
-
-    duplicate_product = Product(product.bar_code, "Hehe >:)")
-    sql_session.add(duplicate_product)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()
-
-
-def test_product_no_duplicate_names(sql_session: Session):
-    product = insert_test_data(sql_session)
-
-    duplicate_product = Product("1918238911928", product.name)
-    sql_session.add(duplicate_product)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()

tests/models/test_transaction.py

@@ -1,199 +0,0 @@
-from datetime import datetime
-
-import pytest
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-from dibbler.queries.product_stock import product_stock
-
-
-def insert_test_data(sql_session: Session) -> tuple[User, Product]:
-    user = User("Test User")
-    product = Product("1234567890123", "Test Product")
-
-    sql_session.add(user)
-    sql_session.add(product)
-    sql_session.commit()
-
-    return user, product
-
-
-def test_transaction_no_duplicate_timestamps(sql_session: Session):
-    user, _ = insert_test_data(sql_session)
-
-    transaction1 = Transaction.adjust_balance(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        user_id=user.id,
-        amount=100,
-    )
-
-    sql_session.add(transaction1)
-    sql_session.commit()
-
-    transaction2 = Transaction.adjust_balance(
-        time=transaction1.time,
-        user_id=user.id,
-        amount=-50,
-    )
-
-    sql_session.add(transaction2)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()
-
-
-def test_user_not_allowed_to_transfer_to_self(sql_session: Session) -> None:
-    user, _ = insert_test_data(sql_session)
-
-    transaction = Transaction.transfer(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        user_id=user.id,
-        transfer_user_id=user.id,
-        amount=50,
-    )
-
-    sql_session.add(transaction)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()
-
-
-def test_product_foreign_key_constraint(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transaction = Transaction.add_product(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        user_id=user.id,
-        product_id=product.id,
-        amount=27,
-        per_product=27,
-        product_count=1,
-    )
-
-    sql_session.add(transaction)
-    sql_session.commit()
-
-    # Attempt to add a transaction with a non-existent product
-    invalid_transaction = Transaction.add_product(
-        time=datetime(2023, 10, 1, 12, 0, 1),
-        user_id=user.id,
-        product_id=9999,  # Non-existent product ID
-        amount=27,
-        per_product=27,
-        product_count=1,
-    )
-
-    sql_session.add(invalid_transaction)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()
-
-
-def test_user_foreign_key_constraint(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transaction = Transaction.add_product(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        user_id=user.id,
-        product_id=product.id,
-        amount=27,
-        per_product=27,
-        product_count=1,
-    )
-
-    sql_session.add(transaction)
-    sql_session.commit()
-
-    # Attempt to add a transaction with a non-existent user
-    invalid_transaction = Transaction.add_product(
-        time=datetime(2023, 10, 1, 12, 0, 1),
-        user_id=9999,  # Non-existent user ID
-        product_id=product.id,
-        amount=27,
-        per_product=27,
-        product_count=1,
-    )
-
-    sql_session.add(invalid_transaction)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()
-
-
-def test_transaction_buy_product_more_than_stock(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            product_count=10,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    assert product_stock(sql_session, product) == 1 - 10
-
-
-def test_transaction_buy_product_dont_allow_no_add_product_transactions(
-    sql_session: Session,
-) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transaction = Transaction.buy_product(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        product_count=1,
-        user_id=user.id,
-        product_id=product.id,
-    )
-
-    sql_session.add(transaction)
-
-    with pytest.raises(ValueError):
-        sql_session.commit()
-
-
-def test_transaction_add_product_deny_amount_over_per_product_times_product_count(
-    sql_session: Session,
-) -> None:
-    user, product = insert_test_data(sql_session)
-
-    with pytest.raises(ValueError):
-        _transaction = Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27 * 2 + 1,  # Invalid amount
-            per_product=27,
-            product_count=2,
-        )
-
-
-def test_transaction_add_product_allow_amount_under_per_product_times_product_count(
-    sql_session: Session,
-) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transaction = Transaction.add_product(
-        time=datetime(2023, 10, 1, 12, 0, 0),
-        user_id=user.id,
-        product_id=product.id,
-        amount=27 * 2 - 1,  # Valid amount
-        per_product=27,
-        product_count=2,
-    )
-
-    sql_session.add(transaction)
-    sql_session.commit()

tests/models/test_user.py

@@ -1,25 +0,0 @@
-from datetime import datetime
-
-import pytest
-from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-
-
-def insert_test_data(sql_session: Session) -> User:
-    user = User("Test User")
-    sql_session.add(user)
-    sql_session.commit()
-
-    return user
-
-
-def test_ensure_no_duplicate_user_names(sql_session: Session):
-    user = insert_test_data(sql_session)
-
-    user2 = User(user.name)
-    sql_session.add(user2)
-
-    with pytest.raises(IntegrityError):
-        sql_session.commit()

tests/queries/test_product_price.py

@@ -1,342 +0,0 @@
-import math
-from datetime import datetime
-from pprint import pprint
-
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-from dibbler.queries.product_price import product_price, product_price_log
-
-# TODO: see if we can use pytest_runtest_makereport to print the "product_price_log"s
-#       only on failures instead of inlining it in every test function
-
-
-def insert_test_data(sql_session: Session) -> tuple[User, Product]:
-    user = User("Test User")
-    product = Product("1234567890123", "Test Product")
-
-    sql_session.add(user)
-    sql_session.add(product)
-    sql_session.commit()
-
-    return user, product
-
-
-def test_product_price_no_transactions(sql_session: Session) -> None:
-    _, product = insert_test_data(sql_session)
-
-    pprint(product_price_log(sql_session, product))
-
-    assert product_price(sql_session, product) == 0
-
-
-def test_product_price_basic_history(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            amount=27 * 2 - 1,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    assert product_price(sql_session, product) == 27
-
-
-def test_product_price_sold_out(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            amount=27 * 2 - 1,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    assert product_price(sql_session, product) == 27
-
-
-def test_product_price_interest(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            interest_rate_percent=110,
-            user_id=user.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            amount=27 * 2 - 1,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    product_price_ = product_price(sql_session, product)
-    product_price_interest = product_price(sql_session, product, include_interest=True)
-
-    assert product_price_ == 27
-    assert product_price_interest == math.ceil(27 * 1.1)
-
-
-def test_product_price_changing_interest(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            interest_rate_percent=110,
-            user_id=user.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            amount=27 * 2 - 1,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 12, 0, 2),
-            interest_rate_percent=120,
-            user_id=user.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    product_price_interest = product_price(sql_session, product, include_interest=True)
-    assert product_price_interest == math.ceil(27 * 1.2)
-
-
-def test_product_price_old_transaction(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            amount=27 * 2,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # Price should be 27
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 2),
-            amount=38 * 3,
-            per_product=38,
-            product_count=3,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # price should be averaged upwards
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    until_transaction = transactions[0]
-
-    pprint(
-        product_price_log(
-            sql_session,
-            product,
-            until=until_transaction,
-        )
-    )
-
-    product_price_ = product_price(
-        sql_session,
-        product,
-        until=until_transaction,
-    )
-    assert product_price_ == 27
-
-
-# Price goes up and gets rounded up to the next integer
-def test_product_price_round_up_from_below(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            amount=27 * 2,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # Price should be 27
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 2),
-            amount=38 * 3,
-            per_product=38,
-            product_count=3,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # price should be averaged upwards
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    product_price_ = product_price(sql_session, product)
-    assert product_price_ == math.ceil((27 * 2 + 38 * 3) / (2 + 3))
-
-
-# Price goes down and gets rounded up to the next integer
-def test_product_price_round_up_from_above(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            amount=27 * 2,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # Price should be 27
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 2),
-            amount=20 * 3,
-            per_product=20,
-            product_count=3,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        # price should be averaged downwards
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    product_price_ = product_price(sql_session, product)
-    assert product_price_ == math.ceil((27 * 2 + 20 * 3) / (2 + 3))
-
-
-def test_product_price_with_negative_stock_single_addition(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            amount=1,
-            per_product=10,
-            product_count=1,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 13, 0, 1),
-            product_count=10,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 2),
-            amount=22,
-            per_product=22,
-            product_count=1,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    # Stock went subzero, price should be the last added product price
-    product1_price = product_price(sql_session, product)
-    assert product1_price == 22
-
-
-# TODO: what happens when stock is still negative and yet new products are added?
-def test_product_price_with_negative_stock_multiple_additions(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            amount=1,
-            per_product=10,
-            product_count=1,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 13, 0, 1),
-            product_count=10,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 2),
-            amount=22,
-            per_product=22,
-            product_count=1,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 3),
-            amount=29,
-            per_product=29,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(product_price_log(sql_session, product))
-
-    # Stock went subzero, price should be the ceiled average of the last added products
-    product1_price = product_price(sql_session, product)
-    assert product1_price == math.ceil((22 + 29 * 2) / (1 + 2))

tests/queries/test_product_stock.py

@@ -1,141 +0,0 @@
-from datetime import datetime
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-from dibbler.queries.product_stock import product_stock
-
-
-def insert_test_data(sql_session: Session) -> None:
-    user1 = User("Test User 1")
-
-    sql_session.add(user1)
-    sql_session.commit()
-
-
-def test_product_stock_basic_history(sql_session: Session) -> None:
-    insert_test_data(sql_session)
-
-    user1 = sql_session.scalars(select(User).where(User.name == "Test User 1")).one()
-
-    product = Product("1234567890123", "Test Product")
-    sql_session.add(product)
-    sql_session.commit()
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            amount=10,
-            per_product=10,
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    assert product_stock(sql_session, product) == 1
-
-
-def test_product_stock_complex_history(sql_session: Session) -> None:
-    insert_test_data(sql_session)
-
-    user1 = sql_session.scalars(select(User).where(User.name == "Test User 1")).one()
-
-    product = Product("1234567890123", "Test Product")
-    sql_session.add(product)
-    sql_session.commit()
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            amount=27 * 2,
-            per_product=27,
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=2,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 13, 0, 1),
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=3,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 13, 0, 2),
-            amount=50 * 4,
-            per_product=50,
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=4,
-        ),
-        Transaction.adjust_stock(
-            time=datetime(2023, 10, 1, 15, 0, 0),
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=3,
-        ),
-        Transaction.adjust_stock(
-            time=datetime(2023, 10, 1, 15, 0, 1),
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=-2,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    assert product_stock(sql_session, product) == 2 - 3 + 4 + 3 - 2
-
-
-def test_product_stock_no_transactions(sql_session: Session) -> None:
-    insert_test_data(sql_session)
-
-    product = Product("1234567890123", "Test Product")
-    sql_session.add(product)
-    sql_session.commit()
-
-    assert product_stock(sql_session, product) == 0
-
-
-def test_negative_product_stock(sql_session: Session) -> None:
-    insert_test_data(sql_session)
-
-    user1 = sql_session.scalars(select(User).where(User.name == "Test User 1")).one()
-
-    product = Product("1234567890123", "Test Product")
-    sql_session.add(product)
-    sql_session.commit()
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 14, 0, 0),
-            amount=50,
-            per_product=50,
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 14, 0, 1),
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=2,
-        ),
-        Transaction.adjust_stock(
-            time=datetime(2023, 10, 1, 16, 0, 0),
-            user_id=user1.id,
-            product_id=product.id,
-            product_count=-1,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    # The stock should be negative because we added and bought the product
-    assert product_stock(sql_session, product) == 1 - 2 - 1

tests/queries/test_user_balance.py

@@ -1,306 +0,0 @@
-import math
-from datetime import datetime
-from pprint import pprint
-
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-from dibbler.queries.user_balance import user_balance, user_balance_log
-
-# TODO: see if we can use pytest_runtest_makereport to print the "user_balance_log"s
-#       only on failures instead of inlining it in every test function
-
-
-def insert_test_data(sql_session: Session) -> tuple[User, Product]:
-    user = User("Test User")
-    product = Product("1234567890123", "Test Product")
-
-    sql_session.add(user)
-    sql_session.add(product)
-    sql_session.commit()
-
-    return user, product
-
-
-def test_user_balance_no_transactions(sql_session: Session) -> None:
-    user, _ = insert_test_data(sql_session)
-
-    pprint(user_balance_log(sql_session, user))
-
-    balance = user_balance(sql_session, user)
-
-    assert balance == 0
-
-
-def test_user_balance_basic_history(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            amount=100,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 1),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    balance = user_balance(sql_session, user)
-
-    assert balance == 100 + 27
-
-
-def test_user_balance_with_transfers(sql_session: Session) -> None:
-    user1, product = insert_test_data(sql_session)
-
-    user2 = User("Test User 2")
-    sql_session.add(user2)
-    sql_session.commit()
-
-    transactions = [
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user1.id,
-            amount=100,
-        ),
-        Transaction.transfer(
-            time=datetime(2023, 10, 1, 10, 0, 1),
-            user_id=user1.id,
-            transfer_user_id=user2.id,
-            amount=50,
-        ),
-        Transaction.transfer(
-            time=datetime(2023, 10, 1, 10, 0, 2),
-            user_id=user2.id,
-            transfer_user_id=user1.id,
-            amount=30,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user1))
-
-    user1_balance = user_balance(sql_session, user1)
-    assert user1_balance == 100 - 50 + 30
-
-    pprint(user_balance_log(sql_session, user2))
-
-    user2_balance = user_balance(sql_session, user2)
-    assert user2_balance == 50 - 30
-
-
-def test_user_balance_complex_history(sql_session: Session) -> None:
-    raise NotImplementedError("This test is not implemented yet.")
-
-
-def test_user_balance_penalty(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 11, 0, 0),
-            user_id=user.id,
-            amount=-200,
-        ),
-        # Penalized, pays 2x the price (default penalty)
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    assert user_balance(sql_session, user) == 27 - 200 - (27 * 2)
-
-
-def test_user_balance_changing_penalty(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 11, 0, 0),
-            user_id=user.id,
-            amount=-200,
-        ),
-        # Penalized, pays 2x the price (default penalty)
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-        Transaction.adjust_penalty(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            user_id=user.id,
-            penalty_multiplier_percent=300,
-            penalty_threshold=-100,
-        ),
-        # Penalized, pays 3x the price
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 14, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    assert user_balance(sql_session, user) == 27 - 200 - (27 * 2) - (27 * 3)
-
-
-def test_user_balance_interest(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 11, 0, 0),
-            user_id=user.id,
-            interest_rate_percent=110,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    assert user_balance(sql_session, user) == 27 - math.ceil(27 * 1.1)
-
-
-def test_user_balance_changing_interest(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27 * 3,
-            per_product=27,
-            product_count=3,
-        ),
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 11, 0, 0),
-            user_id=user.id,
-            interest_rate_percent=110,
-        ),
-        # Pays 1.1x the price
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            user_id=user.id,
-            interest_rate_percent=120,
-        ),
-        # Pays 1.2x the price
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 14, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    assert user_balance(sql_session, user) == 27 * 3 - math.ceil(27 * 1.1) - math.ceil(27 * 1.2)
-
-
-def test_user_balance_penalty_interest_combined(sql_session: Session) -> None:
-    user, product = insert_test_data(sql_session)
-
-    transactions = [
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            amount=27,
-            per_product=27,
-            product_count=1,
-        ),
-        Transaction.adjust_interest(
-            time=datetime(2023, 10, 1, 11, 0, 0),
-            user_id=user.id,
-            interest_rate_percent=110,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            user_id=user.id,
-            amount=-200,
-        ),
-        # Penalized, pays 2x the price (default penalty)
-        # Pays 1.1x the price
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 13, 0, 0),
-            user_id=user.id,
-            product_id=product.id,
-            product_count=1,
-        ),
-    ]
-    sql_session.add_all(transactions)
-    sql_session.commit()
-
-    pprint(user_balance_log(sql_session, user))
-
-    assert user_balance(sql_session, user) == (
-        27
-        - 200
-        - math.ceil(27 * 2 * 1.1)
-    )

tests/queries/test_user_transactions.py

@@ -1,60 +0,0 @@
-from datetime import datetime
-
-from sqlalchemy.orm import Session
-
-from dibbler.models import Product, Transaction, User
-from dibbler.queries.user_transactions import user_transactions
-
-
-def insert_test_data(sql_session: Session) -> User:
-    user = User("Test User")
-    sql_session.add(user)
-    sql_session.commit()
-
-    return user
-
-
-def test_user_transactions(sql_session: Session):
-    user = insert_test_data(sql_session)
-
-    product = Product("1234567890123", "Test Product")
-    user2 = User("Test User 2")
-    sql_session.add_all([product, user2])
-    sql_session.commit()
-
-    transactions = [
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 0),
-            amount=100,
-            user_id=user.id,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 1),
-            amount=50,
-            user_id=user2.id,
-        ),
-        Transaction.adjust_balance(
-            time=datetime(2023, 10, 1, 10, 0, 2),
-            amount=-50,
-            user_id=user.id,
-        ),
-        Transaction.add_product(
-            time=datetime(2023, 10, 1, 12, 0, 0),
-            amount=27 * 2,
-            per_product=27,
-            product_count=2,
-            user_id=user.id,
-            product_id=product.id,
-        ),
-        Transaction.buy_product(
-            time=datetime(2023, 10, 1, 12, 0, 1),
-            product_count=1,
-            user_id=user2.id,
-            product_id=product.id,
-        ),
-    ]
-
-    sql_session.add_all(transactions)
-
-    assert len(user_transactions(sql_session, user)) == 3
-    assert len(user_transactions(sql_session, user2)) == 2