40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.errors import AppError
|
|
from app.models.enums import OrderStatus
|
|
from app.models.orders import Order
|
|
|
|
|
|
def list_orders_admin(db: Session, status: OrderStatus | None = None) -> list[Order]:
|
|
query = select(Order).order_by(Order.created_at.desc())
|
|
if status is not None:
|
|
query = query.where(Order.status == status)
|
|
return list(db.scalars(query).all())
|
|
|
|
|
|
def get_order_admin(db: Session, order_id: int) -> Order:
|
|
order = db.get(Order, order_id)
|
|
if not order:
|
|
raise AppError("Order not found", status_code=404)
|
|
return order
|
|
|
|
|
|
def list_payments_admin(db: Session) -> list[dict]:
|
|
orders = db.scalars(
|
|
select(Order)
|
|
.where(Order.status.in_([OrderStatus.paid, OrderStatus.pending, OrderStatus.failed]))
|
|
.order_by(Order.created_at.desc())
|
|
).all()
|
|
return [
|
|
{
|
|
"order_id": order.id,
|
|
"event_id": order.event_id,
|
|
"status": order.status,
|
|
"total_paise": order.total_paise,
|
|
"provider_payment_id": order.provider_payment_id,
|
|
"payment_reference": order.payment_reference,
|
|
"paid_at": order.paid_at,
|
|
}
|
|
for order in orders
|
|
]
|