xzbautomarkapi/common/endpoints/search.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import datetime

from fastapi import APIRouter, Depends, Query
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from common.const import PERIODS, SUBJECTS
from crud.school import crud_school, crud_grade, crud_class, crud_system
from crud.sysdata.region import crud_region
from crud.sysdata.role import crud_role
from models.user import SysUser
from schemas.base import ReturnField
from schemas.school.school import (SchoolItemList, GradeItemList,
                                   SchoolClassList, SubjectList, PeriodList,
                                   YearsList, SchoolSystemList)
from schemas.sysdata.region import RegionItemList
from schemas.sysdata.role import RoleList
from utils.depends import get_current_user, get_async_db

router = APIRouter()


@router.get("/schools/search",
            summary="学校列表",
            response_model=SchoolItemList,
            response_model_exclude_none=True)
async def get_schools_search(name: str = "",
                             res: ReturnField = Query(
                                 "", description="控制返回字段,字段逗号分隔"),
                             db: AsyncSession = Depends(get_async_db),
                             current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    items = await crud_school.fetch_all(db, filters=filters, return_fields=res)
    return {"data": items}


@router.get("/grades/search",
            summary="年级列表",
            response_model=GradeItemList,
            response_model_exclude_none=True)
async def get_grades_search(name: str = "",
                            sid: int = 0,
                            res: ReturnField = Query(
                                "", description="控制返回字段,字段逗号分隔"),
                            db: AsyncSession = Depends(get_async_db),
                            current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    if sid:
        filters.append(text(f"school_id = {sid}"))
    items = await crud_grade.fetch_all(db, filters=filters, return_fields=res)
    return {"data": items}


@router.get("/classes/search",
            summary="班级列表",
            response_model=SchoolClassList,
            response_model_exclude_none=True)
async def get_classes_search(name: str = "",
                             gid: int = 0,
                             res: ReturnField = Query(
                                 "", description="控制返回字段,字段逗号分隔"),
                             db: AsyncSession = Depends(get_async_db),
                             current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    if gid:
        filters.append(text(f"grade_id = {gid}"))
    items = await crud_class.fetch_all(db, filters=filters, return_fields=res)
    return {"data": items}


@router.get("/roles/search",
            summary="职务列表",
            response_model=RoleList,
            response_model_exclude_none=True)
async def get_roles_search(name: str = "",
                           platform: str = "pc",
                           res: ReturnField = Query(
                               "", description="控制返回字段,字段逗号分隔"),
                           db: AsyncSession = Depends(get_async_db),
                           current_user: SysUser = Depends(get_current_user)):
    filters = []
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    if platform:
        filters.append(text(f"platform = '{platform}'"))
    items = await crud_role.fetch_all(db, filters=filters, return_fields=res)
    return {"data": items}


@router.get("/subjects/search",
            summary="学科列表",
            response_model=SubjectList,
            response_model_exclude_none=True)
async def get_subjects_search(
        current_user: SysUser = Depends(get_current_user)):
    data = [{"name": item} for item in SUBJECTS]
    return {"data": data}


@router.get("/periods/search",
            summary="学段列表",
            response_model=PeriodList,
            response_model_exclude_none=True)
async def get_periods_search(current_user: SysUser = Depends(get_current_user)):
    data = [{"name": item} for item in PERIODS]
    return {"data": data}


@router.get("/schoolsystem/search",
            summary="学制列表",
            response_model=SchoolSystemList,
            response_model_exclude_none=True)
async def get_system_search(db: AsyncSession = Depends(get_async_db),
                            current_user: SysUser = Depends(get_current_user)):
    _, items = await crud_system.find_all(db, filters={})
    return {"data": items}


@router.get("/years/search",
            summary="年份列表",
            response_model=YearsList,
            response_model_exclude_none=True)
async def get_years_search(current_user: SysUser = Depends(get_current_user)):
    now = datetime.datetime.now().year + 1
    data = [{"name": item} for item in range(now - 10, now)]
    data.reverse()
    return {"data": data}


@router.get("/regions/search",
            summary="地区列表(省份列表)",
            response_model=RegionItemList,
            response_model_exclude_none=True)
async def get_regions_search(name: str = "",
                             res: ReturnField = Query(
                                 "", description="控制返回字段,字段逗号分隔"),
                             db: AsyncSession = Depends(get_async_db),
                             current_user: SysUser = Depends(get_current_user)):
    filters = [text("level = 1")]
    if name:
        filters.append(text(f"name LIKE '%{name}%'"))
    items = await crud_region.fetch_all(db, filters=filters, return_fields=res)
    return {"data": items, "total": None}