初始化fastapi 工程模板

.gitignore

@@ -0,0 +1,22 @@
+.idea
+.vscode
+*.pyc
+__pycache__/
+conf/*
+my.setting
+server.setting
+static/img/banners/*
+static/img/modals/*
+static/img/avatars/*
+static/img/filters/*
+static/img/qrcode/*
+static/img/qrcode_zip/*
+static/img/qrcode_temp/*
+log/*.log
+log/*.lock
+log/
+--ini
+output.log
+push_to_gitee.sh
+/static/img/cash_out/
+/venv/

README.md

@@ -0,0 +1,39 @@
+# FastAPI App
+
+>  Python(3.8.6+) and pip(20.2.4+) 
+
+## 一、配置conf文件
+> (没有的话需要自己创建,放在conf文件夹下级)
+``` bash
+===================== conf-dev.ini =========================
+[common]
+static_folder=./static
+template_folder=./templates
+
+[mysql]
+USERNAME=admin
+PASSWORD=123456
+HOST=localhost
+PORT=3306
+DATABASE=test
+SQLALCHEMY_DATABASE_URI=mysql+pymysql://%(USERNAME)s:%(PASSWORD)s@%(HOST)s:%(PORT)s/%(DATABASE)s
+```
+
+## 二、安装依赖
+``` bash
+pip install -r requirements.txt
+```
+
+##三、运行项目
+```bash
+目录下运行 start.bat (windows)文件或者start.sh (mac)
+```
+
+
+##四、models文件夹下的SQLAlchemy Model代码生成
+```bash
+安装完依赖库后，即可通过命令行工具直接生成，无须手写。 例子参考如下：
+sqlacodegen.exe --tables permission_info --outfile .\Desktop\fastapi_app\models\permission_info.py mysql+pymysql://chenwj:123456@localhost/waterv3?charset=utf8
+
+```
+

api/test.py

@@ -0,0 +1,32 @@
+# Test Router 页面
+
+from fastapi import APIRouter, Query
+from typing import Optional
+from fastapi_sqlalchemy import db
+
+from models.devices_place import DevicesPlace
+from pydantic_sqlalchemy import sqlalchemy_to_pydantic
+
+
+router = APIRouter(prefix="/test")
+
+
+@router.get("/")
+def index():
+    return {"msg": "This is Index Page"}
+
+
+@router.get("/login")
+def login():
+    return {}
+
+
+# 数据库查询
+@router.get("/query", response_model=sqlalchemy_to_pydantic(DevicesPlace))
+def query(mid: str = Query(..., description='mid'), region_id: Optional[int] = Query(None, description="区域Id")):
+    if region_id is None:
+        record = db.session.query(DevicesPlace).filter(DevicesPlace.mid == mid).first()
+    else:
+        record = db.session.query(DevicesPlace).filter(DevicesPlace.mid == mid)\
+            .filter(DevicesPlace.region_id == region_id).first()
+    return record

config.py

@@ -0,0 +1,50 @@
+import os
+from configparser import ConfigParser
+from typing import Optional
+from pydantic import BaseModel
+
+
+class ReConfigParser(ConfigParser):
+    def __init__(self, defaults=None):
+        ConfigParser.__init__(self, defaults=defaults)
+
+    """复写方法实现key值区分大小写"""
+    def optionxform(self, optionstr):
+        return optionstr
+
+
+class CommonConfig(BaseModel):
+    SECRET_KEY: str = os.urandom(32)
+    PROJECT_NAME: str
+    API_V1_STR: str
+    # 允许访问的origins
+    BACKEND_CORS_ORIGINS: str
+
+
+class MySQLConfig(BaseModel):
+    USERNAME: str = None
+    PASSWORD: str = None
+    HOST: Optional[str] = "localhost"
+    PORT: Optional[int] = 3306
+    DATABASE: str = None
+    SQLALCHEMY_DATABASE_URI: str = (
+        f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DATABASE}?charset=utf8"
+    )
+
+
+def init_config():
+    """初始化配置文件"""
+    print("加载配置文件...")
+    config = ReConfigParser()
+    try:
+        if os.getenv("FAST_API_ENV") == 'prod':
+            config.read(os.path.join('.', 'conf', 'conf-prod.ini'), encoding='utf-8')
+        else:
+            config.read(os.path.join('.', 'conf', 'conf-dev.ini'), encoding='utf-8')
+
+        mysql_config = MySQLConfig(**dict(config.items('mysql')))
+        # common_config = CommonConfig(**dict(config.items('common')))
+        return mysql_config
+    except Exception as e:
+        print(e)
+        raise Exception("Config Error!")

main.py

@@ -0,0 +1,29 @@
+import os
+from fastapi import FastAPI
+from fastapi_sqlalchemy import DBSessionMiddleware
+import logging.config as logging_config
+from config import init_config
+
+
+def create_app():
+    app = FastAPI()
+
+    @app.on_event("startup")
+    def startup_event():
+        # 日志配置
+        logging_config.fileConfig('conf/log.ini')
+        # 初始化配置文件
+        mysql_config = init_config()
+        # 添加sqlalchemy数据库中间件
+        # once the middleware is applied, any route can then access the database session
+        # from the global ``db``
+        app.add_middleware(DBSessionMiddleware, db_url=mysql_config.SQLALCHEMY_DATABASE_URI)
+
+    # 在这里添加API route
+    from api import test
+    app.include_router(test.router)
+    return app
+
+
+fast_api_app = create_app()
+

models/devices_place.py

@@ -0,0 +1,22 @@
+# coding: utf-8
+from sqlalchemy import Column, DateTime, Integer, text
+from sqlalchemy.dialects.mysql import VARCHAR
+from sqlalchemy.ext.declarative import declarative_base
+
+Base = declarative_base()
+metadata = Base.metadata
+
+
+class DevicesPlace(Base):
+    __tablename__ = 'devices_place'
+    __table_args__ = {'comment': '设备：地址表'}
+
+    id = Column(Integer, primary_key=True)
+    mid = Column(VARCHAR(10), nullable=False, index=True, comment='设备管理ID')
+    customer_account = Column(VARCHAR(12), nullable=False, index=True, comment='甲方')
+    region_id = Column(Integer, nullable=False, server_default=text("'0'"), comment='区域id')
+    building_id = Column(Integer, nullable=False, server_default=text("'0'"), comment='所在楼（建筑）的id，0为待定')
+    floor = Column(Integer, nullable=False, server_default=text("'0'"), comment='楼层，0为待定，可为负数')
+    place = Column(VARCHAR(50), comment='位置（通常是房间号）')
+    ctime = Column(DateTime, nullable=False, index=True, server_default=text("CURRENT_TIMESTAMP"), comment='记录创建时间')
+    utime = Column(DateTime, nullable=False, index=True, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"), comment='记录更新时间')

requirements.txt

@@ -0,0 +1,16 @@
+aiofiles==0.6.0
+fastapi==0.63.0
+fastapi-login==1.5.3
+FastAPI-SQLAlchemy==0.2.1
+Jinja2==2.11.2
+MarkupSafe==1.1.1
+pydantic==1.8.1
+pydantic-sqlalchemy==0.0.8.post1
+PyMySQL==0.10.1
+pytest==6.2.2
+requests==2.24.0
+sqlacodegen==2.3.0
+SQLAlchemy==1.3.23
+starlette==0.13.6
+uvicorn==0.13.4
+Werkzeug==1.0.1

start.bat

@@ -0,0 +1,2 @@
+set FAST_API_ENV=dev
+uvicorn main:fast_api_app --reload

start.sh

@@ -0,0 +1,2 @@
+export FAST_API_ENV=dev
+uvicorn main:app --reload

test/test_api.py

@@ -0,0 +1,19 @@
+import unittest
+import requests
+
+
+class MyTestCase(unittest.TestCase):
+    # def test_something(self):
+    #     self.assertEqual(True, False)
+
+    def test_api_index(self):
+        res = {"msg": "This is Index Page"}
+        r = requests.get("http://localhost:8000/test/")
+        if r.status_code == 200:
+            self.assertEqual(r.json(), res)
+        else:
+            self.assertFalse("测试失败")
+
+
+if __name__ == '__main__':
+    unittest.main()