diff --git a/dibbler/queries/transaction_log.py b/dibbler/queries/transaction_log.py new file mode 100644 index 0000000..bb552a7 --- /dev/null +++ b/dibbler/queries/transaction_log.py @@ -0,0 +1,85 @@ +from sqlalchemy import select +from sqlalchemy.orm import Session + +from dibbler.models import ( + Product, + Transaction, + TransactionType, + User, +) + + +def transaction_log( + sql_session: Session, + user: User | None = None, + product: Product | None = None, + + exclusive_after: bool = False, + after_time=None, + after_transaction_id: int | None = None, + + exclusive_before: bool = False, + before_time=None, + before_transaction_id: int | None = None, + + transaction_type: list[TransactionType] | None = None, + negate_transaction_type_filter: bool = False, + limit: int | None = None, +) -> list[Transaction]: + """ + Retrieve the transaction log, optionally filtered. + + Only one of `user` or `product` may be specified. + Only one of `after_time` or `after_transaction_id` may be specified. + Only one of `before_time` or `before_transaction_id` may be specified. + + The before and after filters are inclusive by default. + """ + + if not (user is None or product is None): + raise ValueError("Cannot filter by both user and product.") + + if not (after_time is None or after_transaction_id is None): + raise ValueError("Cannot filter by both from_time and from_transaction_id.") + + query = select(Transaction) + if user is not None: + query = query.where(Transaction.user_id == user.id) + if product is not None: + query = query.where(Transaction.product_id == product.id) + + if after_time is not None: + if exclusive_after: + query = query.where(Transaction.time > after_time) + else: + query = query.where(Transaction.time >= after_time) + if after_transaction_id is not None: + if exclusive_after: + query = query.where(Transaction.id > after_transaction_id) + else: + query = query.where(Transaction.id >= after_transaction_id) + + if before_time is not None: + if exclusive_before: + query = query.where(Transaction.time < before_time) + else: + query = query.where(Transaction.time <= before_time) + if before_transaction_id is not None: + if exclusive_before: + query = query.where(Transaction.id < before_transaction_id) + else: + query = query.where(Transaction.id <= before_transaction_id) + + if transaction_type is not None: + if negate_transaction_type_filter: + query = query.where(~Transaction.type_.in_(transaction_type)) + else: + query = query.where(Transaction.type_.in_(transaction_type)) + + if limit is not None: + query = query.limit(limit) + + query = query.order_by(Transaction.time.asc(), Transaction.id.asc()) + result = sql_session.scalars(query).all() + + return list(result) diff --git a/dibbler/queries/user_transactions.py b/dibbler/queries/user_transactions.py deleted file mode 100644 index d30b426..0000000 --- a/dibbler/queries/user_transactions.py +++ /dev/null @@ -1,48 +0,0 @@ -from sqlalchemy import select -from sqlalchemy.orm import Session - -from dibbler.models import Transaction, User -from dibbler.models.TransactionType import TransactionType - - -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) - .where(Transaction.user_id == user.id) - .order_by(Transaction.time.asc()) - ).all() - ) diff --git a/tests/queries/test_transaction_log.py b/tests/queries/test_transaction_log.py new file mode 100644 index 0000000..da0949d --- /dev/null +++ b/tests/queries/test_transaction_log.py @@ -0,0 +1,565 @@ +from datetime import datetime, timedelta + +import pytest +from sqlalchemy.orm import Session + +from dibbler.models import ( + Product, + Transaction, + TransactionType, + User, +) +from dibbler.queries.transaction_log import transaction_log + + +def insert_test_data(sql_session: Session) -> tuple[User, User, Product, Product]: + user1 = User("Test User 1") + user2 = User("Test User 2") + + product1 = Product("1234567890123", "Test Product 1") + product2 = Product("9876543210987", "Test Product 2") + sql_session.add_all([user1, user2, product1, product2]) + sql_session.commit() + + return user1, user2, product1, product2 + + +def insert_default_test_transactions( + sql_session: Session, + user1: User, + user2: User, + product1: Product, + product2: Product, +) -> list[Transaction]: + transactions = [ + Transaction.adjust_balance( + time=datetime(2023, 10, 1, 10, 0, 0), + amount=100, + user_id=user1.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=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=product2.id, + ), + Transaction.add_product( + time=datetime(2023, 10, 1, 12, 0, 2), + amount=15 * 1, + per_product=15, + product_count=1, + user_id=user2.id, + product_id=product2.id, + ), + Transaction.transfer( + time=datetime(2023, 10, 1, 14, 0, 0), + amount=30, + user_id=user1.id, + transfer_user_id=user2.id, + ), + ] + + sql_session.add_all(transactions) + sql_session.commit() + + return transactions + + +def test_user_transactions_no_transactions(sql_session: Session): + insert_test_data(sql_session) + + transactions = transaction_log(sql_session) + + assert len(transactions) == 0 + + +def test_transaction_log_filtered_by_user(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert len(transaction_log(sql_session, user=user)) == 4 + assert len(transaction_log(sql_session, user=user2)) == 3 + + +def test_transaction_log_filtered_by_product(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert len(transaction_log(sql_session, product=product)) == 1 + assert len(transaction_log(sql_session, product=product2)) == 2 + + +def test_transaction_log_after_datetime(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + after_time=transactions[2].time, + ) + ) + == len(transactions) - 2 + ) + + +def test_transaction_log_after_datetime_no_transactions(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + assert ( + len( + transaction_log( + sql_session, + after_time=transactions[-1].time + timedelta(seconds=1), + ) + ) + == 0 + ) + + +def test_transaction_log_after_datetime_exclusive(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + after_time=transactions[2].time, + exclusive_after=True, + ) + ) + == len(transactions) - 3 + ) + + +def test_transaction_log_after_transaction_id(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + first_transaction = transactions[0] + + assert len( + transaction_log( + sql_session, + after_transaction_id=first_transaction.id, + ) + ) == len(transactions) + + +def test_transaction_log_after_transaction_id_one_transaction(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + last_transaction = transactions[-1] + + assert ( + len( + transaction_log( + sql_session, + after_transaction_id=last_transaction.id, + ) + ) + == 1 + ) + + +def test_transaction_log_after_transaction_id_exclusive(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + third_transaction = transactions[2] + + assert ( + len( + transaction_log( + sql_session, + after_transaction_id=third_transaction.id, + exclusive_after=True, + ) + ) + == len(transactions) - 3 + ) + + +def test_transaction_log_before_datetime(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + before_time=transactions[-3].time, + ) + ) + == len(transactions) - 2 + ) + + +def test_transaction_log_before_datetime_no_transactions(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + before_time=transactions[0].time - timedelta(seconds=1), + ) + ) + == 0 + ) + + +def test_transaction_log_before_datetime_exclusive(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + before_time=transactions[-3].time, + exclusive_before=True, + ) + ) + == len(transactions) - 3 + ) + + +def test_transaction_log_before_transaction_id(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + last_transaction = transactions[-3] + + assert ( + len( + transaction_log( + sql_session, + before_transaction_id=last_transaction.id, + ) + ) + == len(transactions) - 2 + ) + + +def test_transaction_log_before_transaction_id_one_transaction(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + first_transaction = transactions[0] + + assert ( + len( + transaction_log( + sql_session, + before_transaction_id=first_transaction.id, + ) + ) + == 1 + ) + + +def test_transaction_log_before_transaction_id_exclusive(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + last_transaction = transactions[-3] + + assert ( + len( + transaction_log( + sql_session, + before_transaction_id=last_transaction.id, + exclusive_before=True, + ) + ) + == len(transactions) - 3 + ) + + +def test_transaction_log_before_after_datetime_combined(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + fifth_transaction = transactions[4] + + assert ( + len( + transaction_log( + sql_session, + after_time=second_transaction.time, + before_time=fifth_transaction.time, + ) + ) + == 4 + ) + + +def test_transaction_log_before_after_transaction_id_combined(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + fifth_transaction = transactions[4] + + assert ( + len( + transaction_log( + sql_session, + after_transaction_id=second_transaction.id, + before_transaction_id=fifth_transaction.id, + ) + ) + == 4 + ) + + +def test_transaction_log_before_date_after_transaction_id(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + fifth_transaction = transactions[4] + + assert ( + len( + transaction_log( + sql_session, + before_time=fifth_transaction.time, + after_transaction_id=second_transaction.id, + ) + ) + == 4 + ) + + +def test_transaction_log_before_transaction_id_after_date(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + fifth_transaction = transactions[4] + + assert ( + len( + transaction_log( + sql_session, + before_transaction_id=fifth_transaction.id, + after_time=second_transaction.time, + ) + ) + == 4 + ) + + +def test_transaction_log_after_product_and_user_not_allowed(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + insert_default_test_transactions(sql_session, user, user2, product, product2) + + with pytest.raises(ValueError): + transaction_log( + sql_session, + user=user, + product=product, + after_time=datetime(2023, 10, 1, 11, 0, 0), + ) + + +def test_transaction_log_after_datetime_and_transaction_id_not_allowed(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + insert_default_test_transactions(sql_session, user, user2, product, product2) + + with pytest.raises(ValueError): + transaction_log( + sql_session, + user=user, + after_time=datetime(2023, 10, 1, 11, 0, 0), + after_transaction_id=1, + ) + + +def test_transaction_log_limit(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert len(transaction_log(sql_session, limit=3)) == 3 + assert len(transaction_log(sql_session, limit=len(transactions) + 3)) == len(transactions) + + +def test_transaction_log_filtered_by_transaction_type(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.ADJUST_BALANCE], + ) + ) + == 3 + ) + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.ADD_PRODUCT], + ) + ) + == 2 + ) + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.BUY_PRODUCT, TransactionType.ADD_PRODUCT], + ) + ) + == 3 + ) + + +def test_transaction_log_filtered_by_transaction_type_negated(sql_session: Session): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.ADJUST_BALANCE], + negate_transaction_type_filter=True, + ) + ) + == len(transactions) - 3 + ) + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.ADD_PRODUCT], + negate_transaction_type_filter=True, + ) + ) + == len(transactions) - 2 + ) + assert ( + len( + transaction_log( + sql_session, + transaction_type=[TransactionType.BUY_PRODUCT, TransactionType.ADD_PRODUCT], + negate_transaction_type_filter=True, + ) + ) + == len(transactions) - 3 + ) + + +def test_transaction_log_combined_filter_user_datetime_transaction_type_limit( + sql_session: Session, +): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + sixth_transaction = transactions[5] + + result = transaction_log( + sql_session, + user=user, + after_time=second_transaction.time, + before_time=sixth_transaction.time, + transaction_type=[TransactionType.ADJUST_BALANCE, TransactionType.ADD_PRODUCT], + limit=2, + ) + + assert len(result) == 2 + + +def test_transaction_log_combined_filter_user_transaction_id_transaction_type_limit( + sql_session: Session, +): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + sixth_transaction = transactions[5] + + result = transaction_log( + sql_session, + user=user, + after_transaction_id=second_transaction.id, + before_transaction_id=sixth_transaction.id, + transaction_type=[TransactionType.ADJUST_BALANCE, TransactionType.ADD_PRODUCT], + limit=2, + ) + + assert len(result) == 2 + + +def test_transaction_log_combined_filter_product_datetime_transaction_type_limit( + sql_session: Session, +): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + sixth_transaction = transactions[5] + + result = transaction_log( + sql_session, + product=product2, + after_time=second_transaction.time, + before_time=sixth_transaction.time, + transaction_type=[TransactionType.BUY_PRODUCT, TransactionType.ADD_PRODUCT], + limit=2, + ) + + assert len(result) == 2 + + +def test_transaction_log_combined_filter_product_transaction_id_transaction_type_limit( + sql_session: Session, +): + user, user2, product, product2 = insert_test_data(sql_session) + transactions = insert_default_test_transactions(sql_session, user, user2, product, product2) + + second_transaction = transactions[1] + sixth_transaction = transactions[5] + + result = transaction_log( + sql_session, + product=product2, + after_transaction_id=second_transaction.id, + before_transaction_id=sixth_transaction.id, + transaction_type=[TransactionType.BUY_PRODUCT, TransactionType.ADD_PRODUCT], + limit=2, + ) + + assert len(result) == 2 + + +def test_transaction_log_filtered_by_user_joint_transactions(sql_session: Session): ... diff --git a/tests/queries/test_user_transactions.py b/tests/queries/test_user_transactions.py deleted file mode 100644 index d1029ec..0000000 --- a/tests/queries/test_user_transactions.py +++ /dev/null @@ -1,155 +0,0 @@ -from datetime import datetime - -from sqlalchemy.orm import Session - -from dibbler.models import Product, Transaction, User -from dibbler.models.TransactionType import TransactionType -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_no_transactions(sql_session: Session): - pass - - -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 - - -def test_filtered_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, - transaction_type_filter=[TransactionType.ADJUST_BALANCE], - ) - ) - == 2 - ) - assert ( - len( - user_transactions( - sql_session, - user, - transaction_type_filter=[TransactionType.ADJUST_BALANCE], - negate_filter=True, - ) - ) - == 1 - ) - assert ( - len( - user_transactions( - sql_session, - user2, - transaction_type_filter=[TransactionType.ADJUST_BALANCE], - ) - ) - == 1 - ) - assert ( - len( - user_transactions( - sql_session, - user2, - transaction_type_filter=[TransactionType.ADJUST_BALANCE], - negate_filter=True, - ) - ) - == 1 - ) - - -def test_user_transactions_joint_transactions(sql_session: Session): - pass