Flask数据库模型定义

数据库模型是Flask应用与数据库交互的核心,通过SQLAlchemy ORM可以将Python类映射为数据库表。

1. 模型基础结构

每个模型都是一个Python类,继承自db.Model,类的属性对应数据库表的列。

1.1 基础模型定义

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class User(db.Model):
    """用户模型"""
    # 指定表名(可选,默认使用类名的小写形式)
    __tablename__ = 'users'

    # 表参数(可选)
    __table_args__ = (
        # 创建索引
        db.Index('idx_username', 'username'),
        db.Index('idx_email', 'email'),
        # 添加表注释(MySQL/PostgreSQL支持)
        {'comment': '用户信息表'},
    )

    # 主键字段
    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)

    # 文本字段
    bio = db.Column(db.Text, nullable=True)

    # 布尔字段
    is_active = db.Column(db.Boolean, default=True)
    is_admin = db.Column(db.Boolean, default=False)

    # 整数字段
    age = db.Column(db.Integer, nullable=True)
    login_count = db.Column(db.Integer, default=0)

    # 浮点数字段
    balance = db.Column(db.Float, default=0.0)

    # 日期时间字段
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    last_login = db.Column(db.DateTime, nullable=True)

    # 大数字段
    big_number = db.Column(db.BigInteger, default=0)

    # 可枚举字段(模拟枚举)
    status = db.Column(db.String(20), default='active')

    # 唯一标识字段
    uuid = db.Column(db.String(36), unique=True, nullable=True)

    def __repr__(self):
        return f'<User {self.username}>'

    def to_dict(self):
        """将模型转换为字典(用于JSON序列化)"""
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'is_active': self.is_active,
            'created_at': self.created_at.isoformat() if self.created_at else None,
        }

2. 字段类型详解

字段类型 数据库类型 Python类型 示例 描述
String VARCHAR str db.String(100) 可变长度字符串,需指定最大长度
Text TEXT str db.Text() 长文本,不限长度
Integer INTEGER int db.Integer() 整数
SmallInteger SMALLINT int db.SmallInteger() 小整数
BigInteger BIGINT int db.BigInteger() 大整数
Float FLOAT float db.Float() 浮点数
Numeric NUMERIC Decimal db.Numeric(10,2) 精确数字,适合金额
Boolean BOOLEAN bool db.Boolean() 布尔值
DateTime DATETIME datetime db.DateTime() 日期和时间
Date DATE date db.Date() 日期
Time TIME time db.Time() 时间
Enum ENUM str db.Enum('active','inactive') 枚举值
JSON JSON dict/list db.JSON() JSON数据(MySQL 5.7+/PostgreSQL)
PickleType BLOB object db.PickleType() Python对象序列化存储
LargeBinary BLOB bytes db.LargeBinary() 二进制数据

3. 字段约束和选项

class Product(db.Model):
    """产品模型 - 展示字段约束"""
    __tablename__ = 'products'

    id = db.Column(db.Integer, primary_key=True)

    # 必填约束
    name = db.Column(db.String(100), nullable=False)

    # 唯一约束
    sku = db.Column(db.String(50), unique=True, nullable=False)

    # 默认值
    price = db.Column(db.Numeric(10, 2), default=0.00)
    stock = db.Column(db.Integer, default=0)

    # 自动更新
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 索引
    category = db.Column(db.String(50), index=True)

    # 多列组合索引
    # __table_args__ = (
    #     db.Index('idx_category_status', 'category', 'status'),
    # )

    # 检查约束(SQLAlchemy需要数据库支持)
    # __table_args__ = (
    #     db.CheckConstraint('price >= 0', name='check_price_positive'),
    # )

    # 唯一约束(多列组合唯一)
    __table_args__ = (
        db.UniqueConstraint('name', 'brand_id', name='uq_product_name_brand'),
    )

    # 外键约束
    brand_id = db.Column(db.Integer, db.ForeignKey('brands.id'), nullable=False)

    # 枚举字段
    status = db.Column(db.Enum('draft', 'published', 'archived'), default='draft')

    # JSON字段
    specifications = db.Column(db.JSON, default=dict)

    # 注释
    description = db.Column(db.Text, comment='产品详细描述')

    # 服务器默认值(数据库端生成)
    version = db.Column(db.Integer, server_default='1')

    def __repr__(self):
        return f'<Product {self.name}>'

3.1 常用的字段选项

选项 示例 描述 适用场景
primary_key primary_key=True 设置为主键 唯一标识记录的字段
nullable nullable=False 是否允许NULL值 必填字段
unique unique=True 值必须唯一 用户名、邮箱、手机号等
default default=0 默认值 有常见默认值的字段
index index=True 创建单列索引 经常查询的字段
onupdate onupdate=datetime.utcnow 更新时自动设置 更新时间戳
server_default server_default='1' 数据库端默认值 数据库生成的默认值
comment comment='用户邮箱' 字段注释 文档和数据库管理

4. 模型关系

4.1 一对多关系

class Author(db.Model):
    """作者模型 - 一对多示例(一个作者有多本书)"""
    __tablename__ = 'authors'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # 定义一对多关系
    # backref: 在Book模型中创建author属性
    # lazy: 加载策略
    # cascade: 级联操作
    books = db.relationship('Book',
                          backref='author',
                          lazy='dynamic',
                          cascade='all, delete-orphan')

    def __repr__(self):
        return f'<Author {self.name}>'

class Book(db.Model):
    """书籍模型"""
    __tablename__ = 'books'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    published_date = db.Column(db.Date)

    # 外键:指向作者表
    author_id = db.Column(db.Integer, db.ForeignKey('authors.id'), nullable=False)

    # 可以通过 author 属性访问作者
    # book.author  # 获取作者对象

    def __repr__(self):
        return f'<Book {self.title}>'

# 使用示例
def one_to_many_example():
    # 创建作者和书籍
    author = Author(name='John Doe')
    book1 = Book(title='Python入门', author=author)
    book2 = Book(title='Flask进阶', author=author)

    db.session.add(author)
    db.session.add_all([book1, book2])
    db.session.commit()

    # 查询作者的所有书籍
    author_books = author.books.all()  # 返回所有书籍的列表

    # 通过书籍查询作者
    book = Book.query.first()
    book_author = book.author  # 返回作者对象

4.2 多对多关系

# 方法1:使用关联表
# 首先创建关联表
post_tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True),
    db.Column('created_at', db.DateTime, default=datetime.utcnow)
)

class Post(db.Model):
    """文章模型"""
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)

    # 多对多关系
    tags = db.relationship('Tag',
                          secondary=post_tags,
                          backref=db.backref('posts', lazy='dynamic'),
                          lazy='dynamic')

    def __repr__(self):
        return f'<Post {self.title}>'

class Tag(db.Model):
    """标签模型"""
    __tablename__ = 'tags'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False)
    slug = db.Column(db.String(50), unique=True, nullable=False)

    def __repr__(self):
        return f'<Tag {self.name}>'

# 方法2:使用关联模型(当需要存储额外信息时)
class Enrollment(db.Model):
    """选课关联模型 - 学生和课程的关联表"""
    __tablename__ = 'enrollments'

    id = db.Column(db.Integer, primary_key=True)
    student_id = db.Column(db.Integer, db.ForeignKey('students.id'), nullable=False)
    course_id = db.Column(db.Integer, db.ForeignKey('courses.id'), nullable=False)
    enrolled_at = db.Column(db.DateTime, default=datetime.utcnow)
    grade = db.Column(db.String(2), nullable=True)  # 额外字段:成绩

    # 定义关系
    student = db.relationship('Student', backref='enrollments')
    course = db.relationship('Course', backref='enrollments')

    # 复合唯一约束
    __table_args__ = (
        db.UniqueConstraint('student_id', 'course_id', name='uq_student_course'),
    )

    def __repr__(self):
        return f'<Enrollment {self.student_id}-{self.course_id}>'

class Student(db.Model):
    """学生模型"""
    __tablename__ = 'students'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)

    # 多对多关系(通过关联模型)
    courses = db.relationship('Course',
                            secondary='enrollments',
                            backref='students',
                            viewonly=True)  # 只读,通过关联模型操作

    def __repr__(self):
        return f'<Student {self.name}>'

class Course(db.Model):
    """课程模型"""
    __tablename__ = 'courses'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text)

    def __repr__(self):
        return f'<Course {self.title}>'

4.3 一对一关系

class UserProfile(db.Model):
    """用户详情模型 - 一对一关系示例"""
    __tablename__ = 'user_profiles'

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), unique=True, nullable=False)
    avatar_url = db.Column(db.String(500))
    phone = db.Column(db.String(20))
    address = db.Column(db.Text)

    # 一对一关系
    user = db.relationship('User', backref=db.backref('profile', uselist=False))

    def __repr__(self):
        return f'<UserProfile for user {self.user_id}>'

# 在User模型中添加关系
class User(db.Model):
    """用户模型(扩展)"""
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)

    # 一对一关系已在UserProfile中定义

    def __repr__(self):
        return f'<User {self.username}>'

# 使用示例
def one_to_one_example():
    # 创建用户和详情
    user = User(username='john_doe')
    profile = UserProfile(user=user, phone='1234567890', address='123 Main St')

    db.session.add_all([user, profile])
    db.session.commit()

    # 访问详情
    user_profile = user.profile  # 获取用户详情
    profile_user = profile.user  # 获取详情对应的用户

5. 模型继承

5.1 单表继承

class Content(db.Model):
    """内容基类 - 单表继承"""
    __tablename__ = 'contents'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # 鉴别器列:用于区分子类类型
    type = db.Column(db.String(50))

    __mapper_args__ = {
        'polymorphic_identity': 'content',
        'polymorphic_on': type
    }

    def __repr__(self):
        return f'<Content {self.title}>'

class Article(Content):
    """文章子类"""
    __mapper_args__ = {
        'polymorphic_identity': 'article'
    }

    # 这些字段会存储在同一个表中
    subtitle = db.Column(db.String(200))
    read_count = db.Column(db.Integer, default=0)

    def __repr__(self):
        return f'<Article {self.title}>'

class Video(Content):
    """视频子类"""
    __mapper_args__ = {
        'polymorphic_identity': 'video'
    }

    # 这些字段会存储在同一个表中
    duration = db.Column(db.Integer)  # 时长(秒)
    thumbnail_url = db.Column(db.String(500))

    def __repr__(self):
        return f'<Video {self.title}>'

# 使用示例
def single_table_inheritance():
    # 创建不同类型的內容
    article = Article(
        title='Python教程',
        content='Python是一种高级编程语言...',
        subtitle='从入门到精通',
        read_count=100
    )

    video = Video(
        title='Flask实战',
        content='Flask是一个轻量级Web框架...',
        duration=3600,
        thumbnail_url='https://example.com/thumb.jpg'
    )

    db.session.add_all([article, video])
    db.session.commit()

    # 查询所有内容
    all_contents = Content.query.all()  # 包含文章和视频

    # 只查询文章
    articles = Article.query.all()

    # 只查询视频
    videos = Video.query.all()

5.2 抽象基类

class TimestampMixin:
    """时间戳混入类(不创建表)"""
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

class SoftDeleteMixin:
    """软删除混入类"""
    is_deleted = db.Column(db.Boolean, default=False)
    deleted_at = db.Column(db.DateTime, nullable=True)

    def soft_delete(self):
        """软删除方法"""
        self.is_deleted = True
        self.deleted_at = datetime.utcnow()

class BaseModel(db.Model):
    """抽象基模型(不创建表)"""
    __abstract__ = True

    id = db.Column(db.Integer, primary_key=True)

    def save(self):
        """保存方法"""
        try:
            db.session.add(self)
            db.session.commit()
            return True
        except Exception as e:
            db.session.rollback()
            print(f"保存失败: {e}")
            return False

    def delete(self):
        """删除方法"""
        try:
            db.session.delete(self)
            db.session.commit()
            return True
        except Exception as e:
            db.session.rollback()
            print(f"删除失败: {e}")
            return False

# 使用混入类和基类
class Product(BaseModel, TimestampMixin, SoftDeleteMixin):
    """产品模型 - 继承多个混入类"""
    __tablename__ = 'products'

    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Numeric(10, 2), default=0.00)

    # 查询过滤器:排除已删除的记录
    @classmethod
    def query_active(cls):
        """查询未删除的记录"""
        return cls.query.filter_by(is_deleted=False)

    def __repr__(self):
        return f'<Product {self.name}>'

6. 高级模型特性

6.1 复合主键

from sqlalchemy import PrimaryKeyConstraint

class OrderItem(db.Model):
    """订单项模型 - 复合主键示例"""
    __tablename__ = 'order_items'

    # 复合主键
    order_id = db.Column(db.Integer, primary_key=True)
    product_id = db.Column(db.Integer, primary_key=True)

    # 或者使用 PrimaryKeyConstraint
    # __table_args__ = (
    #     PrimaryKeyConstraint('order_id', 'product_id'),
    # )

    quantity = db.Column(db.Integer, default=1, nullable=False)
    unit_price = db.Column(db.Numeric(10, 2), nullable=False)

    # 外键
    order = db.relationship('Order', backref='items')
    product = db.relationship('Product', backref='order_items')

    def __repr__(self):
        return f'<OrderItem order={self.order_id}, product={self.product_id}>'

6.2 自定义数据类型

from sqlalchemy import TypeDecorator
import json

class JSONEncodedDict(TypeDecorator):
    """JSON编码字典的自定义类型"""
    impl = db.Text  # 底层数据库类型

    def process_bind_param(self, value, dialect):
        """将Python对象转换为数据库值"""
        if value is None:
            return None
        return json.dumps(value)

    def process_result_value(self, value, dialect):
        """将数据库值转换为Python对象"""
        if value is None:
            return None
        return json.loads(value)

    def copy(self, **kw):
        """创建类型的拷贝"""
        return JSONEncodedDict(self.impl.length)

class Configuration(db.Model):
    """配置模型 - 使用自定义类型"""
    __tablename__ = 'configurations'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), unique=True, nullable=False)

    # 使用自定义类型
    settings = db.Column(JSONEncodedDict, default=dict)

    def __repr__(self):
        return f'<Configuration {self.name}>'

# 使用示例
def custom_type_example():
    config = Configuration(
        name='app_settings',
        settings={
            'theme': 'dark',
            'language': 'zh-CN',
            'notifications': True
        }
    )

    config.save()

    # 自动序列化和反序列化
    print(config.settings['theme'])  # 输出: dark

6.3 事件监听器

from sqlalchemy import event
import hashlib
import uuid

class User(db.Model):
    """用户模型 - 带有事件监听器"""
    __tablename__ = 'users'

    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), nullable=False)
    uuid = db.Column(db.String(36), unique=True)

    def set_password(self, password):
        """设置密码(生成哈希)"""
        self.password_hash = hashlib.sha256(password.encode()).hexdigest()

    def check_password(self, password):
        """验证密码"""
        return self.password_hash == hashlib.sha256(password.encode()).hexdigest()

    def __repr__(self):
        return f'<User {self.username}>'

# 事件监听器:在插入前生成UUID
@event.listens_for(User, 'before_insert')
def generate_uuid(mapper, connection, target):
    """在插入前生成UUID"""
    if not target.uuid:
        target.uuid = str(uuid.uuid4())

# 事件监听器:在更新前检查
@event.listens_for(User, 'before_update')
def update_timestamp(mapper, connection, target):
    """在更新前更新时间戳"""
    from datetime import datetime
    target.updated_at = datetime.utcnow()

# 事件监听器:在删除后记录日志
@event.listens_for(User, 'after_delete')
def log_deletion(mapper, connection, target):
    """在删除后记录日志"""
    print(f"用户 {target.username} 已被删除")

7. 完整的博客系统模型示例

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
import hashlib
import uuid

db = SQLAlchemy()

# 关联表
post_categories = db.Table('post_categories',
    db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),
    db.Column('category_id', db.Integer, db.ForeignKey('categories.id'), primary_key=True),
    db.Column('created_at', db.DateTime, default=datetime.utcnow)
)

post_tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('posts.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True),
    db.Column('created_at', db.DateTime, default=datetime.utcnow)
)

class User(db.Model):
    """用户模型"""
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(128), nullable=False)
    display_name = db.Column(db.String(100))
    avatar = db.Column(db.String(500))
    bio = db.Column(db.Text)
    website = db.Column(db.String(200))
    location = db.Column(db.String(100))

    # 状态字段
    is_active = db.Column(db.Boolean, default=True)
    is_admin = db.Column(db.Boolean, default=False)
    is_verified = db.Column(db.Boolean, default=False)

    # 时间戳
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    last_login = db.Column(db.DateTime)
    last_seen = db.Column(db.DateTime)

    # 统计字段
    post_count = db.Column(db.Integer, default=0)
    comment_count = db.Column(db.Integer, default=0)

    # 唯一标识
    uuid = db.Column(db.String(36), unique=True)

    # 关系
    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
    comments = db.relationship('Comment', backref='user', lazy='dynamic', cascade='all, delete-orphan')

    # 方法
    def set_password(self, password):
        self.password_hash = hashlib.sha256(password.encode()).hexdigest()

    def check_password(self, password):
        return self.password_hash == hashlib.sha256(password.encode()).hexdigest()

    def to_dict(self, include_sensitive=False):
        """转换为字典"""
        data = {
            'id': self.id,
            'username': self.username,
            'display_name': self.display_name,
            'avatar': self.avatar,
            'bio': self.bio,
            'website': self.website,
            'location': self.location,
            'post_count': self.post_count,
            'comment_count': self.comment_count,
            'created_at': self.created_at.isoformat() if self.created_at else None,
        }

        if include_sensitive:
            data['email'] = self.email
            data['is_admin'] = self.is_admin

        return data

    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    """文章模型"""
    __tablename__ = 'posts'

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False, index=True)
    slug = db.Column(db.String(200), unique=True, nullable=False, index=True)
    content = db.Column(db.Text, nullable=False)
    excerpt = db.Column(db.Text)  # 摘要
    featured_image = db.Column(db.String(500))

    # 状态
    status = db.Column(db.Enum('draft', 'published', 'archived'), default='draft', index=True)
    visibility = db.Column(db.Enum('public', 'private', 'password'), default='public')
    password = db.Column(db.String(100))

    # SEO
    meta_title = db.Column(db.String(200))
    meta_description = db.Column(db.Text)
    meta_keywords = db.Column(db.String(500))

    # 时间
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    published_at = db.Column(db.DateTime, index=True)

    # 统计
    view_count = db.Column(db.Integer, default=0)
    comment_count = db.Column(db.Integer, default=0)
    like_count = db.Column(db.Integer, default=0)

    # 外键
    author_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False, index=True)

    # 关系
    comments = db.relationship('Comment', backref='post', lazy='dynamic', cascade='all, delete-orphan')
    categories = db.relationship('Category', secondary=post_categories, backref='posts', lazy='dynamic')
    tags = db.relationship('Tag', secondary=post_tags, backref='posts', lazy='dynamic')

    # 方法
    def increment_view_count(self):
        self.view_count += 1

    def to_dict(self):
        """转换为字典"""
        return {
            'id': self.id,
            'title': self.title,
            'slug': self.slug,
            'content': self.content,
            'excerpt': self.excerpt,
            'status': self.status,
            'view_count': self.view_count,
            'comment_count': self.comment_count,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'published_at': self.published_at.isoformat() if self.published_at else None,
            'author': self.author.to_dict() if self.author else None,
            'categories': [cat.to_dict() for cat in self.categories],
            'tags': [tag.to_dict() for tag in self.tags]
        }

    def __repr__(self):
        return f'<Post {self.title}>'

class Comment(db.Model):
    """评论模型"""
    __tablename__ = 'comments'

    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.Text, nullable=False)
    author_name = db.Column(db.String(100))  # 匿名评论者姓名
    author_email = db.Column(db.String(120))  # 匿名评论者邮箱
    author_ip = db.Column(db.String(45))  # IPv4或IPv6地址
    author_user_agent = db.Column(db.Text)

    # 状态
    status = db.Column(db.Enum('pending', 'approved', 'spam', 'trash'), default='pending')
    is_anonymous = db.Column(db.Boolean, default=False)

    # 时间
    created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # 外键
    post_id = db.Column(db.Integer, db.ForeignKey('posts.id'), nullable=False, index=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=True, index=True)
    parent_id = db.Column(db.Integer, db.ForeignKey('comments.id'), nullable=True)  # 父评论

    # 关系
    parent = db.relationship('Comment', remote_side=[id], backref='replies')

    def to_dict(self):
        """转换为字典"""
        data = {
            'id': self.id,
            'content': self.content,
            'status': self.status,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'post_id': self.post_id
        }

        if self.user:
            data['author'] = self.user.to_dict()
        else:
            data['author'] = {
                'name': self.author_name,
                'email': self.author_email,
                'is_anonymous': True
            }

        if self.parent_id:
            data['parent_id'] = self.parent_id

        return data

    def __repr__(self):
        return f'<Comment {self.id}>'

class Category(db.Model):
    """分类模型"""
    __tablename__ = 'categories'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False, index=True)
    slug = db.Column(db.String(50), unique=True, nullable=False, index=True)
    description = db.Column(db.Text)
    parent_id = db.Column(db.Integer, db.ForeignKey('categories.id'), nullable=True)

    # 关系
    parent = db.relationship('Category', remote_side=[id], backref='children')

    def to_dict(self):
        """转换为字典"""
        return {
            'id': self.id,
            'name': self.name,
            'slug': self.slug,
            'description': self.description,
            'parent_id': self.parent_id,
            'post_count': self.posts.count()
        }

    def __repr__(self):
        return f'<Category {self.name}>'

class Tag(db.Model):
    """标签模型"""
    __tablename__ = 'tags'

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True, nullable=False, index=True)
    slug = db.Column(db.String(50), unique=True, nullable=False, index=True)
    description = db.Column(db.Text)

    def to_dict(self):
        """转换为字典"""
        return {
            'id': self.id,
            'name': self.name,
            'slug': self.slug,
            'description': self.description,
            'post_count': self.posts.count()
        }

    def __repr__(self):
        return f'<Tag {self.name}>'

# 事件监听器
@event.listens_for(User, 'before_insert')
def generate_user_uuid(mapper, connection, target):
    if not target.uuid:
        target.uuid = str(uuid.uuid4())

@event.listens_for(Post, 'before_insert')
def set_published_at(mapper, connection, target):
    if target.status == 'published' and not target.published_at:
        target.published_at = datetime.utcnow()

8. 最佳实践

  • 命名规范:表名使用复数形式,字段名使用小写加下划线
  • 添加索引:为经常查询的字段添加索引,特别是外键字段
  • 使用合适的数据类型:根据数据特性选择最合适的字段类型
  • 添加约束:使用NOT NULL、UNIQUE等约束保证数据完整性
  • 使用时间戳:为重要记录添加created_at和updated_at字段
  • 软删除:重要数据使用软删除而不是物理删除
  • 规范化设计:遵循数据库设计三大范式,避免数据冗余
  • 添加注释:为表和字段添加注释,便于维护

9. 常见问题解答

  • String:用于长度有限且可预估的字符串,如用户名、邮箱、标题等
  • Text:用于长度不可预估的长文本,如文章内容、描述、评论等
  • 经验法则:如果字段长度可能超过255个字符,使用Text类型

  • select(默认):第一次访问时加载所有关联对象
  • dynamic:返回一个查询对象,可以继续过滤和分页
  • joined:使用JOIN立即加载关联对象
  • subquery:使用子查询立即加载关联对象
  • noload:不加载关联对象
  • raise:访问时抛出异常(用于防止意外加载)

  1. 为常用查询字段创建索引
  2. 使用select只查询需要的字段
  3. 使用joinedload避免N+1查询问题
  4. 合理使用数据库连接池
  5. 定期清理不需要的数据
  6. 对大表进行分区(数据库级别)