fixup! WIP

dibbler/queries/adjust_interest.py

@@ -1,2 +1,25 @@
-# NOTE: this type of transaction should be password protected.
+from sqlalchemy.orm import Session
+
+from dibbler.models import Transaction
+
+# TODO: this type of transaction should be password protected.
 #       the password can be set as a string literal in the config file.
+
+
+def adjust_interest(
+    sql_session: Session,
+    user_id: int,
+    new_interest: int,
+    message: str | None = None,
+) -> None:
+    if new_interest < 0:
+        raise ValueError("Interest rate cannot be negative")
+
+    transaction = Transaction.adjust_interest(
+        user_id=user_id,
+        interest_rate_percent=new_interest,
+        message=message,
+    )
+
+    sql_session.add(transaction)
+    sql_session.commit()

dibbler/queries/adjust_penalty.py

@@ -1,2 +1,43 @@
-# NOTE: this type of transaction should be password protected.
+from sqlalchemy.orm import Session
+
+from dibbler.models import Transaction
+from dibbler.queries.current_penalty import current_penalty
+
+# TODO: this type of transaction should be password protected.
 #       the password can be set as a string literal in the config file.
+
+def adjust_penalty(
+    sql_session: Session,
+    user_id: int,
+    new_penalty: int | None,
+    new_penalty_multiplier: int | None,
+    message: str | None = None,
+) -> None:
+    """
+    Adjust the penalty rate and/or penalty multiplier.
+    """
+    if new_penalty is None and new_penalty_multiplier is None:
+        raise ValueError("At least one of new_penalty or new_penalty_multiplier must be provided")
+
+    if new_penalty is not None and new_penalty < 0:
+        raise ValueError("Penalty rate cannot be negative")
+
+    if new_penalty_multiplier is not None and new_penalty_multiplier < 0:
+        raise ValueError("Penalty multiplier cannot be negative")
+
+    if new_penalty is None or new_penalty_multiplier is None:
+        existing_penalty, existing_penalty_multiplier = current_penalty(sql_session)
+        if new_penalty is None:
+            new_penalty = existing_penalty
+        if new_penalty_multiplier is None:
+            new_penalty_multiplier = existing_penalty_multiplier
+
+    transaction = Transaction.adjust_penalty(
+        user_id=user_id,
+        penalty_threshold=new_penalty,
+        penalty_multiplier_percent=new_penalty_multiplier,
+        message=message,
+    )
+
+    sql_session.add(transaction)
+    sql_session.commit()

dibbler/queries/current_interest.py

@@ -15,5 +15,7 @@ def current_interest(sql_session: Session) -> int:
 
     if result is None:
         return DEFAULT_INTEREST_RATE_PERCENTAGE
-
-    return result.interest_rate_percent
+    elif result.interest_rate_percent is None:
+        return DEFAULT_INTEREST_RATE_PERCENTAGE
+    else:
+        return result.interest_rate_percent

dibbler/queries/product_price.py

@@ -195,7 +195,6 @@ def product_price_log(
     ]
 
 
-@staticmethod
 def product_price(
     sql_session: Session,
     product: Product,

dibbler/queries/user_transactions.py

@@ -2,15 +2,43 @@ 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'
+from dibbler.models.TransactionType import TransactionType
 
 
-def user_transactions(sql_session: Session, user: User) -> list[Transaction]:
+def user_transactions(
+    sql_session: Session,
+    user: User,
+    transaction_type_filter: list[TransactionType] | None = None,
+    negate_filter: bool = False,
+) -> list[Transaction]:
     """
     Returns the transactions of the user in chronological order.
     """
 
+    if transaction_type_filter is not None:
+        if negate_filter:
+            return list(
+                sql_session.scalars(
+                    select(Transaction)
+                    .where(
+                        Transaction.user_id == user.id,
+                        Transaction.type_.not_in(transaction_type_filter),
+                    )
+                    .order_by(Transaction.time.asc())
+                ).all()
+            )
+        else:
+            return list(
+                sql_session.scalars(
+                    select(Transaction)
+                    .where(
+                        Transaction.user_id == user.id,
+                        Transaction.type_.in_(transaction_type_filter),
+                    )
+                    .order_by(Transaction.time.asc())
+                ).all()
+            )
+
     return list(
         sql_session.scalars(
             select(Transaction)