【在Flask中集成SQLAlchemy】
在Flask中集成SQLAlchemy可以通过使用第三方扩展包flask-sqlalchemy来实现，以下是详细的步骤和说明：
- 首先，需要导入
SQLAlchemy类以及flask_sqlalchemy模块：
from flask_sqlalchemy import SQLAlchemy
- 实例化SQLAlchemy对象：
db = SQLAlchemy()
这个对象将用于在Flask应用程序中处理数据库操作。
- 将db对象注册到Flask应用程序中：
db.init_app(app)
- 这里的
app是Flask应用程序实例，通过init_app()方法将db对象与具体的Flask应用程序关联起来。
- 在视图函数中使用数据库会话：
# 全局的db.session是线程安全的
db.session.query(User).all()
- 在Flask中，可以通过
db.session访问数据库会话，使用SQLAlchemy的查询语法进行数据库操作。
- 在
models.py中定义数据库模型时，需要继承自db.Model：
class User(db.Model):
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
username = db.Column(db.String(80), unique=True, nullable=False)
- 在这段示例代码中，
User类继承自db.Model，可以通过db.Column定义表的列属性。
- 配置文件中添加数据库连接信息：
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root@127.0.0.1:3306/ddd?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
SQLALCHEMY_TRACK_MODIFICATIONS = False
- 在配置文件中，我们需要提供数据库连接的URI及相关的连接池配置参数。
- 这些配置项可以根据具体的数据库环境进行调整。
- 通过以上步骤，就可以在Flask应用程序中集成SQLAlchemy，并通过
db对象进行数据库操作。 - 使用SQLAlchemy与Flask结合，可以更方便地管理数据库模型、执行查询和事务操作，以及处理数据库的迁移等任务。
【项目演示】
结构树
src
__init__.py
models.py
settings.py
views.py
manage.py
models.py
# 第四步：在models中，找所有东西 都从db中拿
from src import db
class User(db.Model):
# 以__开头的是配置
__tablename__ = 'users' # 数据库表名称，如果不写，以类名作为表名
id = db.Column(db.Integer, primary_key=True) # 主键索引，聚簇索引
name = db.Column(db.String(64), index=True, nullable=False) # name字段加辅助索引
email = db.Column(db.String(32), unique=True)
__table_args__ = (
db.UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
db.Index('ix_id_name', 'name', 'email'), # 索引
)
def __str__(self):
return self.name
def __repr__(self):
return self.name
settings.py
DEBUG=True
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123@127.0.0.1:3306/db001?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
views.py
from flask import Blueprint
from .models import User
user = Blueprint('user', __name__)
from src import db
@user.route('/')
def index():
res = db.session.query(User).filter_by(name='lqz').all()
print(res)
return 'asdfasdf'
- ``
from src import app
if __name__ == '__main__':
app.run()