51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.entities import (
|
|
AdminUser,
|
|
InviteRelation,
|
|
RechargeOrder,
|
|
User,
|
|
VideoGenerationTask,
|
|
)
|
|
|
|
|
|
class AdminsRepository:
|
|
def __init__(self, db: Session) -> None:
|
|
self.db = db
|
|
|
|
def get_admin_by_username(self, username: str) -> AdminUser | None:
|
|
return self.db.scalar(select(AdminUser).where(AdminUser.username == username))
|
|
|
|
def list_users(self):
|
|
return self.db.query(User).order_by(User.id.desc())
|
|
|
|
def get_user(self, user_id: int) -> User | None:
|
|
return self.db.scalar(select(User).where(User.id == user_id))
|
|
|
|
def count_users(self) -> int:
|
|
return self.db.query(func.count(User.id)).scalar() or 0
|
|
|
|
def count_paid_orders(self) -> int:
|
|
return (
|
|
self.db.query(func.count(RechargeOrder.id))
|
|
.filter(RechargeOrder.status == "paid")
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
def count_tasks(self) -> int:
|
|
return self.db.query(func.count(VideoGenerationTask.id)).scalar() or 0
|
|
|
|
def count_success_tasks(self) -> int:
|
|
return (
|
|
self.db.query(func.count(VideoGenerationTask.id))
|
|
.filter(VideoGenerationTask.task_status == "succeeded")
|
|
.scalar()
|
|
or 0
|
|
)
|
|
|
|
def invite_relations(self):
|
|
return self.db.query(InviteRelation).order_by(InviteRelation.id.desc())
|
|
|