Adding filters et pagination to ledger

api/app/ledger/resource.py

@@ -1,27 +1,26 @@
 from sqlalchemy import and_, select, func
 from sqlalchemy.orm import aliased
 
-from ledger.schema import TransactionLedgerRead
 from transaction.models import Split, Transaction
 
-
-
 class LedgerResource:
     @classmethod
-    def get_ledger(cls, session, account_id):
+    def get_ledger(cls, account_id, filters):
         split_account = aliased(Split)
         split_balance = aliased(Split)
         transaction_balance = aliased(Transaction)
 
-        balance_stmt = select(func.sum(split_balance.amount)).join(transaction_balance).where(and_(split_balance.account_id == split_account.account_id, transaction_balance.sequence <= Transaction.sequence)).scalar_subquery()
+        balance_stmt = select(func.sum(split_balance.amount)) \
+            .join(transaction_balance) \
+            .where(and_(
+                split_balance.account_id == split_account.account_id,
+                transaction_balance.sequence <= Transaction.sequence)
+            ).scalar_subquery()
 
-        stmt = (select(
+        stmt = select(Transaction,split_account,balance_stmt.label('balance')) \
-            Transaction,
+            .join(
-            split_account,
+                split_account,
-            balance_stmt.label('balance')
+                and_(Transaction.id == split_account.transaction_id, split_account.account_id == account_id)
-        ).join(
+            )
-            split_account,
-            and_(Transaction.id == split_account.transaction_id, split_account.account_id == account_id)
-        ))
 
-        return [TransactionLedgerRead(transaction=transaction, account_split=split, balance=balance) for transaction, split, balance in session.exec(stmt).all()]
+        return filters.filter(stmt)

api/app/ledger/routes.py

@@ -1,15 +1,31 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Depends
+from fastapi_filter import FilterDepends
+from fastapi_filter.contrib.sqlalchemy import Filter
+from fastapi_pagination import Page
+from fastapi_pagination.ext.sqlalchemy import paginate
+from fastapi_pagination.utils import disable_installed_extensions_check
 
 from db import SessionDep
 from ledger.resource import LedgerResource
 from ledger.schema import TransactionLedgerRead
-
+from transaction.models import Transaction
 from user.manager import get_current_user
 
 router = APIRouter()
 
+class LedgerFilters(Filter):
+
+    class Constants(Filter.Constants):
+        model = Transaction
+        search_model_fields = ["id", "sequence"]
+disable_installed_extensions_check()
+
 @router.get("/{account_id}")
-def read_ledger(account_id: UUID, session: SessionDep, current_user=Depends(get_current_user)) -> list[TransactionLedgerRead]:
+def read_ledger(account_id: UUID, session: SessionDep, filters: LedgerFilters = FilterDepends(LedgerFilters), current_user=Depends(get_current_user)) -> Page[TransactionLedgerRead]:
-    return LedgerResource.get_ledger(session, account_id)
+    return paginate(
+        session,
+        LedgerResource.get_ledger(account_id, filters),
+        transformer=lambda items: [TransactionLedgerRead(transaction=transaction, account_split=split, balance=balance) for transaction, split, balance in items],
+    )