[Python] Flask MySQL 連線管理

原始碼

1. import os 
2. import mysql.connector as sql 
3.  
4. from werkzeug.exceptions import HTTPException, BadRequest 
5. from flask import Flask, g, request, json 
6.  
7.  
8. app = Flask(__name__) 
9.  
10.  
11.  
12. #[ DB 處裡 ]############################################################# 
13.  
14. # https://docsxyz.com/zh-hant/wiki/python/connector-python-connectargs 
15. db_config = { 
16.    'host'      : "localhost", 
17.    'user'      : "XXXX", 
18.    'passwd'    : "XXXX", 
19.    'db'        : 'XXXX', 
20.    'use_pure'  : True, 
21.    'autocommit': True, 
22.    'charset'   : 'utf8', 
23. } 
24.  
25.  
26. @app.before_request 
27. def before_request(): 
28.    # 在 request 前開啟 DB 連線 
29.    # g 是 Flask global 在每個 request 有獨立的 context 
30.  
31.    g.cnt = sql.connect(**db_config) 
32.    g.cursor = g.cnt.cursor(dictionary = True) 
33.  
34.  
35.  
36. @app.after_request 
37. def after_request(response): 
38.    # 在 request 後結束 DB 連線 
39.  
40.    cursor = g.get('cursor', None) 
41.    if cursor is not None: 
42.        # 當 cursor 還有 row 沒有取出，close 會發生錯誤 
43.        if cursor.with_rows : cursor.fetchall() 
44.        cursor.close() 
45.  
46.    cnt = g.get('cnt', None) 
47.    if cnt is not None: 
48.        cnt.close() 
49.  
50.    return response 

import os
import mysql.connector as sql

from werkzeug.exceptions import HTTPException, BadRequest
from flask import Flask, g, request, json


app = Flask(__name__)



#[ DB 處裡 ]#############################################################

# https://docsxyz.com/zh-hant/wiki/python/connector-python-connectargs
db_config = {
    'host'      : "localhost",
    'user'      : "XXXX",
    'passwd'    : "XXXX",
    'db'        : 'XXXX',
    'use_pure'  : True,
    'autocommit': True,
    'charset'   : 'utf8',
}


@app.before_request
def before_request():
    # 在 request 前開啟 DB 連線
    # g 是 Flask global 在每個 request 有獨立的 context

    g.cnt = sql.connect(**db_config)
    g.cursor = g.cnt.cursor(dictionary = True)



@app.after_request
def after_request(response):
    # 在 request 後結束 DB 連線

    cursor = g.get('cursor', None)
    if cursor is not None:
        # 當 cursor 還有 row 沒有取出，close 會發生錯誤
        if cursor.with_rows : cursor.fetchall()
        cursor.close()

    cnt = g.get('cnt', None)
    if cnt is not None:
        cnt.close()

    return response