wxy2077 / fastapi-mysql-generator

FastAPI + MySQL Web项目生成器 ，个人认为较为合理的项目组织结构；基于apscheduler的定时任务。

Hi there 👋

I am a full-stack developer, primarily using Golang, Vue, and Python.

Projects

• IM go backend: Golang websocket service
• IM Vue3+VantUI: Vue3 VantUI IM UI
• Python Project Template: Python3 FastAPI template
• Django Blog: Python3 Django blog
• JavaScript Code Decrypt: JavaScript Code Decrypt

GitHub Stats

fastapi-mysql-generator's Issues

下载代码 运行起来之后 有2个错误

1、mysql 端口设置 并没有调用
2、./common/sys_casbin.py 26 行 提示参数数量不对 删除 True 就可以了

  File "./common/sys_casbin.py", line 26, in get_casbin
    e = casbin.Enforcer(settings.CASBIN_MODEL_PATH, adapter, True)
TypeError: __init__() takes from 1 to 3 positional arguments but 4 were given

3、除了 admin/auth/login/access-token /admin/auth/user/info 其他接口 管理员访问都提示权限不足

我现在都走不下去了

博客看不了了

博客园在审核，能不能搬运到github或者其他的博客上呢？
感谢！

项目中以接口的形式添加任务没有问题，但如果单独写一个文件测试添加任务，任务添加成功了，可是没有执行是什么原因

项目中以接口的形式添加任务没有问题，但如果单独写一个文件测试添加任务，任务添加成功了，可是没有执行是什么原因

[BUG]examples/demo_scheduler/main.py

https://github.com/CoderCharm/fastapi-mysql-generator/blob/master/examples/demo_scheduler/main.py#L110
应该去掉参数next_run_time=datetime.now()，因为如果加上就会立即运行，有悖于固定时间运行。
是应该加到async def add_cron_job函数里面？

[Discuss]Some objects cannot be intelligently prompted and auto-completed

可以强制加上type hint，但是对于ScheduleCli类本身的方法比如init_xxx就没办法提示了，现在勉强能用。

# 创建schedule对象 但是这种方式使用方法时没有提示
schedule: AsyncIOScheduler = ScheduleCli()

我这里测试接口 报错参数缺少

想问下作者是改了什么?

你好作者 @CoderCharm 我将 mongodb 以及 ws 注册到了 fastapi 组件，我可以 PR 到此吗？

就像这样

def register_mysql(app: FastAPI):
    # 添加数据库连接和关闭事件
    @app.on_event("startup")
    async def connect_to_mysql():
        await database.connect()
        logger.debug("MYSQL 数据库初始化成功 ... DONE")
        await database.disconnect()

def register_mongodb(app: FastAPI):
    # 添加数据库连接和关闭事件
    @app.on_event("startup")
    async def connect_to_mongo():
        try:
            db.client = AsyncIOMotorClient(settings.MONGODB_URL,
                                           maxPoolSize=settings.MAX_CONNECTIONS_COUNT,
                                           minPoolSize=settings.MIN_CONNECTIONS_COUNT)
            logger.debug("MONGODB 数据库初始化成功 ... DONE")
        except:
            logger.error("MONGODB 数据库初始化失败 ... DONE")

    @app.on_event("shutdown")
    async def close_mongo_connection():
        db.client.close()
        logger.debug("MONGODB 数据库连接关闭 ... DONE")


def register_ws(app: FastAPI):
    @app.websocket("/ws/{taskId}")
    async def websocket_endpoint(websocket: WebSocket, taskId):
        await notifier.connect(websocket)
        try:
            if taskId:
                recent_msgs = query_task_log(taskId)  # 根据 taskId 查询历史日志并传输到页面-代码
                print(recent_msgs)
                for msg in recent_msgs:
                    await websocket.send_json(msg)
            while True:
                data = await websocket.receive_text()
                # await websocket.send_json(data)
                await notifier.push(data)
        except WebSocketDisconnect:
            notifier.remove(websocket)

    @app.on_event("startup")
    async def connect_to_ws():
        await notifier.generator.asend(None)
        logger.debug("WEBSOCKET 连接初始化成功 ... DONE")

看到作者基于 fastapi 官方的开发模板改良，心中甚是激动啊

redis初始化发现了一点点小问题

作者您好，我在阅读您的框架代码，发现了这个小问题，在这里提醒一下~

[Problem] variable basedir in logger.py

# basedir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
basedir = os.path.dirname(os.path.dirname((os.path.dirname(os.path.abspath(__file__)))))

这块basedir因为是按照相对位置硬编码的，如果相对位置发生变化，from logger import logger生成日志的位置就会有问题，请问作者这块有啥好的解决方案吗？

response_model vs. resp_200

response_model是fastapi自带的https://fastapi.tiangolo.com/tutorial/response-model/
resp_200是作者写的

这两者应该怎么能兼容一下呢？作者是抛弃了response_model吗，就是有的时候需要人工convert了，比如sqlalchemy返回的对象不能直接用于resp_200中的data键

没有init.db文件

今天刚看没有init_db文件

怎么区分生产和测试，本地环境

作者的名字打错了...

不是 tianolo, 而是 tiangolo

怎样给他套一个后台界面？

请问怎样给他套一个后台界面？尽量低的学习成本

common.sys_schedule.py应该用单例模式实现，现在使用的时候会在使用的时候被覆盖掉

class ScheduleCli:

_instance = None
def __new__(cls, *args, **kw):
    if cls._instance is None:
        cls._instance = object.__new__(cls, *args, **kw)
    return cls._instance

如果不加单例，调用的时候时候会重新import，会覆盖fastapi启动的时候实例化的对象

单线程，DB连接被死循环占用，导致其它API 无法调用 sql

抱歉了 @CoderCharm ;这个问题可能这个项目关系不大，只是我实在找不到解决方法；我描述下我的问题；
首先基于这个demo 创建了两个api接口；
1、robot_status API 创建一个 BackgroundTasks 后台任务，用于收集第三方平台数据 ，插入进数据库

后台任务执行方法

2、get_robot_profit API 是获取数据库最新的 data

在不启动后台任务时，查询 数据、 插入数据 都能正常执行；

当我启动后台任务时，导致 get_robot_profit 这个API一直loading 获取不到数据;
后台中任务中数据插入是正常的；

请问这个问题要如何处理呢？？

[Suggestion]Share my config.py

个人觉得使用继承比较直观一些，作者可以不采纳，我只是分享一下，感谢作者的项目，很受益。

import logging
import os
import sys
from typing import Union, Optional

from loguru import logger
from pathlib import Path
from pydantic import AnyHttpUrl, BaseSettings as BSettings, IPvAnyAddress


class BaseSettings(BSettings):
    # project
    DEBUG: bool = True
    TITLE: str = "xx"
    DESCRIPTION: str = "xx"
    DOCS_URL: str = "/api/docs"
    OPENAPI_URL: str = "/api/openapi.json"
    REDOC_URL: Optional[str] = "/api/redoc"
    PROJECT_DIR: str = os.path.dirname(os.path.abspath(__file__))

    # token
    JWT_ALGORITHM: str = "HS256"
    SECRET_KEY: str = 'xxx'
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1 * 24 * 60

    # log
    LOG_PATH: str = 'logs'
    INFO_LOG_PATH: str = 'logs/info.log'
    ERROR_LOG_PATH: str = 'logs/error.log'

    # postgresql
    POSTGRESQL_USERNAME: str = 'postgres'
    POSTGRESQL_PASSWORD: str = 'xx'
    POSTGRESQL_HOST: Union[AnyHttpUrl, IPvAnyAddress] = "127.0.0.1"
    POSTGRESQL_PORT: int = 5432
    POSTGRESQL_DB: str = 'xx'
    SQLALCHEMY_DATABASE_URL = f"postgresql://{POSTGRESQL_USERNAME}:{POSTGRESQL_PASSWORD}@{POSTGRESQL_HOST}:{POSTGRESQL_PORT}/{POSTGRESQL_DB}"

    # redis
    REDIS_PASSWORD: str = "xx"
    REDIS_HOST: Union[str, AnyHttpUrl, IPvAnyAddress] = "127.0.0.1"
    REDIS_PORT: int = 6379
    REDIS_DB: int = 0
    REDIS_URL: str = f"redis://:{REDIS_PASSWORD}@{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}?encoding=utf-8"

    # geo
    GEO_MMDB_PATH = 'data/GeoLite2-City.mmdb'
    GEO_IPDATA_PATH = 'data/ipdata.ipdb'

    class Config:
        env_prefix = 'projectName_'  # defaults to no prefix, i.e. ""


    @classmethod
    def configure_logging(cls):
        class InterceptHandler(logging.Handler):
            def emit(self, record):
                logger_opt = logger.opt(depth=6, exception=record.exc_info)
                logger_opt.log(record.levelname, record.getMessage())

        log_path = Path(cls.LOG_PATH)
        info_log_path = Path(cls.INFO_LOG_PATH)
        error_log_path = Path(cls.ERROR_LOG_PATH)
        if not log_path.exists():
            log_path.mkdir()
        if not info_log_path.exists():
            info_log_path.touch()
        if not error_log_path.exists():
            error_log_path.touch()

        # logging.basicConfig(handlers=[InterceptHandler(level='INFO')], level='INFO')  # 将所有的logging都用loguru的logger再发送一遍消息。
        logger.configure(handlers=[{"sink": sys.stdout, "level": 'DEBUG'}])
        logger.add(info_log_path, rotation="500 MB", encoding='utf-8', colorize=False, level='INFO')
        logger.add(error_log_path, rotation="500 MB", encoding='utf-8', colorize=False, level='ERROR')

    def mkdir_path(self):
        runtime_clash_config_path = Path(self.RUNTIME_CLASH_CONFIG__PATH)
        if not runtime_clash_config_path.exists():
            runtime_clash_config_path.mkdir(parents=True)


class ProSettings(BaseSettings):
    # project
    DEBUG: bool = False
    # DOCS_URL: Optional[str] = None  # 生产环境关闭
    # OPENAPI_URL: Optional[str] = None
    # REDOC_URL: Optional[str] = None

    # sk
    SECRET_KEY: str = os.getenv('PROXYPOOL_SECRET_KEY', 'xxx')


class TestSettings(BaseSettings):
    pass


class DevSettings(BaseSettings):
    pass


env = os.getenv('PROXYPOOL_ENV', 'dev')
if 'pro' == env:
    settings = ProSettings()
elif 'test' == env:
    settings = TestSettings()
else:
    settings = DevSettings()

请问下作者关于字段验证的问题

比如像这种字段验证定义，我想返回的验证信息为中文，比如 用户名不能为空，或者用户名格式不正确。而不是图中的英文展示。当然
这种我是可以理解的。第一张图的那种情况怎么解决呢？

全局捕获异常失败

项目环境:centos7
python版本:3.9.9
问题描述:
生成项目之后我在api/v1/items.py写了一个简单的测试接口

@router.get("/demo", name="测试demo")
def items_test() -> Any:
    raise TypeError("demo")

请求之后并没有被server中注册的@app.exception_handler(Exception)捕获到

能给讲一下登录接口吗？

请问怎么创建用户，什么样的密码加密方法，怎么把登录接口密码验证通过了

sys_scheduler.py 中的async接口性能问题？

你好，看到该文件下的接口如果用async定义，函数内未用到await，会降低框架性能的

init_casbin()出错

init_db时报错，

python3 init_db.py
----------开发环境启动------------
角色-超级管理员-创建成功
角色-普通用户-创建成功
用户-王小右-创建成功
用户-测试用户-创建成功
Traceback (most recent call last):
File "init_db.py", line 93, in
init_casbin()
File "init_db.py", line 78, in init_casbin
e = get_casbin()
File "/home/quanpower/fastapi_app/common/sys_casbin.py", line 27, in get_casbin
e = casbin.Enforcer(settings.CASBIN_MODEL_PATH, adapter, True)
TypeError: init() takes from 1 to 3 positional arguments but 4 were given

[Discuss]Suggests upgrading some depdencies to avoid error when using MySQL 8 with utf8mb4 charset.

Got "ERROR 1366 (HY000): Incorrect string value..." error when running app using MySQL 8.0.23 with utf8mb4 charset.
Error would be gone after upgrading below dependencies.

PyMySQL==1.0.2
SQLAlchemy==1.4.6

[Problem]crud_base.py

我看这里面只有update这个方法是staticmethod，其他都是对象方法，请问作者这块有啥特殊用意吗

请问作者接项目做嘛？前后端分离，负责后端

日志优化求助

帮忙看下怎么配置https://juejin.cn/post/6844904187046658062，我试了一下有报错