login page

backend/.gitignore

@@ -0,0 +1,142 @@
+### Python template
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+.idea
+.pdm.toml
+__pypackages__/

backend/controller/__init__.py

@@ -0,0 +1,17 @@
+from fastapi import Depends, FastAPI
+
+from core.security import check_token
+
+
+def register_routers(app: FastAPI):
+    from controller.common import common
+    from controller.menu import menu
+    from controller.role import role
+    from controller.user import user
+
+    app.include_router(router=common)
+    app.include_router(
+        router=user,
+    )
+    app.include_router(router=menu)
+    app.include_router(router=role)

backend/controller/common.py

@@ -0,0 +1,16 @@
+from core.resp import Response
+from core.router import Router
+from core.security import generate_token, verify_password
+from dbhelper.user import get_user
+from schemas.common import LoginForm, LoginResult
+
+common = Router(tags=["公共接口"])
+
+
+@common.post("/login", summary="登录")
+async def login(auth_data: LoginForm) -> Response[LoginResult]:
+    user_obj = await get_user({"username": auth_data.username})
+    if user_obj:
+        if verify_password(auth_data.password, user_obj.password):
+            return dict(id=user_obj.id, access_token=generate_token(user_obj.username))
+    return Response(msg="账号或密码错误")

backend/controller/menu.py

@@ -0,0 +1,31 @@
+from core.resp import Response
+from core.router import Router
+from schemas.common import QueryData
+from schemas.menu import MenuIn, MenuRead
+
+menu = Router(prefix="/menu", tags=["菜单管理"])
+
+
+@menu.post("", summary="菜单添加")
+async def menu_add(data: MenuIn) -> Response[MenuRead]:
+    pass
+
+
+@menu.get("/{pk}", summary="菜单详情")
+async def menu_info(pk: int) -> Response[MenuRead]:
+    pass
+
+
+@menu.delete("/{pk}", summary="删除菜单")
+async def menu_del(pk: int) -> Response:
+    pass
+
+
+@menu.put("/{pk}", summary="编辑菜单")
+async def menu_put(pk: int, data: MenuIn) -> Response[MenuRead]:
+    pass
+
+
+@menu.post("/list", summary="查询菜单列表")
+async def menu_list(data: QueryData) -> Response[list[MenuRead]]:
+    pass

backend/controller/role.py

@@ -0,0 +1,53 @@
+import json
+
+from core.resp import Response
+from core.router import Router
+from core.utils import list_to_tree
+from dbhelper.role import get_role_menus
+from schemas.common import QueryData
+from schemas.role import RoleAdd, RoleInfo
+
+role = Router(prefix="/role", tags=["角色管理"])
+
+
+@role.post("", summary="角色添加")
+async def role_add(data: RoleAdd) -> Response[RoleInfo]:
+    pass
+
+
+@role.get("/{pk}", summary="角色详情")
+async def role_info(pk: int) -> Response[RoleInfo]:
+    pass
+
+
+@role.delete("/{pk}", summary="删除角色")
+async def role_del(pk: int) -> Response:
+    pass
+
+
+@role.put("/{pk}", summary="编辑角色")
+async def role_put(pk: int, data: RoleAdd) -> Response[RoleInfo]:
+    pass
+
+
+@role.post("/list", summary="查询角色列表")
+async def role_list(data: QueryData) -> Response[list[RoleInfo]]:
+    pass
+
+
+@role.get("/{pk}/menu", summary="查询角色菜单权限")
+async def role_menu(pk: int):
+    menus = await get_role_menus(pk)
+    for obj in menus:
+        obj["meta"] = json.loads(obj["meta"]) if obj["meta"] is not None else None
+    return Response(data=list_to_tree(menus))
+
+
+@role.get("/{pk}/menuIds", summary="查询角色菜单ids")
+async def role_menus_id():
+    pass
+
+
+@role.get("/assign", summary="分配权限")
+async def role_assign():
+    pass

backend/controller/user.py

@@ -0,0 +1,41 @@
+from core.resp import Response
+from core.router import Router
+from dbhelper.user import get_user_info, get_users, insert_user
+from schemas.common import ListAll
+from schemas.user import UserAdd, UserInfo, UserList, UserQuery
+
+user = Router(prefix="/users", tags=["用户管理"])
+
+
+@user.post("", summary="用户添加")
+async def user_add(data: UserAdd) -> Response[UserInfo]:
+    roles = data.rids
+    del data.rids
+    return await insert_user(data, roles)
+
+
+@user.get("/{pk}", summary="用户详情")
+async def user_info(pk: int) -> Response[UserInfo]:
+    try:
+        return Response(data=await get_user_info(pk))
+    except Exception as e:
+        return Response(msg=f"用户不存在 {e}")
+
+
+@user.delete("/{pk}", summary="删除用户")
+async def user_del(pk: int) -> Response:
+    pass
+
+
+@user.put("/{pk}", summary="编辑用户")
+async def user_put(pk: int, data: UserAdd) -> Response[UserInfo]:
+    pass
+
+
+@user.post("/list", summary="查询用户列表")
+async def user_list(query: UserQuery) -> Response[ListAll[UserList]]:
+    limit = query.size
+    skip = (query.offset - 1) * limit
+    del query.offset, query.size
+    users, count = await get_users(skip, limit, query.dict())
+    return Response(data=ListAll(total=count, items=users))

backend/core/enums.py

@@ -0,0 +1,35 @@
+import enum
+
+
+class Status(enum.IntEnum):
+    """
+    数据库状态枚举值
+    9 删除 5 无效 1 有效 3 使用
+    """
+
+    DELETED: int = 9
+    INACTIVE: int = 5
+    ACTIVE: int = 1
+    SELECTED: int = 3
+
+
+class UserType(enum.IntEnum):
+    """
+    数据库超级管理员枚举
+    0 超级管理员 1用户
+    """
+
+    ADMIN: int = 0
+    USER: int = 1
+
+
+class MenuType(enum.IntEnum):
+    """
+    菜单类型枚举
+    目录 0
+    组件 1 按钮 2
+    """
+
+    DIR = 0
+    CPN = 1
+    BTN = 2

backend/core/events.py

@@ -0,0 +1,13 @@
+from tortoise import Tortoise
+
+
+async def init_orm():
+    """初始化orm"""
+    await Tortoise.init(
+        db_url="sqlite://mini.db", modules={"models": ["models"]}
+    )
+
+
+async def close_orm():
+    """关闭orm"""
+    await Tortoise.close_connections()

backend/core/exceptions.py

@@ -0,0 +1,6 @@
+from fastapi.exceptions import HTTPException
+
+
+class TokenAuthFailure(HTTPException):
+    status_code = 401
+    detail = "认证失败"

backend/core/log.py

@@ -0,0 +1,15 @@
+import logging
+import sys
+
+fmt = logging.Formatter(
+    fmt="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+sh = logging.StreamHandler(sys.stdout)
+sh.setLevel(logging.DEBUG)
+sh.setFormatter(fmt)
+
+# will print debug sql
+logger_db_client = logging.getLogger("mini-rbac")
+logger_db_client.setLevel(logging.DEBUG)
+logger_db_client.addHandler(sh)

backend/core/middleware.py

@@ -0,0 +1,12 @@
+from fastapi.middleware import Middleware
+from fastapi.middleware.cors import CORSMiddleware
+
+middlewares = [
+    Middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+]

backend/core/resp.py

@@ -0,0 +1,32 @@
+import enum
+from typing import Generic, Optional, TypeVar, Union
+
+from pydantic.generics import GenericModel
+
+T = TypeVar("T")
+
+
+class Status(enum.IntEnum):
+    OK = 200
+    CREATED = 201
+    ACCEPTED = 202
+    NO_CONTENT = 204
+    BAD_REQUEST = 400
+    UNAUTHORIZED = 401
+    FORBIDDEN = 403
+    NOT_FOUND = 404
+    INTERNAL_SERVER_ERROR = 500
+    NOT_IMPLEMENTED = 501
+    BAD_GATEWAY = 502
+    SERVICE_UNAVAILABLE = 503
+
+
+class Msg(enum.Enum):
+    OK = "OK"
+    FAIL = "FAIL"
+
+
+class Response(GenericModel, Generic[T]):
+    code: Status = Status.OK
+    msg: Union[Msg, str] = Msg.OK
+    data: Optional[T]

backend/core/router.py

@@ -0,0 +1,48 @@
+from typing import TYPE_CHECKING, Any, Callable, get_type_hints
+
+from fastapi import APIRouter
+from fastapi.routing import APIRoute
+
+"""
+根据类型标注自动返回模型
+https://github.com/tiangolo/fastapi/issues/620
+"""
+
+
+class Router(APIRouter):
+    """
+    装饰器用法
+    @app.get("/")
+    def index() -> User:
+    """
+
+    if not TYPE_CHECKING:
+
+        def add_api_route(
+            self, path: str, endpoint: Callable[..., Any], **kwargs: Any
+        ) -> None:
+            if kwargs.get("response_model") is None:
+                kwargs["response_model"] = get_type_hints(endpoint).get("return")
+            return super().add_api_route(path, endpoint, **kwargs)
+
+    else:  # pragma: no cover
+        pass
+
+
+class Route(APIRoute):
+    """
+    Django挂载视图方法
+    def index() -> User:
+        pass
+    Route("/", endpoint=index)
+    """
+
+    if not TYPE_CHECKING:
+
+        def __init__(self, path: str, endpoint: Callable[..., Any], **kwargs: Any):
+            if kwargs.get("response_model") is None:
+                kwargs["response_model"] = get_type_hints(endpoint).get("return")
+            super(Route, self).__init__(path=path, endpoint=endpoint, **kwargs)
+
+    else:  # pragma: no cover
+        pass

backend/core/security.py

@@ -0,0 +1,68 @@
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import Depends
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+from core.exceptions import TokenAuthFailure
+from dbhelper.user import get_user
+
+# JWT
+SECRET_KEY = "lLNiBWPGiEmCLLR9kRGidgLY7Ac1rpSWwfGzTJpTmCU"
+
+ALGORITHM = "HS256"
+
+ACCESS_TOKEN_EXPIRE_MINUTES = 60 *24 * 7
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+bearer = HTTPBearer()
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """
+    验证明文密码 vs hash密码
+    :param plain_password: 明文密码
+    :param hashed_password: hash密码
+    :return:
+    """
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str) -> str:
+    """
+    加密明文
+    :param password: 明文密码
+    :return:
+    """
+    return pwd_context.hash(password)
+
+
+def generate_token(username: str, expires_delta: Optional[timedelta] = None):
+    """生成token"""
+    to_encode = {"sub": username}.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(
+            minutes=ACCESS_TOKEN_EXPIRE_MINUTES
+        )
+    to_encode.update(dict(exp=expire))
+    encoded_jwt = jwt.encode(
+        to_encode, SECRET_KEY, algorithm=ALGORITHM
+    )
+    return encoded_jwt
+
+
+async def check_token(security: HTTPAuthorizationCredentials = Depends(bearer)):
+    """检查用户token"""
+    token = security.credentials
+    try:
+        payload = jwt.decode(
+            token, SECRET_KEY, algorithms=[ALGORITHM]
+        )
+        username: str = payload.get("sub")
+        return await get_user({"username": username})
+    except JWTError:
+        raise TokenAuthFailure

backend/core/table.py

@@ -0,0 +1,19 @@
+from tortoise import fields, models
+
+from core.enums import Status
+
+
+class Table(models.Model):
+    """
+    抽象模型
+    """
+
+    id = fields.IntField(pk=True, description="主键")
+    status = fields.IntEnumField(Status, description="状态", default=Status.ACTIVE)
+    created = fields.DatetimeField(auto_now_add=True, description="创建时间", null=True)
+    modified = fields.DatetimeField(auto_now=True, description="更新时间", null=True)
+
+    class Meta:
+        abstract = True
+        ordering = ["-created"]
+        indexes = ("status",)

backend/core/utils.py

@@ -0,0 +1,51 @@
+def list_to_tree(
+    menus, parent_flag: str = "pid", children_key: str = "children"
+) -> list:
+    """
+    list 结构转 树结构
+    :param menus: [{id:1, pid: 3}]
+    :param parent_flag: 节点关系字段
+    :param children_key: 生成树结构的子节点字段
+    :return: list 类型的 树嵌套数据
+    """ ""
+    # 先转成字典 id作为key, 数据作为value
+    menu_map = {menu["id"]: menu for menu in menus}
+    arr = []
+    for menu in menus:
+
+        # 有父级
+        if mid := menu.get(parent_flag):
+            # 有 子项的情况
+            if result := menu_map[mid].get(children_key):
+                result.append(menu)
+            else:
+                # 无子项的情况
+                menu_map[mid][children_key] = [menu]
+        else:
+            arr.append(menu)
+    return arr
+
+
+def menu_table():
+    """生成菜单表数据"""
+    from models import MenuModel
+    MenuModel.bulk_create([
+        MenuModel(name="系统管理",
+                  meta={"icon": "Grid"},
+                  path="/system",
+                  type=0),
+        MenuModel(name="系统设置",
+                  meta={"icon": "Setting"},
+                  path="/setting",
+                  type=0),
+        MenuModel(name="菜单管理",
+                  meta={"icon": "Menu"},
+                  path="/system/menu",
+                  type=1,
+                  component="/system/menu",
+                  pid=1,
+                  api="/menu",
+                  method="{'GET}",
+                  regx="^/menu$"
+                  )
+    ])

backend/dbhelper/role.py

@@ -0,0 +1,15 @@
+from tortoise import connections
+
+
+async def get_role_menus(rid: int):
+    """
+    根据角色id 获取菜单
+    """
+    db = connections.get("default")
+    return await db.execute_query_dict(
+        """
+        select m.id, m.name, m.meta, m.path, m.type, m.component, m.pid, m.identifier, m.api_regx,m.api, m.method, m.sort 
+        FROM sys_menu as m, sys_role_menu WHERE m.id = sys_role_menu.mid
+        AND sys_role_menu.rid = (%s) AND m.`status` = 1 ORDER BY m.sort""",
+        [rid],
+    )

backend/dbhelper/user.py

@@ -0,0 +1,72 @@
+from tortoise.transactions import atomic
+
+from core.enums import Status
+from models import RoleModel, UserModel, UserRoleModel
+from schemas.user import UserRole
+
+
+async def get_user(kwargs):
+    """
+    根据条件查询到第一条符合结果的数据
+    Args:
+        kwargs:
+
+    Returns:
+
+    """
+    return await UserModel.filter(**kwargs).first()
+
+
+async def get_user_info(pk: int):
+    """
+    根据id查用户角色列表,当前激活角色
+    """
+    user = await UserModel.get(pk=pk).values(
+        "id", "username", "nickname", "identity", "created", "modified"
+    )
+    role = (
+        await UserRoleModel.filter(uid=pk, status__not_in=[9, 5])
+        .all()
+        .values("rid", "status")
+    )
+    active_rid = role[0].get("rid")
+    rids = []
+    for obj in role:
+        if obj.get("status") == Status.SELECTED:
+            active_rid = obj.get("rid")
+        rids.append(obj.get("rid"))
+    return {**user, "active_rid": active_rid, "rids": rids}
+
+
+async def get_users(skip: int, limit: int, kwargs: dict):
+    """
+    分页获取用户并且支持字段模糊查询
+    Args:
+        skip: 偏移量
+        limit: 数量
+        kwargs: 查询字典
+
+    Returns:
+
+    """
+    kwargs = {f"{k}__contains": v for k, v in kwargs.items()}
+    result = (
+        UserModel.filter(status__not_in=[9, 5], **kwargs).all().order_by("-created")
+    )
+    return await result.offset(skip).limit(limit), await result.count()
+
+
+@atomic()
+async def insert_user(user, roles):
+    for index, rid in enumerate(roles):
+        # 1. 查角色表是否有该角色
+        await RoleModel.get(pk=rid)
+        # 创建用户
+        obj = await UserModel.create(**user.dict())
+
+        user_role = UserRole(rid=rid, uid=obj.id)
+        if index == 0:
+            user_role.status = Status.SELECTED
+        # 第一个角色默认, 添加到关系表
+        await UserRoleModel.create(**user_role.dict())
+    return user

backend/main.py

@@ -0,0 +1,41 @@
+from fastapi import FastAPI
+from fastapi.openapi.docs import get_swagger_ui_html
+
+from controller import register_routers
+from core.events import close_orm, init_orm
+from core.log import logger_db_client
+from core.utils import menu_table
+
+app = FastAPI(
+    on_startup=[init_orm, menu_table],
+    on_shutdown=[close_orm],
+    docs_url=None,
+    redoc_url=None,
+)
+
+register_routers(app)
+
+
+@app.get("/docs", include_in_schema=False)
+async def custom_swagger_ui_html():
+    return get_swagger_ui_html(
+        openapi_url=app.openapi_url,
+        title=app.title + " - Swagger UI",
+        oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
+        swagger_js_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/4.10.3/swagger-ui-bundle.js",
+        swagger_css_url="https://cdn.bootcdn.net/ajax/libs/swagger-ui/4.10.3/swagger-ui.css",
+    )
+
+
+for i in app.routes:
+    logger_db_client.debug(i.__dict__)
+    logger_db_client.info(f"{i.path}, {i.methods}, {i.path_regex}")
+    """
+     'path_regex': re.compile('^/role/(?P<
+pk>[^/]+)/menu$'), 'path_format': '/role/{pk}/menu', 
+    """
+
+
+if __name__ == '__main__':
+    import uvicorn
+    uvicorn.run("main:app", reload=True)

backend/models/__init__.py

@@ -0,0 +1,4 @@
+from models.menu import MenuModel
+from models.relation import RoleMenuModel, UserRoleModel
+from models.role import RoleModel
+from models.user import UserModel

backend/models/menu.py

@@ -0,0 +1,25 @@
+from core.enums import MenuType
+from core.table import Table, fields
+
+
+class MenuModel(Table):
+    """
+    菜单表
+    """
+
+    name = fields.CharField(max_length=20, description="名称", null=True)
+    meta = fields.JSONField(description="元数据信息", null=True)
+    path = fields.CharField(max_length=128, description="菜单url", null=True)
+    type = fields.IntEnumField(MenuType, description="菜单类型")
+    component = fields.CharField(max_length=128, description="组件地址", null=True)
+    pid = fields.IntField(description="父id", null=True)
+    identifier = fields.CharField(max_length=30, description="权限标识 user:add", null=True)
+    api = fields.CharField(max_length=128, description="接口地址", null=True)
+    method = fields.CharField(max_length=10, description="接口请求方式", null=True)
+    regx = fields.CharField(max_length=50, description="接口地址正则表达式", null=True)
+
+    class Meta:
+        table = "sys_menu"
+        table_description = "菜单表"
+        # 非唯一的索引
+        indexes = ("type", "name")

backend/models/relation.py

@@ -0,0 +1,25 @@
+from core.table import Table, fields
+
+
+class RoleRelationMixin:
+    rid = fields.IntField(description="角色id")
+
+
+class UserRoleModel(Table, RoleRelationMixin):
+    """用户角色关系表"""
+
+    uid = fields.IntField(description="用户id")
+
+    class Meta:
+        table = "sys_user_role"
+        indexes = ("uid", "rid")
+
+
+class RoleMenuModel(Table, RoleRelationMixin):
+    """角色菜单(权限)关系表"""
+
+    mid = fields.IntField(description="菜单ID")
+
+    class Meta:
+        table = "sys_role_menu"
+        indexes = ("mid", "rid")

backend/models/role.py

@@ -0,0 +1,14 @@
+from core.table import Table, fields
+
+
+class RoleModel(Table):
+    """
+    角色表
+    """
+
+    name = fields.CharField(max_length=20, description="角色名称")
+    remark = fields.CharField(max_length=200, description="角色描述")
+
+    class Meta:
+        table = "sys_role"
+        table_description = "角色表"

backend/models/user.py

@@ -0,0 +1,19 @@
+from core.enums import UserType
+from core.table import Table, fields
+
+
+class UserModel(Table):
+    """
+    用户模型类 > user table
+    """
+
+    username = fields.CharField(max_length=16, description="账号", unique=True)
+    nickname = fields.CharField(max_length=20, description="姓名", null=True)
+    password = fields.CharField(max_length=128, description="密码")
+
+    class Meta:
+        table = "sys_user"
+        table_description = "用户表"
+        # 索引
+        unique_together = ("username",)
+

backend/requirements.txt

@@ -0,0 +1,23 @@
+aiosqlite==0.17.0
+anyio==3.6.1
+bcrypt==4.0.0
+click==8.1.3
+colorama==0.4.5
+ecdsa==0.18.0
+fastapi==0.82.0
+h11==0.13.0
+idna==3.3
+iso8601==1.0.2
+passlib==1.7.4
+pyasn1==0.4.8
+pydantic==1.10.2
+pypika-tortoise==0.1.6
+python-jose==3.3.0
+pytz==2022.2.1
+rsa==4.9
+six==1.16.0
+sniffio==1.3.0
+starlette==0.19.1
+tortoise-orm==0.19.2
+typing-extensions==4.3.0
+uvicorn==0.18.3

backend/schemas/common.py

@@ -0,0 +1,36 @@
+"""公共模型"""
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, Field
+from pydantic.generics import GenericModel
+
+T = TypeVar("T")
+
+
+class LoginForm(BaseModel):
+    """用户登录参数"""
+
+    username: str = Field(..., description="账号", max_length=12, min_length=3)
+    password: str = Field(..., description="密码", min_length=6, max_length=16)
+
+
+class LoginResult(BaseModel):
+    """登录响应模型"""
+
+    id: int = Field(..., description="用户ID")
+    access_token: str = Field(..., description="token 串")
+    token_type: str = Field("Bearer", description="token 类型")
+
+
+class QueryData(BaseModel):
+    """分页查询基础数据"""
+
+    offset: int = 1
+    size: int = 10
+
+
+class ListAll(GenericModel, Generic[T]):
+    """查列表时的模型"""
+
+    total: int = Field(..., description="总数")
+    items: T = Field(..., description="数据列表")

backend/schemas/menu.py

@@ -0,0 +1,6 @@
+from tortoise.contrib.pydantic import pydantic_model_creator
+
+from models import MenuModel
+
+MenuRead = pydantic_model_creator(MenuModel, name="MenuOut")
+MenuIn = pydantic_model_creator(MenuModel, name="MenuIn", exclude_readonly=True)

backend/schemas/role.py

@@ -0,0 +1,16 @@
+
+from pydantic import Field
+from tortoise.contrib.pydantic import pydantic_model_creator
+
+from models import RoleModel
+
+RoleRed = pydantic_model_creator(RoleModel, name="RoleOut")
+RoleIn = pydantic_model_creator(RoleModel, name="RoleIn", exclude_readonly=True)
+
+
+class RoleAdd(RoleIn):
+    menus: list[int] = Field(..., description="菜单列表")
+
+
+class RoleInfo(RoleRed):
+    pass

backend/schemas/user.py

@@ -0,0 +1,29 @@
+from typing import List, Optional
+
+from pydantic import Field
+from tortoise.contrib.pydantic import pydantic_model_creator
+
+from models import UserModel, UserRoleModel
+from schemas.common import QueryData
+
+UserRead = pydantic_model_creator(UserModel, name="UserOut", exclude=("password",))
+UserIn = pydantic_model_creator(UserModel, name="UserIn", exclude_readonly=True)
+
+UserRole = pydantic_model_creator(UserRoleModel, name="UserRole", exclude_readonly=True)
+
+
+class UserInfo(UserRead):
+    active_rid: int = Field(..., description="用户当前激活角色")
+    rids: List[int] = Field(..., description="用户拥有角色")
+
+
+class UserAdd(UserIn):
+    rids: List[int] = Field(..., description="用户角色列表")
+
+
+class UserQuery(QueryData):
+    username: Optional[str] = Field("", description="用户名")
+    nickname: Optional[str] = Field("", description="姓名")
+
+
+UserList = List[UserRead]