数据库集成

Flask-SQLAlchemy 是 Flask 最常用的数据库扩展，提供了 ORM 功能。

安装

pip install flask-sqlalchemy flask-migrate

基本配置

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)

# 数据库配置
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)
migrate = Migrate(app, db)

定义模型

from datetime import datetime

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # 关系
    posts = db.relationship('Post', backref='author', lazy=True)
    
    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    
    def __repr__(self):
        return f'<Post {self.title}>'

常用字段类型

类型	说明
db.Integer	整数
db.String(n)	字符串，最大长度 n
db.Text	长文本
db.DateTime	日期时间
db.Float	浮点数
db.Boolean	布尔值
db.PickleType	存储 Python 对象
db.LargeBinary	二进制数据

数据库操作

创建表

# 创建所有表
with app.app_context():
    db.create_all()

增

# 创建记录
user = User(username='john', email='[email protected]')
db.session.add(user)
db.session.commit()

# 批量添加
users = [
    User(username='alice', email='[email protected]'),
    User(username='bob', email='[email protected]')
]
db.session.add_all(users)
db.session.commit()

查

# 查询所有
users = User.query.all()

# 根据主键查询
user = User.query.get(1)

# 条件查询
user = User.query.filter_by(username='john').first()
users = User.query.filter(User.email.endswith('@example.com')).all()

# 复杂查询
from sqlalchemy import and_, or_

users = User.query.filter(
    and_(User.username.like('j%'), User.email.contains('example'))
).all()

# 排序
users = User.query.order_by(User.created_at.desc()).all()

# 分页
pagination = User.query.paginate(page=1, per_page=10, error_out=False)
users = pagination.items
total = pagination.total
pages = pagination.pages

改

user = User.query.get(1)
user.email = '[email protected]'
db.session.commit()

删

user = User.query.get(1)
db.session.delete(user)
db.session.commit()

数据库迁移

初始化迁移

flask db init

创建迁移

flask db migrate -m "Initial migration"

应用迁移

flask db upgrade

回滚

flask db downgrade

关系

一对多

class Department(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    employees = db.relationship('Employee', backref='department', lazy=True)

class Employee(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    dept_id = db.Column(db.Integer, db.ForeignKey('department.id'))

多对多

# 关联表
student_course = db.Table('student_course',
    db.Column('student_id', db.Integer, db.ForeignKey('student.id'), primary_key=True),
    db.Column('course_id', db.Integer, db.ForeignKey('course.id'), primary_key=True)
)

class Student(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    courses = db.relationship('Course', secondary=student_course, backref='students')

class Course(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

一对一

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80))
    profile = db.relationship('Profile', backref='user', uselist=False)

class Profile(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    bio = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), unique=True)

查询方法

# 限制和偏移
users = User.query.limit(10).offset(20).all()

# 计数
count = User.query.filter_by(username='john').count()

# 聚合
from sqlalchemy import func

result = db.session.query(
    func.count(User.id),
    func.max(User.created_at)
).first()

# 分组
results = db.session.query(
    User.username,
    func.count(Post.id)
).join(Post).group_by(User.id).all()

# 连接
posts = Post.query.join(User).filter(User.username == 'john').all()

应用工厂模式

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app():
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
    
    db.init_app(app)
    migrate.init_app(app, db)
    
    return app

下一步

学习用户认证和会话管理。