xzbautomarkapi/admin/api/endpoints/school/utils.py

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

from typing import Dict, Union, List, Any, Optional

from sqlalchemy.ext.asyncio import AsyncSession

from crud.school import crud_school, crud_class, crud_grade
from crud.sysdata.role import crud_role
from crud.user import crud_teacher, crud_student

CRUD_DICT = {
    "school": crud_school,
    "grade": crud_grade,
    "class": crud_class,
    "teacher": crud_teacher,
    "student": crud_student,
    "role": crud_role
}


async def is_existed(db: AsyncSession, obj_type: str, *,
                     filters: Union[List[Any], Dict[str, Any]]):
    if obj_type not in CRUD_DICT:
        return None
    obj = await CRUD_DICT[obj_type].find_one(db, filters=filters)
    return obj


async def check_row(values: List[Any], length: int) -> Any:
    _val = []
    for x in values[:length]:
        if isinstance(x, str):
            _val.append(x.strip())
        else:
            _val.append(x)
    if not _val:
        return None
    elif len(_val) != length:
        return False
    return _val