44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.entities import RechargeOrder, RechargePlan, RedeemCode, Wallet, WalletTransaction
|
|
|
|
|
|
class WalletRepository:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def lock_wallet(self, user_id: int) -> Wallet:
|
|
return self.db.execute(
|
|
select(Wallet).where(Wallet.user_id == user_id).with_for_update()
|
|
).scalar_one()
|
|
|
|
def wallet_transactions(self, user_id: int):
|
|
return self.db.query(WalletTransaction).filter(WalletTransaction.user_id == user_id)
|
|
|
|
def recharge_orders(self, user_id: int):
|
|
return self.db.query(RechargeOrder).filter(RechargeOrder.user_id == user_id)
|
|
|
|
def get_recharge_plan(self, plan_id: int) -> RechargePlan | None:
|
|
return self.db.scalar(select(RechargePlan).where(RechargePlan.id == plan_id))
|
|
|
|
def get_order_by_no(self, order_no: str) -> RechargeOrder | None:
|
|
return self.db.scalar(select(RechargeOrder).where(RechargeOrder.order_no == order_no))
|
|
|
|
def lock_redeem_code(self, redeem_code: str) -> RedeemCode | None:
|
|
return self.db.execute(
|
|
select(RedeemCode)
|
|
.where(RedeemCode.redeem_code == redeem_code)
|
|
.with_for_update()
|
|
).scalar_one_or_none()
|
|
|
|
def latest_transactions_count(self, user_id: int) -> int:
|
|
return (
|
|
self.db.query(func.count(WalletTransaction.id))
|
|
.filter(WalletTransaction.user_id == user_id)
|
|
.scalar()
|
|
)
|
|
|