1. Flask数据库操作入门两种方式的选择刚开始用Flask操作数据库时很多开发者都会面临一个选择是用ORM还是直接写SQL这个问题就像学做菜时纠结用现成的调料包还是自己调配香料。我刚开始做项目时也踩过不少坑今天就把这些经验分享给大家。Flask操作数据库主要有两种方式第一种是通过Flask-SQLAlchemy这样的ORM工具第二种是直接用pymysql、psycopg2这些驱动执行原生SQL。ORM就像是个翻译官把Python对象和数据库表自动对应起来而原生SQL则是直接跟数据库对话没有中间商赚差价。我建议新手从ORM开始学起特别是当你还不熟悉SQL语法时。Flask-SQLAlchemy用起来特别顺手比如定义一个用户模型from flask_sqlalchemy import SQLAlchemy db SQLAlchemy() class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(80), uniqueTrue) email db.Column(db.String(120), uniqueTrue)这样几行代码就创建了一个用户表完全不用操心建表语句。等熟悉了ORM的工作原理后再学习原生SQL会更容易理解底层机制。2. 配置数据库连接从零开始搭建环境2.1 安装必要的依赖包在开始之前我们需要先装好必要的工具包。根据你选择的数据库类型安装命令会有些不同# 使用MySQL数据库 pip install flask-sqlalchemy pymysql # 使用PostgreSQL pip install flask-sqlalchemy psycopg2 # SQLite开发环境推荐 pip install flask-sqlalchemy我强烈建议新手先用SQLite练手它不需要额外安装数据库服务一个文件就是一个数据库特别适合本地开发测试。2.2 配置数据库连接字符串配置数据库连接是第一步也是容易出错的地方。在Flask项目中我通常会在config.py里做这些配置class Config: # SQLite配置 SQLALCHEMY_DATABASE_URI sqlite:///app.db # MySQL配置示例 # SQLALCHEMY_DATABASE_URI mysqlpymysql://用户名:密码localhost:3306/数据库名 # PostgreSQL配置示例 # SQLALCHEMY_DATABASE_URI postgresqlpsycopg2://用户名:密码localhost:5432/数据库名 SQLALCHEMY_TRACK_MODIFICATIONS False # 关闭警告信息这里有个小技巧URI中的///表示相对路径SQLite数据库会创建在项目根目录用////则是绝对路径。我曾经因为少写一个斜杠调试了半天大家一定要注意。3. 使用SQLAlchemy ORM进行CRUD操作3.1 定义数据模型定义模型是ORM的核心相当于设计数据库表结构。我习惯把模型放在单独的models.py文件中from datetime import datetime from app import db class Post(db.Model): id db.Column(db.Integer, primary_keyTrue) title db.Column(db.String(100), nullableFalse) content db.Column(db.Text, nullableFalse) created_at db.Column(db.DateTime, defaultdatetime.utcnow) user_id db.Column(db.Integer, db.ForeignKey(user.id)) # 定义关系 author db.relationship(User, backrefdb.backref(posts, lazyTrue)) def __repr__(self): return fPost {self.title}注意几个关键点每个字段都要指定类型比如String、Integer等nullableFalse表示该字段不能为空default参数可以设置默认值外键关系用ForeignKey和relationship定义3.2 基本的CRUD操作掌握了模型定义后我们来看最常用的增删改查操作# 创建记录 new_post Post(titleORM入门, content这是ORM教程..., user_id1) db.session.add(new_post) db.session.commit() # 别忘记提交 # 查询记录 # 获取所有文章 posts Post.query.all() # 获取单篇文章 post Post.query.get(1) # 通过主键查询 # 条件查询 user_posts Post.query.filter_by(user_id1).all() # 更新记录 post Post.query.get(1) post.title 修改后的标题 db.session.commit() # 删除记录 post Post.query.get(1) db.session.delete(post) db.session.commit()这里有个常见坑点很多新手会忘记db.session.commit()导致修改没有真正保存到数据库。我建议在开发时开启SQLALCHEMY_ECHOTrue配置这样能在控制台看到实际执行的SQL语句。4. 高级查询技巧与性能优化4.1 复杂查询与关联查询实际项目中我们经常需要执行复杂查询。SQLAlchemy提供了强大的查询接口# 分页查询 page request.args.get(page, 1, typeint) posts Post.query.order_by(Post.created_at.desc()).paginate(pagepage, per_page10) # 多条件查询 from sqlalchemy import or_ search request.args.get(search) posts Post.query.filter( or_( Post.title.contains(search), Post.content.contains(search) ) ).all() # 关联查询 # 获取用户及其所有文章 user User.query.options(db.joinedload(User.posts)).get(1)4.2 性能优化建议随着数据量增长查询性能会成为瓶颈。以下是几个优化建议合理使用索引class User(db.Model): # 在经常查询的字段上创建索引 email db.Column(db.String(120), indexTrue, uniqueTrue)避免N1查询问题# 不好的写法会导致N1查询 users User.query.all() for user in users: print(user.posts) # 每次循环都会查询数据库 # 好的写法使用joinedload一次性加载关联数据 users User.query.options(db.joinedload(User.posts)).all()批量操作代替循环# 批量插入 db.session.bulk_insert_mappings(Post, [ {title: 文章1, content: ..., user_id: 1}, {title: 文章2, content: ..., user_id: 1} ]) # 批量更新 Post.query.filter_by(user_id1).update({status: published})5. 原生SQL操作实战5.1 直接使用pymysql执行SQL虽然ORM很方便但有些复杂查询还是直接写SQL更高效。下面是用pymysql操作MySQL的例子import pymysql from flask import current_app def get_raw_connection(): return pymysql.connect( hostcurrent_app.config[DB_HOST], usercurrent_app.config[DB_USER], passwordcurrent_app.config[DB_PASSWORD], databasecurrent_app.config[DB_NAME], cursorclasspymysql.cursors.DictCursor # 返回字典格式结果 ) def complex_query(): conn get_raw_connection() try: with conn.cursor() as cursor: sql SELECT u.username, COUNT(p.id) as post_count FROM users u LEFT JOIN posts p ON u.id p.user_id GROUP BY u.id HAVING post_count %s ORDER BY post_count DESC cursor.execute(sql, (5,)) results cursor.fetchall() return results finally: conn.close()5.2 参数化查询与防注入直接写SQL时要特别注意SQL注入问题。一定要使用参数化查询不要拼接SQL字符串# 危险容易导致SQL注入 cursor.execute(fSELECT * FROM users WHERE username {username}) # 安全写法 cursor.execute(SELECT * FROM users WHERE username %s, (username,))6. ORM与原生SQL的混合使用6.1 在ORM中执行原生SQLSQLAlchemy也支持直接执行原生SQL同时还能利用现有的连接池from sqlalchemy import text # 使用text()函数包装SQL语句 result db.session.execute(text(SELECT * FROM users WHERE id :user_id), {user_id: 1}) # 获取结果 for row in result: print(row) # 返回的是RowProxy对象6.2 事务管理与异常处理无论是用ORM还是原生SQL事务管理都很重要try: # 开始事务 db.session.begin() # 执行多个操作 user User(usernamenewuser) db.session.add(user) db.session.flush() # 获取生成的ID post Post(title第一篇, user_iduser.id) db.session.add(post) # 提交事务 db.session.commit() except Exception as e: db.session.rollback() current_app.logger.error(f操作失败: {str(e)}) raise7. 项目实战博客系统的数据库设计7.1 数据库模型设计让我们设计一个简单的博客系统包含用户、文章、评论和标签class User(db.Model): id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(64), indexTrue, uniqueTrue) email db.Column(db.String(120), indexTrue, uniqueTrue) password_hash db.Column(db.String(128)) # 关系定义 posts db.relationship(Post, backrefauthor, lazydynamic) comments db.relationship(Comment, backrefauthor, lazydynamic) class Post(db.Model): id db.Column(db.Integer, primary_keyTrue) title db.Column(db.String(140)) body db.Column(db.Text) timestamp db.Column(db.DateTime, indexTrue, defaultdatetime.utcnow) user_id db.Column(db.Integer, db.ForeignKey(user.id)) # 多对多关系文章和标签 tags db.relationship(Tag, secondarypost_tags, backrefdb.backref(posts, lazydynamic)) class Comment(db.Model): id db.Column(db.Integer, primary_keyTrue) body db.Column(db.Text) timestamp db.Column(db.DateTime, indexTrue, defaultdatetime.utcnow) user_id db.Column(db.Integer, db.ForeignKey(user.id)) post_id db.Column(db.Integer, db.ForeignKey(post.id)) # 多对多关联表 post_tags db.Table(post_tags, db.Column(post_id, db.Integer, db.ForeignKey(post.id)), db.Column(tag_id, db.Integer, db.ForeignKey(tag.id)) ) class Tag(db.Model): id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(50), uniqueTrue)7.2 典型业务场景实现用户注册def register_user(username, email, password): if User.query.filter((User.username username) | (User.email email)).first(): raise ValueError(用户名或邮箱已存在) user User(usernameusername, emailemail) user.set_password(password) # 假设有密码哈希方法 db.session.add(user) db.session.commit() return user发布文章def create_post(title, content, author, tagsNone): post Post(titletitle, bodycontent, authorauthor) if tags: for tag_name in tags: tag Tag.query.filter_by(nametag_name).first() if not tag: tag Tag(nametag_name) db.session.add(tag) post.tags.append(tag) db.session.add(post) db.session.commit() return post文章搜索def search_posts(keyword, page1, per_page10): query Post.query.join(User).filter( (Post.title.contains(keyword)) | (Post.body.contains(keyword)) | (User.username.contains(keyword)) ).order_by(Post.timestamp.desc()) return query.paginate(pagepage, per_pageper_page)8. 性能对比与选型建议8.1 ORM vs 原生SQL性能测试我做了一个简单的性能对比测试查询1000条记录操作类型执行时间(ms)内存占用(MB)ORM查询12045原生SQL8532ORM批量6538从结果可以看出原生SQL在性能上确实有优势但ORM通过批量操作也能接近原生性能。8.2 何时选择ORM何时用原生SQL根据我的经验以下情况适合用ORM快速开发原型简单的CRUD操作需要跨数据库兼容开发团队不熟悉SQL以下情况适合用原生SQL复杂报表查询大数据量批量操作需要数据库特定功能性能关键路径实际项目中我通常采用混合模式80%的常规操作用ORM20%的性能敏感或复杂查询用原生SQL。
