Back cdr : Endpoint to get total of validated payments

app/modules/cdr/cruds_cdr.py

@@ -902,6 +902,84 @@ async def delete_payment(
     )
 
 
+async def get_payment_total_by_seller(
+    db: AsyncSession,
+    cdr_year: int,
+) -> schemas_cdr.TotalPaymentBySeller:
+    result = await db.execute(
+        select(
+            models_cdr.Seller.name,
+            func.sum(
+                models_cdr.ProductVariant.price * models_cdr.Purchase.quantity,
+            ).label("total_amount"),
+        )
+        .join(
+            models_cdr.CdrProduct,
+            models_cdr.Seller.id == models_cdr.CdrProduct.seller_id,
+        )
+        .join(
+            models_cdr.ProductVariant,
+            models_cdr.ProductVariant.product_id == models_cdr.CdrProduct.id,
+        )
+        .join(
+            models_cdr.Purchase,
+            models_cdr.Purchase.product_variant_id == models_cdr.ProductVariant.id,
+        )
+        .where(
+            models_cdr.Purchase.validated,
+            models_cdr.ProductVariant.year == cdr_year,
+        )
+        .group_by(models_cdr.Seller.id),
+    )
+
+    return schemas_cdr.TotalPaymentBySeller(
+        total_amounts=[
+            schemas_cdr.TotalPaymentOfSeller(
+                total_amount=row.total_amount,
+                name=row.name,
+            )
+            for row in result.all()
+        ],
+    )
+
+
+async def get_payment_total_by_type(
+    db: AsyncSession,
+    cdr_year: int,
+) -> list[schemas_cdr.PaymentBase]:
+    result = await db.execute(
+        select(
+            models_cdr.Payment.payment_type,
+            func.sum(models_cdr.Payment.total).label("total"),
+        )
+        .where(models_cdr.Payment.year == cdr_year)
+        .group_by(models_cdr.Payment.payment_type),
+    )
+
+    return [
+        schemas_cdr.PaymentBase(
+            total=row.total,
+            payment_type=row.payment_type,
+        )
+        for row in result.all()
+    ]
+
+
+async def get_payment_total(
+    db: AsyncSession,
+    cdr_year: int,
+) -> schemas_cdr.TotalPayment:
+
+    result = await db.execute(
+        select(
+            func.sum(models_cdr.Payment.total).label("total"),
+        ).where(models_cdr.Payment.year == cdr_year),
+    )
+    total = result.scalar()
+
+    return schemas_cdr.TotalPayment(total_amount=total if total is not None else 0)
+
+
 def create_action(
     db: AsyncSession,
     action: models_cdr.CdrAction,

app/modules/cdr/endpoints_cdr.py

@@ -191,6 +191,75 @@ async def get_cdr_users_pending_validation(
     ]
 
 
+@module.router.get(
+    "/cdr/stats/payment_total_by_seller/",
+    response_model=schemas_cdr.TotalPaymentBySeller,
+    status_code=200,
+)
+async def get_payment_total_by_seller(
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(
+        is_user_allowed_to([CdrPermissions.manage_cdr]),
+    ),
+    cdr_year: coredata_cdr.CdrYear = Depends(get_current_cdr_year),
+):
+    """
+    Get the total of all payments made in the CDR for each seller.
+
+    **User must be a CDR Admin to use this endpoint**
+    """
+    return await cruds_cdr.get_payment_total_by_seller(
+        db=db,
+        cdr_year=cdr_year.year,
+    )
+
+
+@module.router.get(
+    "/cdr/stats/payment_total_per_type/",
+    response_model=list[schemas_cdr.PaymentBase],
+    status_code=200,
+)
+async def get_payment_total_per_type(
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(
+        is_user_allowed_to([CdrPermissions.manage_cdr]),
+    ),
+    cdr_year: coredata_cdr.CdrYear = Depends(get_current_cdr_year),
+):
+    """
+    Get the total of all payments made in the CDR for each payment type.
+
+    **User must a CDR Admin to use this endpoint**
+    """
+    return await cruds_cdr.get_payment_total_by_type(
+        db=db,
+        cdr_year=cdr_year.year,
+    )
+
+
+@module.router.get(
+    "/cdr/stats/payment_total/",
+    response_model=schemas_cdr.TotalPayment,
+    status_code=200,
+)
+async def get_payment_total(
+    db: AsyncSession = Depends(get_db),
+    user: models_users.CoreUser = Depends(
+        is_user_allowed_to([CdrPermissions.manage_cdr]),
+    ),
+    cdr_year: coredata_cdr.CdrYear = Depends(get_current_cdr_year),
+):
+    """
+    Get the total of all payments made in the CDR.
+
+    **User must a CDR Admin to use this endpoint**
+    """
+    return await cruds_cdr.get_payment_total(
+        db=db,
+        cdr_year=cdr_year.year,
+    )
+
+
 @module.router.get(
     "/cdr/users/{user_id}/",
     response_model=schemas_cdr.CdrUser,

app/modules/cdr/schemas_cdr.py

@@ -341,3 +341,16 @@ class BatchValidation(BaseModel):
     user_emails: list[str]
     product_variant_id: UUID
     validated: bool
+
+
+class TotalPaymentOfSeller(BaseModel):
+    name: str
+    total_amount: int
+
+
+class TotalPaymentBySeller(BaseModel):
+    total_amounts: list[TotalPaymentOfSeller]
+
+
+class TotalPayment(BaseModel):
+    total_amount: int

tests/commons.py

@@ -44,6 +44,10 @@ class FailedToAddObjectToDB(Exception):
     """Exception raised when an object cannot be added to the database."""
 
 
+class FailedToUpdateObjectInDB(Exception):
+    """Exception raised when an object cannot be updated in the database."""
+
+
 async def override_init_state(
     app: FastAPI,
     settings: Settings,
@@ -327,6 +331,21 @@ async def add_object_to_db(db_object: Base) -> None:
             await db.close()
 
 
+async def update_object_in_db(db_object: Base) -> None:
+    """
+    Update an existing object in the database
+    """
+    async with TestingSessionLocal() as db:
+        try:
+            await db.merge(db_object)
+            await db.commit()
+        except Exception as error:
+            await db.rollback()
+            raise FailedToUpdateObjectInDB from error
+        finally:
+            await db.close()
+
+
 async def add_coredata_to_db(
     core_data: core_data.BaseCoreData,
 ) -> None:

tests/modules/cdr/test_cdr.py

@@ -28,6 +28,7 @@
     create_user_with_groups,
     get_TestingSessionLocal,
     mocked_checkout_id,
+    update_object_in_db,
 )
 
 year = datetime.now(UTC).year
@@ -479,6 +480,77 @@ async def init_objects():
     await add_object_to_db(ticket)
 
 
+def test_get_payment_total_as_admin(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total/",
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+    assert response.status_code == 200
+    assert response.json()["total_amount"] == 5000
+
+
+def test_get_payment_total_as_user(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total/",
+        headers={"Authorization": f"Bearer {token_user}"},
+    )
+    assert response.status_code == 403
+
+
+async def test_get_payment_total_by_seller_as_admin(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total_by_seller/",
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+    assert response.status_code == 200
+    assert response.json() == {"total_amounts": []}
+
+    purchase.validated = True
+    await update_object_in_db(purchase)
+    response = client.get(
+        "/cdr/stats/payment_total_by_seller/",
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+
+    data = response.json()
+    assert len(data["total_amounts"]) == 1
+    first_data = data["total_amounts"][0]
+    assert first_data["name"] == "BDE"
+    assert first_data["total_amount"] == 100
+
+    purchase.validated = False
+    await update_object_in_db(purchase)
+
+
+def test_get_payment_total_by_seller_as_user(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total_by_seller/",
+        headers={"Authorization": f"Bearer {token_user}"},
+    )
+    assert response.status_code == 403
+
+
+def test_get_payment_total_per_type_as_admin(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total_per_type/",
+        headers={"Authorization": f"Bearer {token_admin}"},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert len(data) == 1
+    first_data = data[0]
+    assert first_data["payment_type"] == "cash"
+    assert first_data["total"] == 5000
+
+
+def test_get_payment_total_per_type_as_user(client: TestClient):
+    response = client.get(
+        "/cdr/stats/payment_total_per_type/",
+        headers={"Authorization": f"Bearer {token_user}"},
+    )
+    assert response.status_code == 403
+
+
 def test_get_all_cdr_users_seller(client: TestClient):
     response = client.get(
         "/cdr/users/",