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:访问时抛出异常(用于防止意外加载)
- 为常用查询字段创建索引
- 使用
select只查询需要的字段
- 使用
joinedload避免N+1查询问题
- 合理使用数据库连接池
- 定期清理不需要的数据
- 对大表进行分区(数据库级别)