A set of back-end API development templates based on FastAPI that integrates basic functions such as SqlAlchemy ORM, JWT authentication, environment configuration, logging, etc.
from typing import Literal


class Config:
    """Configuration"""

    def __getitem__(self, key):
        return self.__getattribute__(key)

    HOST: str = "0.0.0.0"
    PORT: int = 8090
    LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
    UVICORN_LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "WARNING"
    DATABASE_URL: str = "sqlite:///./test.sqlite3.db"
    JWT_SECRET_KEY: str = "secret:Miraixy-TEMPLATE"
    JWT_REFRESH_SECRET_KEY: str = "refresh:Miraixy-TEMPLATE"
    SUPER_ACCESS_KEY: str = "Miraixy-TEMPLATE"
    RELOAD: bool = False
    DEBUG: bool = True
    ACCESS_TOKEN_EXPIRE_DAYS: int = 7


class DevConfig(Config):
    """Development environment configuration"""

    LOG_LEVEL = "DEBUG"
    UVICORN_LOG_LEVEL = "DEBUG"
    DATABASE_URL = "sqlite:///./test.sqlite3.db"
    RELOAD = True
    DEBUG = True


class ProdConfig(Config):
    """Production environment configuration"""

    LOG_LEVEL = "WARNING"
    UVICORN_LOG_LEVEL = "WARNING"
    DEBUG = False

configs/config.py
Miraixy/mx-template-py-fastapi-bed
import uvicorn
from fastapi import Depends, FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security import OAuth2PasswordRequestForm

from src.conf import APP_ENV, config
from src.log import get_logging_config, logger
from src.models import database_init

# Custom routers import
from src.routers.user import login
from src.routers.user import router as user_router

# $import_routers$ 路由导入锚 *请不要修改此行* (Anchor of the router import line *Do not modify this line*)
from src.schemas.message import UserToken
from src.schemas.user import UserLogin

database_init()

app = FastAPI(
    title="FastAPI Quickstart",
    description="FastAPI Quickstart",
    version="0.0.1",
    docs_url="/",
)

""" 跨域中间件配置 """
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


""" TODO 挂载路由表 """
app.include_router(user_router, prefix="/user", tags=["User"])

# $include_routers$ 路由导入锚 *请不要修改此行* (Anchor of the router mount line *Do not modify this line*)


@app.get("/ping")
async def root():
    return {"message": "Miraixy FastAPI Quickstart Running..."}


@app.post("/token", response_model=UserToken)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    return await login(
        UserLogin(username=form_data.username, password=form_data.password),
    )


def start():
    logger.info(f"Current environment: {APP_ENV}")
    uvicorn.run(
        "src.app:app",
        host=config.HOST,
        port=config.PORT,
        log_config=get_logging_config(),
        reload=config.RELOAD,
        log_level=config.UVICORN_LOG_LEVEL.lower(),
    )

src/app.py
Miraixy/mx-template-py-fastapi-bed
import sys
from enum import Enum
from typing import Dict, Type

from configs.config import Config, DevConfig, ProdConfig


class Env(Enum):
    Dev = "dev"
    Prod = "prod"
    Default = "dev"  # noqa: PIE796


mapping: Dict[str, Type[Config]] = {"dev": DevConfig, "prod": ProdConfig}
# 获取运行参数中的环境参数 env=dev
APP_ENV: Env = Env.Default

# Parse the command-line arguments
for arg in sys.argv[1:]:
    if arg.startswith("env="):
        env = arg.split("=")[-1]
        APP_ENV = Env(env)

# Use the updated APP_ENV value to retrieve the config
config: Config = mapping[APP_ENV.value]()
src/conf.py
Miraixy/mx-template-py-fastapi-bed
import logging
import sys
from typing import TYPE_CHECKING, Any, Callable, TextIO, Union

from loguru import logger as logger

from src.conf import config

if TYPE_CHECKING:
    from logging import Handler

    from loguru import Message as LoguruMessage
    from loguru import Record as LoguruRecord
    from loguru import Writable


DEFAULT_FORMAT = (
    "<g>{time:MM-DD HH:mm:ss}</g> "
    "[<lvl>{level}</lvl>] "
    "<c><u>{name}</u></c> | "
    # "<c>{function}:{line}</c>| "
    "{message}"
)


def abstract_filter(error_no: int) -> bool:
    config_level = config.LOG_LEVEL
    config_no = (
        logger.level(config_level).no if isinstance(config_level, str) else config_level
    )
    return error_no >= config_no


def default_filter(record: "LoguruRecord"):
    return abstract_filter(record["level"].no)


class LoguruHandler(logging.Handler):
    """logging 与 loguru 之间的桥梁，将 logging 的日志转发到 loguru。"""

    def emit(self, record: logging.LogRecord):
        level_no = record.levelno
        try:
            level = logger.level(record.levelname).name
        except ValueError:
            level = level_no

        if not abstract_filter(level_no):
            return

        frame, depth = sys._getframe(6), 6  # noqa: SLF001
        while frame and frame.f_code.co_filename == logging.__file__:
            frame = frame.f_back
            depth += 1

        logger.opt(depth=depth, exception=record.exc_info).log(
            level,
            record.getMessage(),
        )


def add_handler(
    handler: Union[TextIO, "Writable", Callable[["LoguruMessage"], Any], "Handler"],
    **kwargs,
) -> int:
    return logger.add(
        handler,
        level=0,
        diagnose=False,
        filter=default_filter,
        format=DEFAULT_FORMAT,
        **kwargs,
    )


def get_logging_config():
    uvicorn_lvl = config.UVICORN_LOG_LEVEL
    return {
        "version": 1,
        "disable_existing_loggers": False,
        "handlers": {
            "default": {
                "class": "src.log.LoguruHandler",
            },
        },
        "loggers": {
            "uvicorn.error": {
                "handlers": ["default"],
                "level": uvicorn_lvl,
            },
            "uvicorn.access": {
                "handlers": ["default"],
                "level": uvicorn_lvl,
            },
        },
    }


def configure_logger():
    logger.remove()
    add_handler(sys.stdout)

src/log.py
Miraixy/mx-template-py-fastapi-bed
from src.log import logger

# TODO: Import all models that need to be automatically created
from src.models.user import DBUser  # noqa: F401

# $table_create$ 自动创建表追加锚 *请不要修改此行* (Anchor of the table creation line *Do not modify this line*)
from src.utils.db import Base, engine


def database_init():
    Base.metadata.create_all(engine, checkfirst=True)
    # 创建对象的基类:
    logger.success("Database initialized.")

src/models/__init__.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime

from sqlalchemy import Column, DateTime, Integer, String

from src.schemas.template import QueryCondition
from src.utils.db import Base, db


# 定义 _TableName_ 模型
class DB_TableName_(Base):
    __tablename__ = "_table_name_"

    id = Column(Integer, primary_key=True, autoincrement=True)
    # 在此添加表结构信息:
    name = Column(String)
    # ...

    last_update_time = Column(DateTime, default=datetime.now)
    created_time = Column(DateTime, default=datetime.now)

    @classmethod
    def add(cls, data: "DB_TableName_"):
        """新增 _TableName_ 资源"""

        data.last_update_time = datetime.now()
        data.created_time = datetime.now()
        db.add(data)
        db.commit()

    @classmethod
    def get_by_id(cls, _id: int):
        """根据 id 查询 _TableName_ 资源"""

        return db.query(cls).filter(cls.id == _id).first()

    @classmethod
    def query(cls, condition: QueryCondition):
        """根据条件查询 Order 资源"""

        page = condition.page if condition.page else 1
        page_size = condition.page_size if condition.page_size else 10
        order_field_name = condition.order_by.field_name
        order_desc = condition.order_by.desc
        keyword = condition.keyword

        query = db.query(cls)

        # for _filter in condition.filters:
        #     field_name = _filter.field_name
        #     value = _filter.value

        #     # TODO 待实现: 检查参数类型，根据不同类型添加不同筛选条件

        if keyword:
            query = db.query(cls).filter(cls.name.like(f"%{keyword}%"))

        if order_field_name:
            query = query.order_by(
                getattr(cls, order_field_name).asc()
                if not order_desc
                else getattr(cls, order_field_name).desc(),
            )

        total = query.count()

        if page and page_size:
            query = query.offset((page - 1) * page_size)
        query = query.limit(page_size)

        return query, total

    @classmethod
    def update(cls, data: "DB_TableName_", **kwargs):
        """更新 _TableName_ 资源"""

        if "id" in kwargs:
            del kwargs["id"]
        if "created_time" in kwargs:
            del kwargs["created_time"]
        if "last_update_time" in kwargs:
            del kwargs["last_update_time"]
        data.last_update_time = datetime.now()
        db.query(cls).filter(cls.id == data.id).update(dict(**kwargs))
        db.commit()

    @classmethod
    def delete(cls, data: "DB_TableName_"):
        """删除 _TableName_ 资源"""

        db.query(cls).filter(cls.id == data.id).delete()
        db.commit()

src/models/template.py
Miraixy/mx-template-py-fastapi-bed
from sqlalchemy import Column, DateTime, Integer, String

from src.utils.db import Base, db


# 定义User模型
class DBUser(Base):
    # 表的名字:
    __tablename__ = "user"

    # 表的结构:
    id = Column(Integer(), primary_key=True, autoincrement=True)  # noqa: A003
    username = Column(String(32))
    password = Column(String(32))
    perm_level = Column(Integer())
    login_time = Column(DateTime)

    @classmethod
    def add(cls, data: "DBUser"):
        db.add(data)
        db.commit()

    @classmethod
    def get_by_username(cls, username: str):
        return db.query(cls).filter(cls.username == username).first()

    @classmethod
    def update(cls, data: "DBUser", **kwargs):
        db.query(cls).filter(cls.username == data.username).update(dict(**kwargs))
        db.commit()

    @classmethod
    def delete(cls, data: "DBUser"):
        db.query(cls).filter(cls.username == data.username).delete()
        db.commit()

src/models/user.py
Miraixy/mx-template-py-fastapi-bed
from fastapi import APIRouter, Depends, HTTPException, status

from src.conf import config  # noqa: F401
from src.log import logger
from src.models.template import DB_TableName_
from src.models.user import DBUser
from src.schemas.message import Ret
from src.schemas.perm import Role
from src.schemas.template import (
    _TableName_Create,
    _TableName_Query,
    _TableName_Update,
)
from src.utils.deps import get_current_active_user

ROUTER_TAG = "_Table_name_"

router = APIRouter()


@router.post("/create", tags=[ROUTER_TAG], summary="创建")
async def create(data: _TableName_Create):
    """创建 _TableName_ 资源"""
    try:
        item = DB_TableName_(**data.model_dump())
        DB_TableName_.add(item)
        return Ret.success(
            "Create success",
            data={
                "id": item.id,
                "name": item.name,
                "last_update_time": item.last_update_time,
                "created_time": item.created_time,
            },
        )
    except:
        logger.error(f"Create {data} resource failed")
        return Ret.fail("Create failed")


@router.get("/detail", tags=[ROUTER_TAG], summary="查询详情")
async def get(_id):
    """根据 id 查询 _TableName_ 资源"""

    item = DB_TableName_.get_by_id(_id)
    if not item:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    return Ret.success(
        "query success",
        data={
            "id": item.id,
            "name": item.name,
            "last_update_time": item.last_update_time,
            "created_time": item.created_time,
        },
    )


@router.post("/list", tags=[ROUTER_TAG], summary="检索分页")
async def query(data: _TableName_Query):
    """根据条件检索 _TableName_ 资源"""

    try:
        # TODO DB_TableName_.query 方法默认提供了分页、排序、关键字过滤，如果需要其他条件需自行实现
        try:
            items, total = DB_TableName_.query(data.condition)
        except Exception as e:
            logger.error(f"Query {data} resource failed: {e}")
            return Ret.fail("Query failed, please check your parameter and try again")

        return Ret.success(
            "query success",
            data={
                "list":[{
                    "id": item.id,
                    "name": item.name,
                    "created_time": item.created_time,
                    "last_update_time": item.last_update_time,
                } for item in items],
                "total": total,
            },
        )
    except Exception as e:
        logger.error(f"Query {data} resource failed: {e}")
        return Ret.fail("Query failed")


@router.put("/update", tags=[ROUTER_TAG], summary="更新数据")
async def update(data: _TableName_Update):
    """根据 id 更新 Example 资源"""
    item = DB_TableName_.get_by_id(data.id)
    if not item:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    try:
        update_data = data.model_dump()
        # ... deal with update

        DB_TableName_.update(item, **update_data)
        return Ret.success("Update success", data={"id": item.id, "name": item.name})
    except Exception as e:
        logger.error(f"Update {data} resource failed: {e}")
        return Ret.fail("Update failed")


@router.delete("/delete", tags=[ROUTER_TAG], summary="删除数据")
async def delete(_id: int, current_user: DBUser = Depends(get_current_active_user)):
    """根据 id 删除 _TableName_ 资源"""
    if current_user.perm_level < Role.Admin: # type: ignore
        return Ret.fail("Permission denied")
    try:
        item = DB_TableName_.get_by_id(_id)
        DB_TableName_.delete(item)
        return Ret.success("Delete success")
    except Exception as e:
        logger.error(f"Delete resource(id: {_id}) failed: {e}")
        return Ret.fail("Delete failed")


logger.success(f"Router {ROUTER_TAG} initialized")

src/routers/template.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime

from fastapi import APIRouter, Depends, HTTPException, status

from src.conf import config
from src.log import logger
from src.models.user import DBUser
from src.schemas.message import Ret, UserToken
from src.schemas.user import UserCreate, UserLogin, UserUpdate
from src.utils.auth import (
    create_access_token,
    create_refresh_token,
    get_hashed_password,
    get_perm_role,
    verify_password,
)
from src.utils.deps import get_current_active_user

ROUTER_TAG = "User"

router = APIRouter()


@router.post("/register", tags=[ROUTER_TAG], summary="用户注册")
async def register(data: UserCreate):
    if data.access_key != config.SUPER_ACCESS_KEY:
        return Ret.fail("Permission denied")
    logger.info(f"User: {data} is registering...")
    if DBUser.get_by_username(data.username):
        return Ret.fail("Username already exists")
    try:
        DBUser.add(
            DBUser(
                username=data.username,
                password=get_hashed_password(data.password),
                perm_level=data.perm_level,
                login_time=None,
            ),
        )
        return Ret.success("Register success")
    except:
        return Ret.fail("Register failed")


@router.post("/login", tags=[ROUTER_TAG], summary="用户登录")
async def login(data: UserLogin):
    user = DBUser.get_by_username(data.username)
    logger.info(f"User: {user.username if user else 'Unknown'} is logging in...")
    if user and verify_password(data.password, user.password):  # type: ignore
        logger.info(f"User {user.username} logging in successfully...")
        DBUser.update(user, login_time=datetime.now())
        return UserToken(
            access_token=create_access_token(user.username),
            refresh_token=create_refresh_token(user.username),
            token_type="bearer",
        )
    logger.info(f"User {data.username} logging in failed...")
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Incorrect username or password",
        headers={"WWW-Authenticate": "Bearer"},
    )


@router.post("/edit", tags=[ROUTER_TAG], summary="用户更新")
async def edit(data: UserUpdate):
    if data.access_key != config.SUPER_ACCESS_KEY:
        return Ret.fail("Permission denied")
    if user := DBUser.get_by_username(data.username):
        model_dict = data.model_dump()
        del model_dict["access_key"]
        del model_dict["username"]
        DBUser.update(user, **model_dict)
        return Ret.success("Update success")
    return Ret.fail("User not found")


@router.get("/me", tags=[ROUTER_TAG], summary="用户个人信息")
async def info(current_user: DBUser = Depends(get_current_active_user)):
    logger.info(f"User: {current_user.username} querying personal information...")
    return Ret.success(
        "query success",
        data={
            "username": current_user.username,
            "userId": current_user.id,
            "perm_level": current_user.perm_level,
            "perm_role": get_perm_role(current_user.perm_level),  # type: ignore
        },
    )

logger.success(f"Successfully initialized {ROUTER_TAG} router")

src/routers/user.py
Miraixy/mx-template-py-fastapi-bed
from typing import Any

from pydantic import BaseModel


class Ret(BaseModel):
    code: int
    msg: str
    data: Any

    def __init__(self, code: int, msg: str, data: Any):
        super().__init__(code=code, msg=msg, data=data)

    @classmethod
    def success(cls, msg: str, data: Any = None):
        return cls(code=200, msg=msg, data=data)

    @classmethod
    def fail(cls, msg: str, data: Any = None):
        return cls(code=400, msg=msg, data=data)

    @classmethod
    def error(cls, msg: str, data: Any = None):
        return cls(code=500, msg=msg, data=data)


class UserToken(BaseModel):
    code: int
    access_token: str
    refresh_token: str
    token_type: str

    def __init__(self, access_token: str, refresh_token: str, token_type: str):
        super().__init__(
            code=200,
            access_token=access_token,
            refresh_token=refresh_token,
            token_type=token_type,
        )


class LoginRet(Ret):
    data: UserToken

src/schemas/message.py
Miraixy/mx-template-py-fastapi-bed
from enum import IntEnum


class Role(IntEnum):
    User = 0
    Admin = 10
    Super = 20

src/schemas/perm.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime
from typing import List

from pydantic import BaseModel


class _TableName_(BaseModel):
    id: int
    last_update_time: datetime
    created_time: datetime

class OrderOption(BaseModel):
    field_name: str
    desc: bool

class FilterOption(BaseModel):
    field_name: str
    value: str

class QueryCondition(BaseModel):
    page: int
    page_size: int
    order_by: OrderOption
    keyword: str
    filters: List[FilterOption]

class _TableName_Create(BaseModel):
    name: str

class _TableName_Update(BaseModel):
    id: int
    name: str

class _TableName_Query(BaseModel):
    condition: QueryCondition

src/schemas/template.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime

from pydantic import BaseModel


class User(BaseModel):
    id: int #  # noqa: A003
    username: str
    password: str
    perm_level: int
    login_time: datetime

    class Config:
        from_attributes = True


class UserCreate(BaseModel):
    username: str
    password: str
    perm_level: int
    access_key: str


class UserUpdate(BaseModel):
    access_key: str
    username: str
    perm_level: int


class UserLogin(BaseModel):
    username: str
    password: str


class TokenData(BaseModel):
    username: str

src/schemas/user.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime, timedelta
from typing import Any, Optional, Union

from jose import jwt
from passlib.context import CryptContext

from src.conf import config
from src.schemas.perm import Role

ACCESS_TOKEN_EXPIRE_MINUTES = 24 * 60  # 1 day
REFRESH_TOKEN_EXPIRE_MINUTES = 60 * 24 * config.ACCESS_TOKEN_EXPIRE_DAYS
ALGORITHM = "HS256"
JWT_SECRET_KEY = config.JWT_SECRET_KEY
JWT_REFRESH_SECRET_KEY = config.JWT_REFRESH_SECRET_KEY

password_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def get_hashed_password(password: str) -> str:
    """获取加密后的密码"""

    return password_context.hash(password)


def verify_password(password: str, hashed_pass: str) -> bool:
    """验证密码"""

    return password_context.verify(password, hashed_pass)


def create_access_token(
    subject: Union[str, Any],
    expires_delta: Optional[int] = None,
) -> str:
    """创建token"""

    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta  # type: ignore
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)  # type: ignore

    to_encode = {"exp": expires_delta, "sub": str(subject)}
    return jwt.encode(to_encode, JWT_SECRET_KEY, ALGORITHM)


def create_refresh_token(
    subject: Union[str, Any],
    expires_delta: Optional[int] = None,
) -> str:
    """创建refresh token"""

    if expires_delta is not None:
        expires_delta = datetime.utcnow() + expires_delta  # type: ignore
    else:
        expires_delta = datetime.utcnow() + timedelta(minutes=REFRESH_TOKEN_EXPIRE_MINUTES)  # type: ignore

    to_encode = {"exp": expires_delta, "sub": str(subject)}
    return jwt.encode(to_encode, JWT_REFRESH_SECRET_KEY, ALGORITHM)


def get_perm_role(perm_level: int) -> str:
    """获取权限角色"""

    if perm_level < Role.Admin:
        return "User"
    if perm_level < Role.Super:
        return "Admin"
    return "Super"

src/utils/auth.py
Miraixy/mx-template-py-fastapi-bed
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session, sessionmaker

from src.conf import config
from src.log import logger

# Create Database Connection:
Base = declarative_base()

# Init Database engine:
engine = create_engine(config.DATABASE_URL)
Base.metadata.create_all(engine)

connection = engine.connect()
logger.info(f"Connected to database {config.DATABASE_URL}")

# Create Database Session:
try:
    db: Session = sessionmaker(bind=engine)()
except:
    logger.exception("Failed to create DBSession")
    raise

src/utils/db.py
Miraixy/mx-template-py-fastapi-bed
from datetime import datetime, timedelta

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

from src.conf import config
from src.models.user import DBUser
from src.schemas.user import TokenData

reuseable_oauth = OAuth2PasswordBearer(tokenUrl="/login", scheme_name="JWT")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
ALGORITHM = "HS256"


async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, config.JWT_SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception  # noqa: B904
    user = DBUser.get_by_username(token_data.username)
    if user is None:
        raise credentials_exception
    return user


async def get_current_active_user(
    current_user: DBUser = Depends(get_current_user),
) -> DBUser:
    expiration_time = current_user.login_time + timedelta(
        minutes=config.ACCESS_TOKEN_EXPIRE_DAYS * 24 * 60,
    )

    if expiration_time < datetime.now():  # type: ignore
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Inactive user",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return current_user

src/utils/deps.py
Miraixy/mx-template-py-fastapi-bed
import sys
from pathlib import Path
from time import sleep

TARGET_NAME: str = ""
AUTO_ADDITION: bool = False

for env in sys.argv[:]:
    if env.startswith("name="):
        env = env.split("=")[-1]
        TARGET_NAME = env
        break
else:
    raise Exception("缺少必须的 `name` 参数")

if "-a" in sys.argv:
    AUTO_ADDITION = True

OUTPUT_CLASS_NAME = "".join([i.capitalize() for i in TARGET_NAME.split("_")])
OUTPUT_LOWER_NAME = TARGET_NAME.lower().replace(" ", "_").replace("-", "_")
OUTPUT_CAPITAL_NAME = (
    TARGET_NAME[0].upper()
    + TARGET_NAME.lower().replace(" ", "_").replace("-", "_")[1:]
)

MODEL_TEMPLATE_PATH: str = "./src/models/template.py"
SCHEMA_TEMPLATE_PATH: str = "./src/schemas/template.py"
ROUTER_TEMPLATE_PATH: str = "./src/routers/template.py"

MODEL_OUTPUT_PATH: str = f"./src/models/{OUTPUT_LOWER_NAME}.py"
SCHEMA_OUTPUT_PATH: str = f"./src/schemas/{OUTPUT_LOWER_NAME}.py"
ROUTER_OUTPUT_PATH: str = f"./src/routers/{OUTPUT_LOWER_NAME}.py"

APP_PY_PATH: str = "./src/app.py"
MODEL_INIT_PY_PATH: str = "./src/models/__init__.py"


def content_replace(content: str) -> str:
    return (
        content.replace(
            "_TableName_",
            OUTPUT_CLASS_NAME,
        )
        .replace(
            "_Table_name_",
            OUTPUT_CAPITAL_NAME,
        )
        .replace(
            "_table_name_",
            OUTPUT_LOWER_NAME,
        )
        .replace(
            "from src.schemas.template import",
            f"from src.schemas.{OUTPUT_LOWER_NAME} import",
        )
        .replace(
            "from src.models.template import",
            f"from src.models.{OUTPUT_LOWER_NAME} import",
        )
    )


def append_line_before_anchor_text(
    original_text, anchor_text: str, new_line: str,
) -> str:
    if new_line in original_text:
        print(f"* WARNING: `{new_line}` 已经存在，跳过该行")
        return original_text
    return original_text.replace(anchor_text, f"{new_line}\n{anchor_text}")


def main():
    model_template = Path.open(Path(MODEL_TEMPLATE_PATH), encoding="utf-8").read()
    schema_template = Path.open(Path(SCHEMA_TEMPLATE_PATH), encoding="utf-8").read()
    router_template = Path.open(Path(ROUTER_TEMPLATE_PATH), encoding="utf-8").read()

    for output_path in [MODEL_OUTPUT_PATH, SCHEMA_OUTPUT_PATH, ROUTER_OUTPUT_PATH]:
        if Path.exists(Path(output_path)):
            print(f"* WARNING: 文件 {output_path} 已经存在，正在退出...")
            return

    with Path.open(Path(MODEL_OUTPUT_PATH), "w", encoding="utf-8") as f:
        f.write(content_replace(model_template))

    with Path.open(Path(SCHEMA_OUTPUT_PATH), "w", encoding="utf-8") as f:
        f.write(content_replace(schema_template))

    with Path.open(Path(ROUTER_OUTPUT_PATH), "w", encoding="utf-8") as f:
        f.write(content_replace(router_template))

    if AUTO_ADDITION:
        print(
            "自动追加已启用，将在 5s 后执行自动追加操作"
            "! 注意: 自动追加完成后如需撤销请手动删除追加的代码，请勿在 IED 中使用 Ctrl+Z 撤销操作，否则可能造成代码格式异常",
        )
        sleep(5)
        original_file_text = Path.open(Path(APP_PY_PATH), encoding="utf-8").read()
        app_file_text = append_line_before_anchor_text(
            original_file_text,
            "# $import_routers$",
            f"from src.routers.{OUTPUT_LOWER_NAME} import router as {OUTPUT_LOWER_NAME}_router",
        )
        app_file_text = append_line_before_anchor_text(
            app_file_text,
            "# $include_routers$",
            f'app.include_router({OUTPUT_LOWER_NAME}_router, prefix="/{OUTPUT_LOWER_NAME}", tags=["{OUTPUT_CAPITAL_NAME}"])',
        )
        with Path.open(Path(APP_PY_PATH), "w", encoding="utf-8") as f:
            f.write(app_file_text)

        original_file_text = Path.open(
            Path(MODEL_INIT_PY_PATH),
            encoding="utf-8",
        ).read()
        new_file_text = append_line_before_anchor_text(
            original_file_text,
            "# $table_create$",
            f"from src.models.{OUTPUT_LOWER_NAME} import DB{OUTPUT_CLASS_NAME}  # noqa: F401",
        )
        with Path.open(Path(MODEL_INIT_PY_PATH), "w", encoding="utf-8") as f:
            f.write(new_file_text)

        print(
            "CRUD 模板创建成功,请在 `src/models`、`src/routers` 和 `src/schemas` 目录下查看生成的文件\n",
            "* 自动追加：已在 `./src/app.py` 和 `./src/models/__init__.py` 中添加模板引入",
        )

    else:
        print(
            "CRUD 模板创建成功,请在 `src/models`、`src/routers` 和 `src/schemas` 目录下查看生成的文件\n\n"
            f"* 请在 `/src/models/__init__.py` 中使用 `from src.models.{OUTPUT_LOWER_NAME} import DB{OUTPUT_CLASS_NAME}` 导入生成的数据模型\n\n"
            "* 请在 `/src/app.py` 中使用\n"
            f"  > `from src.routers.{OUTPUT_LOWER_NAME} import router as {OUTPUT_LOWER_NAME}_router` 导入生成的路由\n"
            f'  > `app.include_router({OUTPUT_LOWER_NAME}_router, prefix="/{OUTPUT_LOWER_NAME}", tags=["{OUTPUT_CAPITAL_NAME}"])` 来注册路由\n',
        )

tools/create_crud.py
Miraixy/mx-template-py-fastapi-bed