flask+python快速搭建

app.py

"""APP 入口模块"""
from traceback import format_exc

from api_limiter import limiter
from flask import Flask, jsonify
import logging

from controller import api_sql_blue

app = 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)), 500

if __name__ == '__main__':
    app.run(debug=True,threaded=True)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

controller.py

from flask import Blueprint, make_response

from response_dic import ResDic
from api_limiter import limiter
from flask import jsonify, request
from functools import wraps
from jsonschema import validate, ValidationError

from sql_service import sql_import_deal, sql_query_deal

api_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))), 200
                return f(*args, **kwargs)
            else:
                return jsonify(ResDic.from_params(code="400", msg="请求数据不是JSON格式")), 200

        return wrapper

    return decorator


def 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="请求参数异常")), 200
        return 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
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

api_limiter.py

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    key_func=get_remote_address,
    default_limits=["50000 per day", "3600 per hour"]
)
1
2
3
4
5
6
7

response_dic.py

class ResDic:
    def __init__(self, code, data, msg):
        self.code = code
        self.data = data
        self.msg = msg

    @classmethod
    def from_params(cls, code="200", data="", msg=""):
        return {"code": code, "data": data,"msg": msg}

1
2
3
4
5
6
7
8
9
10

sql_service.py

def sql_import_deal(req:str):

    return "import"


def sql_query_deal(req:str):
    return "query"
1
2
3
4
5
6
7
8
9

config_utils.py

from dotenv import load_dotenv
import os

load_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)

1
2
3
4
5
6
7
8
9
10
11
12

.env

OPENAI_API_KEY=""
1

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24