flask中使用sqlite的方式

Flask官网的sqlite例子可以查看这里http://www.pythondoc.com/flask/patterns/sqlite3.html,我这里根据我项目的实际情况是这样写的

我的入口文件manager.py

from flask import Flask,current_app,g
from app.application import create_app
app = create_app()
@app.teardown_request
def teardown_request(exception):
    if hasattr(g, 'db'):
        g.db.close()
if __name__ == '__main__':
    app.run(host=app.config["HOST"], port=app.config["PORT"], debug=app.config["DEBUG"])

然后是app目录下的__init__.py文件中创建g对象的sqlite数据库连接

from flask import Flask,current_app,g
from app.db_helper import SQLite_DB
import sqlite3
def get_db():
    if 'db' not in g:
        g.db = SQLite_DB(current_app.config['DB_FILE'])
        # g.db.row_factory = sqlite3.Row
    return g.db

最后是调用方法

from app import get_db
@app.route("/view", methods=['POST','GET'])
def view():
    db = get_db()
    data = db.execute_sql("SELECT * FROM user")
    return jsonify(data)

我自己的sqlite操作类db_helper.py大概如下

import sqlite3
class SQLite_DB(object):
    def __init__(self, db_path):
        self.db_path = db_path
        self.conn = self._conn()
        self.cur = self.conn.cursor()
    def _conn(self):
        conn = False
        try:
            conn = sqlite3.connect(self.db_path)
        except Exception as e:
            print("Connect sqlite failed , {}".format(e))
        return conn
    def _close(self):
        if not self._conn:
            self._conn.close()
        else:
            print("DataBase doesn't close connectiong , please check the db status .")
    def execute_sql(self, sql):
        data = self.cur.execute(sql)
        # d = []
        # for dat in data.fetchall():
        #     d.append(dat)
        return data.fetchall()
    ...

我的目录结构如下

微信截图_20220715105231.png

整个操作sqlite原理是,在需要使用sqlite的地方通过将sqlite类实例化以后添加到全局g对象,在使用完成以后有一个teardown_request的钩子方法自动关闭连接,这个钩子的用法可以参考我之前的笔记:https://sulao.cn/post/666.html


内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://sulao.cn/post/860.html