flask+python快速搭建
app.py
"""APP 入口模块"""
from traceback import format_excfrom api_limiter import limiter
from flask import Flask, jsonify
import loggingfrom controller import api_sql_blueapp = Flask(__name__)
limiter.init_app(app)
app.register_blueprint(api_sql_blue)@app.errorhandler(Exception)
def handle_exception(e):"""处理所有异常"""logging.error(format_exc())# 返回 JSON 错误响应return jsonify(error=str(e)), 500if __name__ == '__main__':app.run(debug=True,threaded=True)controller.py
from flask import Blueprint, make_responsefrom response_dic import ResDic
from api_limiter import limiter
from flask import jsonify, request
from functools import wraps
from jsonschema import validate, ValidationErrorfrom sql_service import sql_import_deal, sql_query_dealapi_sql_blue = Blueprint('sql', __name__)def validate_json_and_schema(schema):"""验证请求数据是否是JSON格式，以及是否符合schema"""def decorator(f):@wraps(f)def wrapper(*args, **kwargs):if request.is_json:data = request.get_json()try:validate(instance=data, schema=schema)except ValidationError as e_validation:return jsonify(ResDic.from_params(code="400", msg=str(e_validation))), 200return f(*args, **kwargs)else:return jsonify(ResDic.from_params(code="400", msg="请求数据不是JSON格式")), 200return wrapperreturn decoratordef header_required(func):@wraps(func)def decorated_function(*args, **kwargs):authorization_header = request.headers.get('Authorization')if not authorization_header:return jsonify(ResDic.from_params(code="401", msg="请求参数异常")), 200return func(*args, **kwargs)return decorated_function@api_sql_blue.route('/import', methods=['POST'])
@validate_json_and_schema({"type": "object","properties": {"sql_txt": {"type": "string"}},"required": ["sql_txt"],
})
@limiter.limit("5/second", override_defaults=False) # 一秒5次
@header_required
def sql_import():req_json_data = request.get_json()req = req_json_data["sql_txt"]result = sql_import_deal(req)response_data = jsonify(ResDic.from_params(code="0", data=result))response = make_response(response_data)return response@api_sql_blue.route('/query', methods=['POST'])
@limiter.limit("5/second", override_defaults=False) # 一秒5次
@header_required
@validate_json_and_schema({"type": "object","properties": {"question": {"type": "string"}},"required": ["question"],
})
def sql_query():req_json_data = request.get_json()req = req_json_data["question"]result = sql_query_deal(req)response_data = jsonify(ResDic.from_params(code="0", data=result))response = make_response(response_data)return response
api_limiter.py
from flask_limiter import Limiter
from flask_limiter.util import get_remote_addresslimiter = Limiter(key_func=get_remote_address,default_limits=["50000 per day", "3600 per hour"]
)
response_dic.py
class ResDic:def __init__(self, code, data, msg):self.code = codeself.data = dataself.msg = msg@classmethoddef from_params(cls, code="200", data="", msg=""):return {"code": code, "data": data,"msg": msg}sql_service.py
def sql_import_deal(req:str):return "import"def sql_query_deal(req:str):return "query"
config_utils.py
from dotenv import load_dotenv
import osload_dotenv('.env') # 先加载默认的 .env 文件APP_ENV = os.environ.get('APP_ENV', 'dev').lower()
load_dotenv(f'.env_{APP_ENV}', override=True) # 再加载环境相关的，def get_config(key: str) -> str:return os.environ.get(key).env
OPENAI_API_KEY=""
requirements.txt
Flask==2.3.2
Flask-Limiter==3.3.1
gevent==22.10.2
gunicorn==20.1.0
jsonschema==4.18.0
jsonschema-specifications==2023.6.1
numpy==1.23.5
numexpr==2.8.4
openai==0.27.4
openapi-schema-pydantic==1.2.4
py-healthcheck==1.10.1
pydantic==1.10.7
pydub==0.25.1
faiss-cpu==1.7.4
typing-inspect==0.8.0
typing_extensions==4.5.0
tiktoken==0.4.0
beautifulsoup4==4.12.2
gradio==3.36.1
pytest==7.4.0
python-dotenv==1.0.0
pymysql==1.1.0
urllib3==1.25.11