Tue Nov 14 16:19:00 CST 2023 inscode

.inscode

-run = "pip install -r requirements.txt;python main.py"
+run = "pip install -r requirements.txt;uvicorn main:app"
+language = "python3.11"
 
 [packager]
 AUTO_PIP = true
@@ -6,7 +7,10 @@ AUTO_PIP = true
 [env]
 VIRTUAL_ENV = "/root/${PROJECT_DIR}/venv"
 PATH = "${VIRTUAL_ENV}/bin:${PATH}"
-PYTHONPATH = "$PYTHONHOME/lib/python3.10:${VIRTUAL_ENV}/lib/python3.10/site-packages"
-REPLIT_POETRY_PYPI_REPOSITORY = "http://mirrors.csdn.net.cn/repository/csdn-pypi-mirrors/simple"
+PYTHONPATH = "$PYTHONHOME/lib/python3.11:${VIRTUAL_ENV}/lib/python3.11/site-packages"
+REPLIT_POETRY_PYPI_REPOSITORY = "https://repo.msshsst.com:8081/simple/"
 MPLBACKEND = "TkAgg"
-POETRY_CACHE_DIR = "/root/${PROJECT_DIR}/.cache/pypoetry"
\ No newline at end of file
+POETRY_CACHE_DIR = "/root/${PROJECT_DIR}/.cache/pypoetry"
+
+[debugger]
+program = "main.py"

app/dependencies/__init__.py

+__author__ = "ziyan.yin"
+__describe__ = "dependencies"

app/dependencies/permission.py

+__author__ = "ziyan.yin"
+__describe__ = "authentication dependencies"
+
+from typing import Optional, Sequence
+
+from basex.core import background
+from fastapi import Depends, HTTPException
+from fastapi.params import Security
+from fastapi.security import APIKeyHeader
+from starlette.authentication import AuthCredentials
+from starlette.requests import Request
+from starlette.responses import Response
+
+from app.domains.user.entity import UserSessionSchema
+from app.domains.user.service import UserService
+from app.extensions import jose
+from app.extensions.utils import has_required_scope
+
+AUTHORIZATION: str = "Authorization"
+OAUTH_SCHEMA: APIKeyHeader = APIKeyHeader(name=AUTHORIZATION)
+
+
+async def get_authorization_header(response: Response, header: str = Depends(OAUTH_SCHEMA)) -> dict:
+    try:
+        return await background.call_async(jose.identify, header, response)
+    except ValueError:
+        raise HTTPException(
+            status_code=403,
+            detail="Invalid authorization header",
+        )
+
+
+async def set_authorization_header(user: UserSessionSchema, **kwargs) -> str:
+    user.access_token = await background.call_async(jose.sign, user.id, user.version, **kwargs)
+    return user.access_token
+
+
+async def authenticate(request: Request, user: dict = Depends(get_authorization_header)) -> AuthCredentials:
+    simple_user, scopes = await UserService().get_current_user(user["id"], user["version"])
+    request.scope["user"] = simple_user
+    request.scope["auth"] = AuthCredentials(scopes)
+    return request.scope["auth"]
+
+
+class Permission(Security):
+    def __init__(self, scopes: Optional[Sequence[str]] = None, status_code: int = 403):
+        super().__init__(scopes=scopes)
+        self.dependency = self
+        self.status_code = status_code
+
+    def __call__(self, auth: AuthCredentials = Depends(authenticate)) -> None:
+        # all scopes
+        if "all" in auth.scopes:
+            return
+        if not has_required_scope(auth, self.scopes):
+            raise HTTPException(self.status_code, "Insufficient permissions")

app/domains/__init__.py

+__author__ = "ziyan.yin"
+__describe__ = "domain space"
+
+
+__all__ = ("user", "system")  # system放在最后

app/domains/system/__init__.py

+__author__ = 'ziyan.yin'
+__describe__ = 'system'
+
+from . import model

app/domains/system/controller.py

+__author__ = "ziyan.yin"
+__describe__ = "system controller"
+
+from basex.core.entity import ResultEntity
+from fastapi import APIRouter
+
+from app.dependencies.permission import set_authorization_header
+from app.domains.system.entity import LoginEntity
+from app.domains.user.entity import UserSessionSchema
+from app.domains.user.service import UserService
+
+router = APIRouter(tags=["system"], prefix="")
+
+
+@router.post("/login", name="登录")
+async def login(entity: LoginEntity) -> ResultEntity[UserSessionSchema]:
+    """
+    登录
+    """
+    data = await UserService().authorize(entity.username, entity.password)
+    await set_authorization_header(data)
+    return ResultEntity.ok(data)

app/domains/system/data.py

+__author__ = "ziyan.yin"
+__describe__ = "system data"

app/domains/system/entity.py

+__author__ = "ziyan.yin"
+__describe__ = "system entity"
+
+from basex.core.entity import SnakeBase
+from pydantic import Field
+
+
+class LoginEntity(SnakeBase):
+    username: str = Field(title="用户名")
+    password: str = Field(title="密码")

app/domains/system/model.py

+__author__ = "ziyan.yin"
+__describe__ = "system model"
+
+from basex.db.mapper import SQLBase
+from basex.db.orm import Field
+from basex.db.sqltypes import BigIntDecorator
+from sqlalchemy import Column, SmallInteger, String, Text
+from sqlalchemy.orm import Mapped
+
+
+class SysLoggerModel(SQLBase, table=True):
+    module: Mapped[str] = Field(default="", title="日志模块", sa_column=Column(String(32), nullable=False))
+    user_id: Mapped[str] = Field(title="操作人id", sa_column=Column(BigIntDecorator(), nullable=False))
+    object_id: Mapped[str] = Field(default="", title="操作对象id", sa_column=Column(BigIntDecorator(), nullable=False))
+    log_level: Mapped[int] = Field(default=0, title="日志级别", sa_column=Column(SmallInteger(), nullable=False))
+    title: Mapped[str] = Field(default="", title="日志标题", sa_column=Column(String(16), nullable=False))
+    message: Mapped[str] = Field(default="", title="日志信息", sa_column=Column(Text(), nullable=False))
+    device: Mapped[str] = Field(default="", title="设备名称", sa_column=Column(String(16), nullable=False))
+    ip: Mapped[str] = Field(default="", title="ip地址", sa_column=Column(String(16), nullable=False))

app/domains/system/service.py

+__author__ = "ziyan.yin"
+__describe__ = "system service"
+
+from typing import Optional
+
+from basex.core.service import BaseService
+from basex.db.mapper import SQLBase
+from starlette.requests import Request
+
+from app.domains.system.model import SysLoggerModel
+from app.extensions import utils
+
+
+class LoggerService(BaseService[SysLoggerModel]):
+    DEBUG_LEVEL: int = 2
+    INFO_LEVEL: int = 1
+    ERROR_LEVEL: int = 0
+
+    async def _log(
+        self,
+        module: str,
+        object_id: str,
+        log_level: int,
+        title: str,
+        message: str,
+        request: Request,
+    ) -> None:
+        user_agent: str = request.headers.get("User-Agent", default="")
+        user_id: str = request.user.identity if "user" in request.scope else "0"
+        device = utils.get_device_from_ua(user_agent)
+        ip = utils.check_ip_from_request(request)
+
+        await self.save(
+            SysLoggerModel(
+                module=module,
+                user_id=user_id,
+                object_id=object_id,
+                log_level=log_level,
+                title=title,
+                message=message,
+                device=device,
+                ip=ip,
+            )
+        )
+
+    async def debug(
+        self,
+        request: Request,
+        obj: Optional[SQLBase] = None,
+        module: str = "",
+        title: str = "",
+        message: str = "",
+    ) -> None:
+        obj_id = obj.id if obj else "0"
+        await self._log(module, obj_id, self.DEBUG_LEVEL, title, message, request)
+
+    async def info(
+        self,
+        request: Request,
+        obj: Optional[SQLBase] = None,
+        module: str = "",
+        title: str = "",
+        message: str = "",
+    ) -> None:
+        obj_id = obj.id if obj else "0"
+        await self._log(module, obj_id, self.INFO_LEVEL, title, message, request)
+
+    async def error(
+        self,
+        request: Request,
+        obj: Optional[SQLBase] = None,
+        module: str = "",
+        title: str = "",
+        message: str = "",
+    ) -> None:
+        obj_id = obj.id if obj else "0"
+        await self._log(module, obj_id, self.ERROR_LEVEL, title, message, request)

app/domains/user/__init__.py

+__author__ = 'ziyan.yin'
+__describe__ = 'user'
+
+from . import model

app/domains/user/controller.py

+__author__ = "ziyan.yin"
+__describe__ = "user controller"
+
+from basex.core.entity import BusinessError, Page, ResultEntity
+from basex.core.enums import ResultEnum
+from fastapi import APIRouter
+
+from app.dependencies.permission import Permission
+from app.domains.user.data import UserById, UserMe
+from app.domains.user.entity import UserEntity, UserPasswordEntity
+from app.domains.user.model import SysUserModel
+from app.domains.user.service import UserService
+
+router = APIRouter(tags=["用户信息"], prefix="/users")
+
+
+@router.get("/me", name="当前用户", dependencies=[Permission()])
+async def current_user_info(user: UserMe) -> ResultEntity[UserEntity]:
+    """
+    当前用户信息
+    """
+    if data := await UserService().get(user.identity):
+        return ResultEntity.ok(
+            UserEntity(
+                id=data.id,
+                username=data.username,
+                nickname=data.nickname,
+                gender=data.gender,
+                avatar=data.avatar,
+                is_admin=data.is_admin,
+            )
+        )
+    raise BusinessError(ResultEnum.A0301)
+
+
+@router.put("/me/password", name="修改密码", dependencies=[Permission()])
+async def reset_current_user(password: UserPasswordEntity, user: UserMe) -> ResultEntity:
+    """
+    当前用户修改密码
+    """
+    await UserService().reset_user_password(
+        user.identity,
+        old_password=password.old_password,
+        password=password.new_password,
+    )
+    return ResultEntity.ok()
+
+
+@router.get("/query", name="查询用户", dependencies=[Permission()])
+async def query_user(current: int = 1, size: int = 10) -> ResultEntity[Page[SysUserModel]]:
+    """
+    查询用户列表
+    """
+    page = Page[SysUserModel](current=current, size=size)
+    page = await UserService().paginate(page)
+    return ResultEntity.ok(page)
+
+
+@router.post("", name="创建用户", dependencies=[Permission()])
+async def create_user(user: SysUserModel) -> ResultEntity[SysUserModel]:
+    """
+    创建新用户
+    """
+    return ResultEntity.ok(
+        await UserService().create_user(
+            user.username,
+            user.nickname,
+            user.password,
+            gender=user.gender,
+            email=user.email,
+            mobile=user.mobile,
+            avatar=user.avatar,
+        )
+    )
+
+
+@router.get("/{index:int}", name="用户详情", dependencies=[Permission()])
+async def get_user(user: UserById) -> ResultEntity[SysUserModel]:
+    """
+    用户信息
+    """
+    user.password = ""
+    return ResultEntity.ok(user)
+
+
+@router.put("/{index:int}", name="修改用户", dependencies=[Permission()])
+async def modify_user(origin_user: UserById, user: SysUserModel) -> ResultEntity[SysUserModel]:
+    """
+    修改用户信息
+    """
+    return ResultEntity.ok(
+        await UserService().update(
+            origin_user,
+            nickname=user.nickname,
+            email=user.email,
+            gender=user.gender,
+            mobile=user.mobile,
+            avatar=user.avatar,
+        )
+    )
+
+
+@router.delete("/{index:int}", name="删除详情", dependencies=[Permission()])
+async def delete_user(user: UserById) -> ResultEntity:
+    """
+    删除用户
+    """
+    await UserService().delete_user(user)
+    return ResultEntity.ok()

app/domains/user/data.py

+__author__ = "ziyan.yin"
+__describe__ = "user data"
+
+from typing import Annotated
+
+from basex.core.entity import BusinessError
+from basex.core.enums import ResultEnum
+from fastapi import Depends
+from starlette.requests import Request
+
+from app.domains.user.entity import CurrentUser
+from app.domains.user.model import SysUserModel
+from app.domains.user.service import UserService
+
+
+def get_current_user(request: Request) -> CurrentUser:
+    return request.user
+
+
+async def get_user_by_index(index: int) -> SysUserModel:
+    if user := await UserService().get(index):
+        return user
+    raise BusinessError(ResultEnum.A0201)
+
+
+UserMe = Annotated[CurrentUser, Depends(get_current_user)]
+UserById = Annotated[SysUserModel, Depends(get_user_by_index)]

app/domains/user/entity.py

+__author__ = "ziyan.yin"
+__describe__ = "user entity"
+
+from basex.config import settings
+from basex.core.entity import SnakeBase
+from basex.core.enums import GenderEnum
+from starlette.authentication import SimpleUser
+
+
+class CurrentUser(SimpleUser):
+    __slots__ = ("_id", "username", "nickname")
+
+    def __init__(self, username: str, identity: str, nickname: str):
+        super().__init__(username)
+        self._id = identity
+        self.nickname = nickname
+
+    @property
+    def identity(self) -> str:
+        return self._id
+
+    @property
+    def display_name(self) -> str:
+        return self.nickname
+
+
+class UserPasswordEntity(SnakeBase):
+    old_password: str
+    new_password: str
+
+
+class UserEntity(SnakeBase):
+    id: str
+    username: str = ""
+    nickname: str = ""
+    gender: GenderEnum = GenderEnum.UNKNOWN
+    avatar: str = ""
+    is_admin: bool = False
+
+
+class UserSessionSchema(UserEntity):
+    access_token: str = ""
+    token_type: str = "bearer"
+    version: int = 0
+    expires_in: int = settings.session.timeout

app/domains/user/model.py

+__author__ = "ziyan.yin"
+__describe__ = "user model"
+
+from basex.core.enums import GenderEnum, StatusEnum
+from basex.db.mapper import SQLBase
+from basex.db.orm import Field
+from basex.db.sqltypes import IntEnumDecorator
+from sqlalchemy import Boolean, Column, String, UniqueConstraint
+from sqlalchemy.orm import Mapped
+
+
+class SysUserModel(SQLBase, table=True):
+    __table_args__ = (UniqueConstraint("username", name="UK_USER"),)
+
+    username: Mapped[str] = Field(default="", title="用户名", sa_column=Column(String(64), nullable=False))
+    nickname: Mapped[str] = Field(default="", title="昵称", sa_column=Column(String(64), nullable=False))
+    password: Mapped[str] = Field(default="", title="密码", sa_column=Column(String(64), nullable=False))
+    avatar: Mapped[str] = Field(default="", title="头像", sa_column=Column(String(128), nullable=False))
+    gender: Mapped[GenderEnum] = Field(
+        default=GenderEnum.UNKNOWN,
+        sa_column=Column(IntEnumDecorator(GenderEnum), nullable=False),
+        title="性别",
+    )
+    mobile: Mapped[str] = Field(default="", title="手机号", sa_column=Column(String(16), nullable=False))
+    email: Mapped[str] = Field(default="", title="邮箱", sa_column=Column(String(32), nullable=False))
+    is_admin: Mapped[bool] = Field(default=False, title="是否管理员", sa_column=Column(Boolean, nullable=False))
+    status: Mapped[StatusEnum] = Field(
+        default=StatusEnum.ACTIVE,
+        sa_column=Column(IntEnumDecorator(StatusEnum), nullable=False),
+        title="状态",
+    )

app/domains/user/service.py

+__author__ = "ziyan.yin"
+__describe__ = "user service"
+
+import uuid
+from functools import lru_cache
+from typing import Any, Tuple
+
+import bcrypt
+from basex.core import background
+from basex.core.entity import BusinessError
+from basex.core.enums import GenderEnum, ResultEnum
+from basex.core.service import BaseService
+from basex.db.session import transactional
+from starlette.authentication import BaseUser
+
+from app.domains.user.entity import CurrentUser, UserSessionSchema
+from app.domains.user.model import SysUserModel
+
+
+@lru_cache
+def check_pwd(password: str, hashed_password: str) -> bool:
+    """
+    缓存bcrypt（bcrypt性能原因）
+    :param password:
+    :param hashed_password:
+    :return:
+    """
+    return bcrypt.checkpw(password.encode(), hashed_password.encode())
+
+
+class UserService(BaseService[SysUserModel]):
+    async def get_current_user(self, index: int, version: int) -> Tuple[BaseUser, list[str]]:
+        if res := await self.get(index):
+            if res.version != version:
+                raise BusinessError(ResultEnum.A0200)
+            scopes = ["authenticated"]
+            if res.is_admin:
+                scopes.append("all")
+            return (
+                CurrentUser(identity=res.id, username=res.username, nickname=res.nickname),
+                scopes,
+            )
+        raise BusinessError(ResultEnum.A0200)
+
+    async def authorize(self, username: str, password: str) -> UserSessionSchema:
+        if res := await self.find_one(SysUserModel.username == username):
+            if not await background.call_async(check_pwd, password, res.password):
+                raise BusinessError(ResultEnum.A0210)
+            return UserSessionSchema(
+                id=res.id,
+                username=res.username,
+                nickname=res.nickname,
+                gender=GenderEnum(res.gender),
+                avatar=res.avatar,
+                is_admin=res.is_admin,
+                version=res.version,
+            )
+        raise BusinessError(ResultEnum.A0210)
+
+    @transactional(rollback_for=Exception)
+    async def create_user(self, username: str, nickname: str, password: str, **kwargs: Any) -> SysUserModel:
+        if await self.find_one(SysUserModel.username == username):
+            raise BusinessError(ResultEnum.A0111)
+        raw_password: bytes = await background.call_async(bcrypt.hashpw, password.encode(), bcrypt.gensalt())
+        user = SysUserModel(
+            username=username,
+            nickname=nickname,
+            password=raw_password.decode(),
+            **kwargs,
+        )
+        if res := await self.save(user):
+            return res
+        raise BusinessError(ResultEnum.C0300)
+
+    async def delete_user(self, user: SysUserModel) -> int:
+        user.username += str(uuid.uuid4())
+        return await self.delete(user)
+
+    @transactional(rollback_for=Exception)
+    async def reset_user_password(self, index: str, old_password: str, password: str) -> None:
+        if res := await self.get(index):
+            if not await background.call_async(check_pwd, old_password, res.password):
+                raise BusinessError(ResultEnum.A0210)
+            res.password = (await background.call_async(bcrypt.hashpw, password.encode(), bcrypt.gensalt())).decode()
+            return
+        raise BusinessError(ResultEnum.A0201)

app/extensions/__init__.py

+__author__ = "ziyan.yin"
+__describe__ = ""

app/extensions/jose.py

+__author__ = "ziyan.yin"
+__describe__ = ""
+
+import time
+from typing import Final, Union
+
+from basex.config import settings
+from basex.core.entity import BusinessError
+from basex.core.enums import ResultEnum
+from basex.security import jwt
+from starlette.responses import Response
+
+secret_key: Final[str] = settings.session.secret_key
+time_out: Final[int] = settings.session.timeout
+
+
+def sign(index: str, version: int, **kwargs: Union[None, str, int, float, bool]) -> str:
+    """
+    jose签名
+    :param index:
+    :param version:
+    :param kwargs:
+    """
+    try:
+        exp = time.time() + time_out
+        token = jwt.encode({"id": index, "version": version, "exp": exp} | kwargs, secret_key)
+        return token
+    except jwt.JWTError:
+        return ""
+
+
+def identify(token: str, response: Response) -> dict:
+    """
+    jose认证
+    :param token:
+    :param response:
+    """
+    try:
+        schema, token = token.split(" ")
+        if schema.lower() != "bearer":
+            raise ValueError("Invalid token schema")
+        data = jwt.decode(token, secret_key, algorithm="HS256")
+        time_gap = data["exp"] - time.time()
+        if response and 0 < time_gap < time_out / 2:
+            response.headers["x-auth-token"] = sign(data["id"], data["version"])
+        return data
+    except jwt.ExpiredSignatureError:
+        raise BusinessError(ResultEnum.A0230)
+    except jwt.JWTError:
+        raise BusinessError(ResultEnum.A0200)

app/extensions/sqlmap.py

+__author__ = "ziyan.yin"
+__describe__ = "sql template"
+
+import os
+
+from loguru import logger
+from sqlalchemy import text
+from sqlalchemy.sql.elements import TextClause
+
+_template: dict[str, str] = {}
+
+
+def initial_template() -> None:
+    _template.clear()
+    root_dir = os.path.join("template", "sql")
+    logger.info("start generating sql template ")
+    for file in os.listdir(root_dir):
+        filename = file.lower()
+        if filename.endswith(".sql"):
+            logger.info("generate %s" % file)
+            with open(os.path.join(root_dir, file), "r", encoding="utf-8") as fs:
+                _template[filename.removesuffix(".sql")] = fs.read()
+    logger.info("finish generating sql template")
+
+
+def get_sql(template_name: str) -> TextClause:
+    """
+    获取sql模板
+    :param template_name: 模板名称
+    """
+    key = template_name.lower()
+    return text(_template[key] if key in _template else "")

app/extensions/utils.py

+__author__ = "ziyan.yin"
+__describe__ = ""
+
+from typing import Sequence
+
+from starlette.authentication import AuthCredentials
+from starlette.requests import Request
+
+
+def has_required_scope(auth: AuthCredentials, scopes: Sequence[str]) -> bool:
+    """
+    判断scope是否存在
+    :param auth:
+    :param scopes:
+    :return:
+    """
+    for scope in scopes:
+        if scope not in auth.scopes:
+            return False
+    return True
+
+
+def get_device_from_ua(user_agent: str) -> str:
+    """
+    通过user-agent获取设备类型
+    :param user_agent:
+    :return:
+    """
+    if not user_agent:
+        return ""
+    if "Android" in user_agent:
+        return "android"
+    if "iPhone" in user_agent:
+        return "ios"
+    if "iPad" in user_agent:
+        return "ios"
+    if "Windows Phone" in user_agent:
+        return "windows_phone"
+    if "Windows" in user_agent:
+        return "windows"
+    if "Mac" in user_agent:
+        return "mac"
+    return "unknown device"
+
+
+def inet_aton(ip: str) -> int:
+    """
+    将ip转换为整数
+    :param ip:
+    :return:
+    """
+    addr = ip.split(".")
+    return int(addr[0]) << 24 | int(addr[1]) << 16 | int(addr[2]) << 8 | int(addr[3])
+
+
+def inet_ntoa(ip: int) -> str:
+    """
+    将整数转换为ip
+    :param ip:
+    :return:
+    """
+    return "{}.{}.{}.{}".format(
+        (ip & 0xFF000000) >> 24,
+        (ip & 0x00FF0000) >> 16,
+        (ip & 0x0000FF00) >> 8,
+        ip & 0x000000FF,
+    )
+
+
+def check_ip_from_request(request: Request, suffix: str = "") -> str:
+    """
+    通过request判断ip
+    :param request:
+    :param suffix:
+    :return:
+    """
+    if not request:
+        return ""
+    ip = request.headers.get(
+        "X-Forwarded-For",
+        default=request.client.host if request.client else "ANONYMOUS",
+    )
+    ip = ip.removesuffix(suffix).split(",")[-1]
+    return ip

application.toml

+# global
+[global.project]
+title = 'Demo'
+description = 'Python Web Framework'
+version = '1.0.0'
+logo = ''
+language = 'CN'
+
+[global.session]
+auth_key = 'Authorization'
+timeout = 7200
+secret_key = 'secret_key'
+secure = false
+
+# default
+[default.server]
+domain = ''
+path = '/'
+port = 8080
+workers = 1
+debug = true
+
+[default.log]
+sink = 'log/run.log'
+level = 'INFO'
+rotation = '100MB'
+format = '{time} {name} {level} {message}'
+encoding = 'utf-8'
+
+[default.datasource.default]
+package = 'asyncmy'
+url = 'mysql+asyncmy://localhost:3306'
+echo = true
+pool_pre_ping = true
+pool_size = 20
+max_overflow = 50
+pool_recycle = 3600
+
+[default.cache]
+host = 'localhost'
+port = 6379
+password = 'secret'
+db = 0
+

generator.py

+__author__ = 'ziyan.yin'
+__describe__ = 'code generator'
+
+import os
+from typing import List
+
+total_header = ["__author__ = '{author_name}'"]
+
+init_template = total_header + [
+    "__describe__ = '{module_name}'",
+    "",
+    "from . import model",
+    ""
+]
+
+controller_template = total_header + [
+    "__describe__ = '{module_name} controller'"
+    "",
+    "from fastapi import APIRouter",
+    "",
+    "router = APIRouter(tags=['{module_name}'], prefix='/{module_name}')",
+    ""
+]
+entity_template = total_header + [
+    "__describe__ = '{module_name} entity'",
+    ""
+]
+model_template = total_header + [
+    "__describe__ = '{module_name} model'",
+    "",
+    "from basex.db.mapper import SQLBase",
+    ""
+]
+service_template = total_header + [
+    "__describe__ = '{module_name} service'",
+    "",
+    "from basex.core.service import BaseService",
+    ""
+]
+data_template = total_header + [
+    "__describe__ = '{module_name} data'",
+    "",
+    "from typing import Annotated",
+    ""
+]
+
+
+def write_py_file(root_path: str, file: str, lines: List[str]):
+    file_path = os.path.join(root_path, file)
+    with open(file_path, 'a+', encoding='utf-8') as f:
+        f.write(('\n'.join(lines)).format(author_name=author_name, module_name=module_name))
+        f.flush()
+
+
+def main():
+    init_context = []
+    controller_context = []
+    entity_context = []
+    model_context = []
+    service_context = []
+    data_context = []
+
+    root_path = os.path.join('app/domains', module_name)
+
+    if not os.path.isdir(root_path):
+        os.mkdir(root_path)
+        init_context = init_template
+        controller_context = controller_template
+        entity_context = entity_template
+        model_context = model_template
+        service_context = service_template
+        data_context = data_template
+
+    if init_context:
+        write_py_file(root_path, "__init__.py", init_context)
+    if controller_context:
+        write_py_file(root_path, "controller.py", controller_context)
+    if entity_context:
+        write_py_file(root_path, "entity.py", entity_context)
+    if model_context:
+        write_py_file(root_path, "model.py", model_context)
+    if service_context:
+        write_py_file(root_path, "service.py", service_context)
+    if data_context:
+        write_py_file(root_path, "data.py", data_context)
+
+
+if __name__ == '__main__':
+    author_name = 'ziyan.yin'
+    module_name = input("请输入模块名:")
+
+    main()

lifespan.py

+__author__ = 'ziyan.yin'
+__describe__ = ''
+
+import sys
+from contextlib import asynccontextmanager
+
+from basex.config import settings
+from basex.db import session
+from basex.ext import cache
+from fastapi import FastAPI
+from loguru import logger
+
+from app import domains
+from app.extensions import sqlmap
+
+
+@asynccontextmanager
+async def setup(app: FastAPI):
+    init_logger()
+    init_routers(app)
+
+    session.initial_engine()
+    cache.initialize()
+    sqlmap.initial_template()
+    yield
+    await session.shutdown()
+    await cache.shutdown()
+
+
+def init_logger() -> None:
+    logger.configure(
+        handlers=[
+            {'sink': sys.stdout, "level": settings.log.level},  # io
+            settings.log  # file
+        ]
+    )
+
+
+def init_routers(app: FastAPI) -> None:
+    for domain in domains.__all__:
+        __import__(f'app.domains.{domain}.controller')
+        controller = getattr(domains, domain).controller
+        app.include_router(controller.router)
+        logger.info(f'add {domain} router')

main.py

-print('欢迎来到 InsCode')
\ No newline at end of file
+__author__ = 'ziyan.yin'
+__describe__ = 'main'
+
+from basex.config import settings
+from basex.core.entity import BusinessError, ResultEntity
+from basex.core.response import EntityResponse
+from fastapi import FastAPI
+from fastapi.responses import ORJSONResponse
+from pydantic import ValidationError
+from starlette.middleware.cors import CORSMiddleware
+
+import lifespan
+
+app = FastAPI(
+    root_path=settings.server.path,
+    title=settings.project.title,
+    description=settings.project.description,
+    default_response_class=EntityResponse,
+    debug=settings.server.debug,
+    lifespan=lifespan.setup
+)
+
+app.add_middleware(CORSMiddleware, allow_methods=['*'], allow_origins=["*"])
+
+
+@app.exception_handler(BusinessError)
+async def business_exception_handler(_, exc: BusinessError):
+    status_code = 200
+    if exc.code.startswith('A020') or exc.code.startswith('A023'):
+        status_code = 401
+    return EntityResponse(
+        status_code=status_code,
+        content=ResultEntity.error(exc)
+    )
+
+
+@app.exception_handler(ValidationError)
+async def validation_exception_handler(_, exc: ValidationError):
+    return ORJSONResponse(
+        status_code=400,
+        content={
+            'detail': exc.errors()
+        }
+    )
+
+
+@app.exception_handler(Exception)
+async def common_exception_handler(_, exc: Exception):
+    return ORJSONResponse(
+        status_code=500,
+        content={
+            'detail': [
+                {
+                    "loc": (exc.__traceback__.tb_lineno if exc.__traceback__ else 0,),
+                    "msg": str(exc),
+                    "type": exc.__class__.__name__
+                }
+            ]
+        }
+    )

requirements.txt

+basex[redis,mysql]>=2.4.1,<2.5.0
+bcrypt
+fastapi
+starlette
+pydantic
+uvicorn[standard]
+gunicorn
+sqlalchemy
+alembic
+loguru
+mypy>=1.5.0,<1.6.0
\ No newline at end of file

template/sql/user.sql

+select
+        id,
+        username,
+        nickname,
+        case
+            when gender = 0 then '未知'
+            when gender = 1 then '男'
+            when gender = 2 then '女'
+            when gender = 9 then '其他'
+            else ''
+        end as gender,
+        mobile,
+        email
+from sys_user
+where deleted = 0
\ No newline at end of file