128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status, Query
|
||
from sqlalchemy.orm import Session
|
||
from typing import List, Literal, TypedDict
|
||
|
||
from app.core.database import get_db
|
||
from app.core.auth import get_current_user, requires_role
|
||
from app.models.user import User
|
||
from app.models.topup import Topup
|
||
from app.models.booking import Booking
|
||
|
||
router = APIRouter(prefix="/transactions", tags=["transactions"])
|
||
|
||
|
||
class TransactionOut(TypedDict):
|
||
id: str
|
||
type: Literal["topup", "booking"]
|
||
amount_cents: int
|
||
created_at: str # ISO
|
||
|
||
|
||
@router.get("/me")
|
||
def list_my_transactions(
|
||
limit: int = Query(100, ge=1, le=1000),
|
||
offset: int = Query(0, ge=0),
|
||
db: Session = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""
|
||
Kombiniert Topups (+) und Buchungen (−) des aktuellen Users.
|
||
"""
|
||
topups = (
|
||
db.query(Topup)
|
||
.filter(Topup.user_id == current_user.id)
|
||
.order_by(Topup.id.desc())
|
||
.limit(limit)
|
||
.offset(offset)
|
||
.all()
|
||
)
|
||
bookings = (
|
||
db.query(Booking)
|
||
.filter(Booking.user_id == current_user.id)
|
||
.order_by(Booking.id.desc())
|
||
.limit(limit)
|
||
.offset(offset)
|
||
.all()
|
||
)
|
||
|
||
tx: List[TransactionOut] = []
|
||
|
||
for t in topups:
|
||
tx.append(
|
||
{
|
||
"id": f"topup-{t.id}",
|
||
"type": "topup",
|
||
"amount_cents": int(t.amount_cents or 0),
|
||
"created_at": t.created_at.isoformat() if hasattr(t, "created_at") and t.created_at else "",
|
||
}
|
||
)
|
||
|
||
for b in bookings:
|
||
# Wenn du Booking.total_cents hast, nutze das. Sonst product.price_cents * amount.
|
||
total = getattr(b, "total_cents", None)
|
||
if total is None:
|
||
price = getattr(b, "price_cents", None)
|
||
if price is None and hasattr(b, "product") and b.product:
|
||
price = getattr(b.product, "price_cents", 0)
|
||
total = (price or 0) * (getattr(b, "amount", 1) or 1)
|
||
tx.append(
|
||
{
|
||
"id": f"booking-{b.id}",
|
||
"type": "booking",
|
||
"amount_cents": -int(total or 0), # Buchungen als negative Beträge
|
||
"created_at": b.created_at.isoformat() if hasattr(b, "created_at") and b.created_at else "",
|
||
}
|
||
)
|
||
|
||
# Absteigend nach Datum
|
||
tx.sort(key=lambda x: x["created_at"], reverse=True)
|
||
# einfache Paginierung nach Sortierung (optional): tx = tx[offset:offset+limit]
|
||
return tx
|
||
|
||
|
||
@router.get("/", dependencies=[Depends(requires_role("manager", "admin"))])
|
||
def list_all_transactions_admin(
|
||
limit: int = Query(200, ge=1, le=2000),
|
||
offset: int = Query(0, ge=0),
|
||
db: Session = Depends(get_db),
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
"""
|
||
Admin/Manager: kombinierte Übersicht (einfach, nicht performant für große Datenmengen).
|
||
"""
|
||
topups = (
|
||
db.query(Topup).order_by(Topup.id.desc()).limit(limit).offset(offset).all()
|
||
)
|
||
bookings = (
|
||
db.query(Booking).order_by(Booking.id.desc()).limit(limit).offset(offset).all()
|
||
)
|
||
|
||
tx: List[TransactionOut] = []
|
||
for t in topups:
|
||
tx.append(
|
||
{
|
||
"id": f"topup-{t.id}",
|
||
"type": "topup",
|
||
"amount_cents": int(t.amount_cents or 0),
|
||
"created_at": t.created_at.isoformat() if hasattr(t, "created_at") and t.created_at else "",
|
||
}
|
||
)
|
||
for b in bookings:
|
||
total = getattr(b, "total_cents", None)
|
||
if total is None:
|
||
price = getattr(b, "price_cents", None)
|
||
if price is None and hasattr(b, "product") and b.product:
|
||
price = getattr(b.product, "price_cents", 0)
|
||
total = (price or 0) * (getattr(b, "amount", 1) or 1)
|
||
tx.append(
|
||
{
|
||
"id": f"booking-{b.id}",
|
||
"type": "booking",
|
||
"amount_cents": -int(total or 0),
|
||
"created_at": b.created_at.isoformat() if hasattr(b, "created_at") and b.created_at else "",
|
||
}
|
||
)
|
||
|
||
tx.sort(key=lambda x: x["created_at"], reverse=True)
|
||
return tx
|