43 lines
1.6 KiB
Python
43 lines
1.6 KiB
Python
from sqlalchemy.orm import Session
|
|
|
|
from app.common.errors.app_error import NotFoundAppError
|
|
from app.modules.payments.repository import PaymentsRepository
|
|
from app.modules.wallets.service import WalletService
|
|
|
|
|
|
class PaymentsService:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
self.repository = PaymentsRepository(db)
|
|
self.wallet_service = WalletService(db)
|
|
|
|
def repair_order(self, order_id: int) -> dict:
|
|
order = self.repository.get_order_by_id(order_id)
|
|
if not order:
|
|
raise NotFoundAppError("order not found", code=30001)
|
|
return self.wallet_service.handle_mock_payment(order.order_no)
|
|
|
|
def list_orders(self) -> list[dict]:
|
|
rows = self.repository.list_orders().limit(200).all()
|
|
return [self.serialize_order(item) for item in rows]
|
|
|
|
def get_order_detail(self, order_id: int) -> dict:
|
|
order = self.repository.get_order_by_id(order_id)
|
|
if not order:
|
|
raise NotFoundAppError("order not found", code=30001)
|
|
return self.serialize_order(order)
|
|
|
|
@staticmethod
|
|
def serialize_order(order) -> dict:
|
|
return {
|
|
"id": order.id,
|
|
"orderNo": order.order_no,
|
|
"userId": order.user_id,
|
|
"payAmount": f"{order.pay_amount:.2f}",
|
|
"arrivalPoints": order.arrival_points,
|
|
"status": order.status,
|
|
"paymentChannelCode": order.payment_channel_code,
|
|
"paidAt": order.paid_at.isoformat() if order.paid_at else None,
|
|
"createdAt": order.created_at.isoformat(),
|
|
}
|