尧图网站建设 
1. 使用Flask-SQLAlchemySQLAlchemy是一个强大的ORM对象关系映射库通过Python对象与数据库表进行交互无需编写SQL语句。Flask-SQLAlchemy是Flask的官方扩展集成了SQLAlchemy。安装Flask-SQLAlchemypip install flask-sqlalchemy配置SQLAlchemyfrom flask import Flask from flask_sqlalchemy import SQLAlchemy app Flask(__name__) # SQLite数据库开发环境推荐 app.config[SQLALCHEMY_DATABASE_URI] sqlite:///example.db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app)2. 定义模型模型是数据库表的Python类每个模型类代表数据库中的一张表。class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) email db.Column(db.String(120), uniqueTrue, nullableFalse) age db.Column(db.Integer, nullableTrue) def __repr__(self): return fUser {self.username}常用字段类型字段类型说明Python类型db.Integer整数intdb.String(n)可变长度字符串strdb.Text长文本strdb.Boolean布尔值booldb.DateTime日期时间datetime.datetimedb.Date日期datetime.datedb.Float浮点数floatdb.Decimal高精度小数decimal.Decimaldb.JSONJSON数据dict/list常用字段参数参数说明primary_keyTrue主键uniqueTrue值唯一nullableFalse不允许为空defaultvalue默认值indexTrue创建索引autoincrementTrue自增整数主键默认3. 创建和管理数据库在定义了模型后需要在应用上下文中创建数据库和表with app.app_context(): db.create_all() # 创建所有模型对应的表删除所有表谨慎使用with app.app_context(): db.drop_all() # 删除所有表数据会丢失4. CRUD操作创建记录Createapp.route(/add_user) def add_user(): new_user User(usernamejohn_doe, emailjohnexample.com, age25) db.session.add(new_user) # 添加到会话 db.session.commit() # 提交事务保存到数据库 return User added! # 批量添加 app.route(/add_users) def add_users(): users [ User(usernamealice, emailaliceexample.com), User(usernamebob, emailbobexample.com), ] db.session.add_all(users) db.session.commit() return Users added!读取记录Read# 查询所有记录 app.route(/get_users) def get_users(): users User.query.all() return br.join([f{user.username} ({user.email}) for user in users]) # 通过主键查询单个记录 user User.query.get(1) # 查询id1的用户 # 通过条件查询精确匹配 user User.query.filter_by(usernamejohn_doe).first() # 通过条件查询多个条件 users User.query.filter_by(usernamejohn_doe, age25).all() # 获取第一条记录 user User.query.first()更新记录Updateapp.route(/update_user/int:user_id) def update_user(user_id): user User.query.get(user_id) if user: user.username new_username user.age 26 db.session.commit() return User updated! return User not found!删除记录Deleteapp.route(/delete_user/int:user_id) def delete_user(user_id): user User.query.get(user_id) if user: db.session.delete(user) db.session.commit() return User deleted! return User not found!5. 高级查询操作条件查询from sqlalchemy import or_, and_, not_ # AND条件默认 users User.query.filter(User.username john_doe, User.age 18).all() # OR条件 users User.query.filter( or_(User.username john_doe, User.email johnexample.com) ).all() # AND显式 users User.query.filter( and_(User.age 18, User.age 60) ).all() # NOT条件 users User.query.filter(not_(User.username admin)).all() # 模糊查询LIKE users User.query.filter(User.username.like(%john%)).all() # 包含查询IN users User.query.filter(User.username.in_([alice, bob])).all() # 为空判断 users User.query.filter(User.email.is_(None)).all()排序与分页# 升序 users User.query.order_by(User.username).all() # 降序 users User.query.order_by(User.username.desc()).all() # 多字段排序 users User.query.order_by(User.age.desc(), User.username).all() # 分页查询每页10条第1页 page User.query.paginate(page1, per_page10) users page.items # 当前页的数据 total page.total # 总记录数 pages page.pages # 总页数 has_prev page.has_prev # 是否有上一页 has_next page.has_next # 是否有下一页聚合查询from sqlalchemy import func # 计数 count User.query.count() # 求和 total_age db.session.query(func.sum(User.age)).scalar() # 平均值 avg_age db.session.query(func.avg(User.age)).scalar() # 最大值/最小值 max_age db.session.query(func.max(User.age)).scalar() min_age db.session.query(func.min(User.age)).scalar()6. 使用Flask-Migrate进行数据库迁移当模型发生变化时如新增字段Flask-Migrate可以生成迁移脚本自动更新表结构而不丢失数据。安装Flask-Migratepip install flask-migrate配置Flask-Migratefrom flask_migrate import Migrate migrate Migrate(app, db)迁移命令# 1. 初始化迁移环境只需执行一次 flask db init # 2. 生成迁移脚本检测模型变化 flask db migrate -m 添加age字段 # 3. 应用迁移到数据库 flask db upgrade # 4. 回滚到上一个版本 flask db downgrade命令作用flask db init初始化迁移环境生成migrations/目录flask db migrate -m 描述生成迁移脚本不修改数据库flask db upgrade应用迁移更新数据库结构flask db downgrade回滚到上一个迁移版本7. 执行原始SQL虽然SQLAlchemy提供了强大的ORM功能但在某些复杂场景下可能需要执行原始SQL语句。# 查询 app.route(/raw_sql) def raw_sql(): result db.session.execute(SELECT * FROM user) return br.join([str(row) for row in result]) # 带参数的查询防止SQL注入 app.route(/user/int:user_id) def get_user_raw(user_id): result db.session.execute( SELECT * FROM user WHERE id :id, {id: user_id} ) user result.fetchone() return fUser: {user.username} if user else Not found8. 连接MySQL数据库除了SQLiteFlask-SQLAlchemy也支持MySQL、PostgreSQL等主流数据库。安装MySQL驱动# 选择一种安装 pip install mysqlclient # 推荐性能好 # 或 pip install pymysql # 纯Python实现配置MySQL连接# URI格式: mysql://用户名:密码主机:端口/数据库名 app.config[SQLALCHEMY_DATABASE_URI] mysql://username:passwordlocalhost:3306/dbname # 如需指定字符集 # app.config[SQLALCHEMY_DATABASE_URI] mysql://username:passwordlocalhost/dbname?charsetutf8mb4完整示例from flask import Flask from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] mysql://root:123456localhost/mydb app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app) class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) email db.Column(db.String(120), uniqueTrue, nullableFalse) app.route(/) def index(): users User.query.all() return br.join([f{user.username} ({user.email}) for user in users]) if __name__ __main__: with app.app_context(): db.create_all() app.run(debugTrue)9. 常用数据库URI格式数据库URI格式SQLitesqlite:///path/to/database.dbSQLite内存sqlite:///:memory:MySQLmysql://username:passwordhost/dbnamePostgreSQLpostgresql://username:passwordhost/dbnameMariaDBmariadb://username:passwordhost/dbname10. CRUD操作速查表操作代码创建记录db.session.add(obj); db.session.commit()批量创建db.session.add_all([obj1, obj2]); db.session.commit()查询所有Model.query.all()主键查询Model.query.get(id)条件查询单条Model.query.filter_by(keyvalue).first()条件查询多条Model.query.filter_by(keyvalue).all()复杂条件Model.query.filter(or_(...)).all()排序Model.query.order_by(Model.field.desc()).all()分页Model.query.paginate(page1, per_page10)更新obj.field new_value; db.session.commit()删除db.session.delete(obj); db.session.commit()计数Model.query.count()小结本章全面讲解了Flask的数据库操作。Flask-SQLAlchemy是推荐的ORM方案——通过定义继承自db.Model的模型类来映射数据库表使用db.Column定义字段CRUD操作通过db.session.add()创建、Model.query查询、直接修改属性后commit()更新、db.session.delete()删除高级查询支持filter/filter_by条件过滤、or_/and_组合条件、order_by排序、paginate分页Flask-Migrate管理数据库迁移也支持执行原始SQL和连接MySQL等主流数据库。这些知识构成了Flask数据持久化的完整基础。 Flask入门教程（十三）：数据库操作——SQLAlchemy与CRUD完整实践
1. 使用Flask-SQLAlchemySQLAlchemy是一个强大的ORM对象关系映射库通过Python对象与数据库表进行交互无需编写SQL语句。Flask-SQLAlchemy是Flask的官方扩展集成了SQLAlchemy。安装Flask-SQLAlchemypip install flask-sqlalchemy配置SQLAlchemyfrom flask import Flask from flask_sqlalchemy import SQLAlchemy app Flask(__name__) # SQLite数据库开发环境推荐 app.config[SQLALCHEMY_DATABASE_URI] sqlite:///example.db app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app)2. 定义模型模型是数据库表的Python类每个模型类代表数据库中的一张表。class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) email db.Column(db.String(120), uniqueTrue, nullableFalse) age db.Column(db.Integer, nullableTrue) def __repr__(self): return fUser {self.username}常用字段类型字段类型说明Python类型db.Integer整数intdb.String(n)可变长度字符串strdb.Text长文本strdb.Boolean布尔值booldb.DateTime日期时间datetime.datetimedb.Date日期datetime.datedb.Float浮点数floatdb.Decimal高精度小数decimal.Decimaldb.JSONJSON数据dict/list常用字段参数参数说明primary_keyTrue主键uniqueTrue值唯一nullableFalse不允许为空defaultvalue默认值indexTrue创建索引autoincrementTrue自增整数主键默认3. 创建和管理数据库在定义了模型后需要在应用上下文中创建数据库和表with app.app_context(): db.create_all() # 创建所有模型对应的表删除所有表谨慎使用with app.app_context(): db.drop_all() # 删除所有表数据会丢失4. CRUD操作创建记录Createapp.route(/add_user) def add_user(): new_user User(usernamejohn_doe, emailjohnexample.com, age25) db.session.add(new_user) # 添加到会话 db.session.commit() # 提交事务保存到数据库 return User added! # 批量添加 app.route(/add_users) def add_users(): users [ User(usernamealice, emailaliceexample.com), User(usernamebob, emailbobexample.com), ] db.session.add_all(users) db.session.commit() return Users added!读取记录Read# 查询所有记录 app.route(/get_users) def get_users(): users User.query.all() return br.join([f{user.username} ({user.email}) for user in users]) # 通过主键查询单个记录 user User.query.get(1) # 查询id1的用户 # 通过条件查询精确匹配 user User.query.filter_by(usernamejohn_doe).first() # 通过条件查询多个条件 users User.query.filter_by(usernamejohn_doe, age25).all() # 获取第一条记录 user User.query.first()更新记录Updateapp.route(/update_user/int:user_id) def update_user(user_id): user User.query.get(user_id) if user: user.username new_username user.age 26 db.session.commit() return User updated! return User not found!删除记录Deleteapp.route(/delete_user/int:user_id) def delete_user(user_id): user User.query.get(user_id) if user: db.session.delete(user) db.session.commit() return User deleted! return User not found!5. 高级查询操作条件查询from sqlalchemy import or_, and_, not_ # AND条件默认 users User.query.filter(User.username john_doe, User.age 18).all() # OR条件 users User.query.filter( or_(User.username john_doe, User.email johnexample.com) ).all() # AND显式 users User.query.filter( and_(User.age 18, User.age 60) ).all() # NOT条件 users User.query.filter(not_(User.username admin)).all() # 模糊查询LIKE users User.query.filter(User.username.like(%john%)).all() # 包含查询IN users User.query.filter(User.username.in_([alice, bob])).all() # 为空判断 users User.query.filter(User.email.is_(None)).all()排序与分页# 升序 users User.query.order_by(User.username).all() # 降序 users User.query.order_by(User.username.desc()).all() # 多字段排序 users User.query.order_by(User.age.desc(), User.username).all() # 分页查询每页10条第1页 page User.query.paginate(page1, per_page10) users page.items # 当前页的数据 total page.total # 总记录数 pages page.pages # 总页数 has_prev page.has_prev # 是否有上一页 has_next page.has_next # 是否有下一页聚合查询from sqlalchemy import func # 计数 count User.query.count() # 求和 total_age db.session.query(func.sum(User.age)).scalar() # 平均值 avg_age db.session.query(func.avg(User.age)).scalar() # 最大值/最小值 max_age db.session.query(func.max(User.age)).scalar() min_age db.session.query(func.min(User.age)).scalar()6. 使用Flask-Migrate进行数据库迁移当模型发生变化时如新增字段Flask-Migrate可以生成迁移脚本自动更新表结构而不丢失数据。安装Flask-Migratepip install flask-migrate配置Flask-Migratefrom flask_migrate import Migrate migrate Migrate(app, db)迁移命令# 1. 初始化迁移环境只需执行一次 flask db init # 2. 生成迁移脚本检测模型变化 flask db migrate -m 添加age字段 # 3. 应用迁移到数据库 flask db upgrade # 4. 回滚到上一个版本 flask db downgrade命令作用flask db init初始化迁移环境生成migrations/目录flask db migrate -m 描述生成迁移脚本不修改数据库flask db upgrade应用迁移更新数据库结构flask db downgrade回滚到上一个迁移版本7. 执行原始SQL虽然SQLAlchemy提供了强大的ORM功能但在某些复杂场景下可能需要执行原始SQL语句。# 查询 app.route(/raw_sql) def raw_sql(): result db.session.execute(SELECT * FROM user) return br.join([str(row) for row in result]) # 带参数的查询防止SQL注入 app.route(/user/int:user_id) def get_user_raw(user_id): result db.session.execute( SELECT * FROM user WHERE id :id, {id: user_id} ) user result.fetchone() return fUser: {user.username} if user else Not found8. 连接MySQL数据库除了SQLiteFlask-SQLAlchemy也支持MySQL、PostgreSQL等主流数据库。安装MySQL驱动# 选择一种安装 pip install mysqlclient # 推荐性能好 # 或 pip install pymysql # 纯Python实现配置MySQL连接# URI格式: mysql://用户名:密码主机:端口/数据库名 app.config[SQLALCHEMY_DATABASE_URI] mysql://username:passwordlocalhost:3306/dbname # 如需指定字符集 # app.config[SQLALCHEMY_DATABASE_URI] mysql://username:passwordlocalhost/dbname?charsetutf8mb4完整示例from flask import Flask from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] mysql://root:123456localhost/mydb app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app) class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue, nullableFalse) email db.Column(db.String(120), uniqueTrue, nullableFalse) app.route(/) def index(): users User.query.all() return br.join([f{user.username} ({user.email}) for user in users]) if __name__ __main__: with app.app_context(): db.create_all() app.run(debugTrue)9. 常用数据库URI格式数据库URI格式SQLitesqlite:///path/to/database.dbSQLite内存sqlite:///:memory:MySQLmysql://username:passwordhost/dbnamePostgreSQLpostgresql://username:passwordhost/dbnameMariaDBmariadb://username:passwordhost/dbname10. CRUD操作速查表操作代码创建记录db.session.add(obj); db.session.commit()批量创建db.session.add_all([obj1, obj2]); db.session.commit()查询所有Model.query.all()主键查询Model.query.get(id)条件查询单条Model.query.filter_by(keyvalue).first()条件查询多条Model.query.filter_by(keyvalue).all()复杂条件Model.query.filter(or_(...)).all()排序Model.query.order_by(Model.field.desc()).all()分页Model.query.paginate(page1, per_page10)更新obj.field new_value; db.session.commit()删除db.session.delete(obj); db.session.commit()计数Model.query.count()小结本章全面讲解了Flask的数据库操作。Flask-SQLAlchemy是推荐的ORM方案——通过定义继承自db.Model的模型类来映射数据库表使用db.Column定义字段CRUD操作通过db.session.add()创建、Model.query查询、直接修改属性后commit()更新、db.session.delete()删除高级查询支持filter/filter_by条件过滤、or_/and_组合条件、order_by排序、paginate分页Flask-Migrate管理数据库迁移也支持执行原始SQL和连接MySQL等主流数据库。这些知识构成了Flask数据持久化的完整基础。