✨ [feat][backend] Extend read txn API to support txn id filter

Author: tomy0000000

backend/kayman/crud/transaction.py

@@ -27,6 +27,7 @@ def create_transactions(
 
 def read_transactions(
     session: Session,
+    transaction_id: int | None = None,
     account_id: int | None = None,
     start: datetime | None = None,
     end: datetime | None = None,
@@ -35,6 +36,8 @@ def read_transactions(
     descending: bool = False,
 ) -> Sequence[Transaction]:
     scalar = select(Transaction)
+    if transaction_id is not None:
+        scalar = scalar.where(Transaction.id == transaction_id)
     if account_id:
         scalar = scalar.where(Transaction.account_id == account_id)
     if event_id == "empty":

backend/kayman/logics/transaction.py

@@ -19,7 +19,9 @@ def get_transactions_with_running_balance(
         if start is not None
         else Decimal(0)
     )
-    txns = read_transactions(session, account_id, start, end, order_by="created_at")
+    txns = read_transactions(
+        session, account_id=account_id, start=start, end=end, order_by="created_at"
+    )
     running = opening
     results: list[TransactionReadWithBalance] = []
     for txn in txns:

backend/kayman/routers/transaction.py

@@ -44,9 +44,17 @@ def create(
 def reads(
     *,
     session: Session = Depends(get_session),
+    transaction_id: int | None = None,
     account_id: int | None = None,
     start: datetime | None = None,
     end: datetime | None = None,
     event_id: int | Literal["empty"] | None = None,
 ) -> Sequence[TransactionBase]:
-    return read_transactions(session, account_id, start, end, event_id=event_id)
+    return read_transactions(
+        session,
+        transaction_id=transaction_id,
+        account_id=account_id,
+        start=start,
+        end=end,
+        event_id=event_id,
+    )

backend/kayman/tests/crud/test_transaction.py

@@ -107,6 +107,16 @@ def test_read_transactions_all(session: Session):
     assert len(read_transactions(session)) == 10
 
 
+def test_read_transactions_by_id(session: Session):
+    txn_1 = TransactionFactory()
+    TransactionFactory()
+
+    results = read_transactions(session, transaction_id=txn_1.id)
+
+    assert len(results) == 1
+    assert results[0].id == txn_1.id
+
+
 def test_read_transactions_by_account(session: Session):
     account_1 = AccountFactory()
     account_2 = AccountFactory()