45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
# -*- coding = utf-8 -*-
|
|
# @Time : 2025/2/17 上午1:00
|
|
# @File : services.py
|
|
# @Software : PyCharm
|
|
# @Author : xingc
|
|
# @Desc :
|
|
from typing import Dict
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.db.crud import CRUD
|
|
from core.db.models import SearchCardCache, User, UserCollection
|
|
from core.routers.auth.schema import LoginUser
|
|
|
|
from config import settings
|
|
|
|
|
|
async def get_cache_card_by_number(db_session: AsyncSession, *, cert_number: str) -> SearchCardCache | None:
|
|
query = select(SearchCardCache).where(SearchCardCache.cert_number == cert_number)
|
|
result = await db_session.execute(query)
|
|
card: SearchCardCache | None = result.scalar_one_or_none()
|
|
|
|
return card
|
|
|
|
|
|
async def check_user_card_created_permission(
|
|
db_session: AsyncSession, *, login_user: LoginUser, other_conditions: Dict = None
|
|
) -> bool:
|
|
if other_conditions is None:
|
|
other_conditions = {}
|
|
|
|
db_crud = CRUD(db_session, User)
|
|
user = await db_crud.get(filters={'user_id': login_user.user_id})
|
|
if user.is_vip():
|
|
return True
|
|
|
|
total = await db_crud.total(
|
|
model=UserCollection, filters={
|
|
'user_id': login_user.user_id,
|
|
**other_conditions
|
|
}
|
|
)
|
|
return settings.USER_MAX_CREATED_CARD_NUM > total
|