Flask SQLAlchemy không đóng các kết nối cơ sở dữ liệu MySQL

Cuối cùng tôi đã tìm thấy giải pháp cho sự cố trên.

Tôi đã sử dụng mô hình khai báo được xác định trong tại đây thay vì làm theo tài liệu bắt đầu nhanh cho Flask-SQLAlchemy được cung cấp tại đây .

Các tệp đã thay đổi như sau:

db.py :

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine(DB_URI, convert_unicode=True)
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    import user
    Base.metadata.create_all(bind=engine)

main.py :

from db import init_db, db_session

init_db()

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

user.py :

from sqlalchemy import Column, Integer, String
from data_models.db import Base
class User(Base):
    id = db.Column(Integer, primary_key=True)
    username = db.Column(String(80), unique=True, nullable=False)
    email = db.Column(String(120), unique=True, nullable=False)

Để truy vấn các bản ghi, chúng ta có thể sử dụng User.query.filter_by().all() hoặc db_engine.execute() .

Để ghi dữ liệu mới vào cơ sở dữ liệu, chúng ta có thể sử dụng như sau:

new_user_entry = User(username = "abc", email = "[email protected]")
db_session.add(new_user_entry)
db_session.commit()