flask结构示例

结构

project
    main.py
    util_base/db_util
    one_moudle/views.py

util_base/db_util

import cx_Oracle


def get_db_pool():
    try:
        db_pool = cx_Oracle.SessionPool(username, password, dsn=cx_Oracle.makedsn(ip, port, sid=sid), max=max, min=min, increment=5, threaded=True)
    except Exception as e:
        raise Exception("failed create db_pool, error={0}".format(str(e)))

    return db_pool


def get_database_connection_from_pool(db_pool):
    db_conn = db_pool.acquire()
    return db_conn


def release_database_connection_into_pool(db_pool, connection):
    db_pool.release(connection)

one_moudle/views.py

from flask import Blueprint, g
from flask_jsonrpc import JSONRPC


assets = Blueprint("assets", __name__)
assets_rpc = JSONRPC(None)


@assets_rpc.method("my_func")
def my_func():
    print(123)

main.py

import logging
import os
from logging import handlers

from flask import Flask, current_app, g, request
from flask_jsonrpc import JSONRPC

from one_moudle.views import assets
from util_base.db_util import get_db_pool, get_database_connection_from_pool, release_database_connection_into_pool


def get_log_handle():
    handler = handlers.RotatingFileHandler(os.path.join("./log/", "project_name.log"), maxBytes=81920, encoding="utf-8", backupCount=9)
    logging_format = logging.Formatter("%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s")
    handler.setFormatter(logging_format)
    return handler


app = Flask(__name__)
app.logger.addHandler(get_log_handle())
app.logger.setLevel(logging.INFO)

jsonrpc = JSONRPC(app, "/api")

with app.app_context():
    setattr(current_app, "_db_pool", get_db_pool())

jsonrpc.register_blueprint(assets)


@app.before_request
def before_request():
    with app.app_context():
        db_pool = getattr(current_app, "_db_pool")
    g.db_conn = get_database_connection_from_pool(db_pool)  # 没有使用sqlalchemy扩展, 这里使用g存放数据库链接


@app.after_request
def after_request(response):
    response.handlers["Access-Control-Allow-Origin"] = "*"  # CROS js 跨域请求验证
    response.handlers["Access-Control-Allow-Headers"] = "Content-Type"
    response.handlers["Access-Control-Allow-Methods"] = "POST, OPTIONS"

    if response.status_code != 200 or b"error" in response.data:  # jsonrpc会再次封装错误, 返回值不一定是200
        app.logger.error("request: " + request.data.decode())
        app.logger.error("response: " + response.data.decode())

    return response


@app.teardown_request
def teardown_request(exception)
    if getattr(g, "db_conn", None):
        with app.app_context():
            db_pool = getattr(current_app, "_db_pool")
        release_database_connection_into_pool(db_pool=db_pool, connection=g.db_conn)


if __name__ == "__main__":
    app.run(host=0.0.0.0, port=5001, threaded=True)

猜你喜欢