跳转至

代码实战项目6:简易博客系统

项目概述

博客系统是 Web 开发的经典项目,涵盖了用户认证、内容管理、交互功能等多个方面。本项目将使用 Python 的 Flask 框架和 SQLite 数据库实现一个功能完整的博客系统。

学习目标

  • 掌握 Flask Web 框架的基本使用
  • 学会使用 SQLAlchemy 进行数据库操作
  • 理解用户认证和会话管理
  • 掌握 RESTful API 设计
  • 学会前端模板渲染

功能需求

核心功能

  1. 用户管理
  2. 用户注册
  3. 用户登录
  4. 用户登出
  5. 个人信息管理

  6. 文章管理

  7. 发布文章
  8. 编辑文章
  9. 删除文章
  10. 文章列表
  11. 文章详情

  12. 评论系统

  13. 发表评论
  14. 删除评论
  15. 评论列表

  16. 标签和分类

  17. 文章标签
  18. 文章分类
  19. 按标签/分类筛选

  20. 其他功能

  21. 文章搜索
  22. 分页显示
  23. 富文本编辑

技术栈

  • 后端框架:Flask
  • 数据库:SQLite + SQLAlchemy
  • 模板引擎:Jinja2
  • 前端框架:Bootstrap 5
  • 富文本编辑器:TinyMCE

项目结构

blog_system/
├── app/
│   ├── __init__.py          # Flask 应用工厂
│   ├── models.py            # 数据库模型
│   ├── routes/              # 路由
│   │   ├── __init__.py
│   │   ├── auth.py          # 认证路由
│   │   ├── main.py          # 主页路由
│   │   ├── post.py          # 文章路由
│   │   └── api.py           # API 路由
│   ├── templates/           # 模板
│   │   ├── base.html        # 基础模板
│   │   ├── index.html       # 首页
│   │   ├── login.html       # 登录页
│   │   ├── register.html    # 注册页
│   │   ├── post_list.html   # 文章列表
│   │   ├── post_detail.html # 文章详情
│   │   └── create_post.html # 创建文章
│   ├── static/              # 静态文件
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   └── forms.py             # 表单类
├── config.py                # 配置文件
├── requirements.txt         # 依赖列表
└── run.py                   # 启动文件

代码实现

1. 配置文件(config.py)

import os

class Config:
    """配置类"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
        'sqlite:///blog.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    POSTS_PER_PAGE = 10

2. 数据库模型(models.py)

from datetime import datetime
from app import db
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin

class User(UserMixin, db.Model):
    """用户模型"""
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), index=True, unique=True)
    email = db.Column(db.String(120), index=True, unique=True)
    password_hash = db.Column(db.String(128))
    posts = db.relationship('Post', backref='author', lazy='dynamic')
    comments = db.relationship('Comment', backref='author', lazy='dynamic')

    def set_password(self, password):
        """设置密码"""
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        """验证密码"""
        return check_password_hash(self.password_hash, password)

    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(140))
    body = db.Column(db.Text)
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    category_id = db.Column(db.Integer, db.ForeignKey('category.id'))
    tags = db.relationship('Tag', secondary='post_tags',
                          backref=db.backref('posts', lazy='dynamic'))
    comments = db.relationship('Comment', backref='post', lazy='dynamic')

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

class Category(db.Model):
    """分类模型"""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)
    posts = db.relationship('Post', backref='category', lazy='dynamic')

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

class Tag(db.Model):
    """标签模型"""
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True)

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

class Comment(db.Model):
    """评论模型"""
    id = db.Column(db.Integer, primary_key=True)
    body = db.Column(db.Text)
    timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    post_id = db.Column(db.Integer, db.ForeignKey('post.id'))

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

# 多对多关系表
post_tags = db.Table('post_tags',
    db.Column('post_id', db.Integer, db.ForeignKey('post.id'), primary_key=True),
    db.Column('tag_id', db.Integer, db.ForeignKey('tag.id'), primary_key=True)
)

3. Flask 应用工厂(app/init.py)

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from config import Config

db = SQLAlchemy()
login = LoginManager()

def create_app(config_class=Config):
    """创建 Flask 应用"""
    app = Flask(__name__)
    app.config.from_object(config_class)

    # 初始化扩展
    db.init_app(app)
    login.init_app(app)
    login.login_view = 'auth.login'

    # 注册蓝图
    from app.routes import auth, main, post, api
    app.register_blueprint(auth.bp)
    app.register_blueprint(main.bp)
    app.register_blueprint(post.bp)
    app.register_blueprint(api.bp, url_prefix='/api')

    return app

4. 认证路由(app/routes/auth.py)

from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, current_user
from app import db
from app.models import User
from app.forms import LoginForm, RegistrationForm

bp = Blueprint('auth', __name__)

@bp.route('/login', methods=['GET', 'POST'])
def login():
    """用户登录"""
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        if user is None or not user.check_password(form.password.data):
            flash('无效的用户名或密码')
            return redirect(url_for('auth.login'))

        login_user(user, remember=form.remember_me.data)
        next_page = request.args.get('next')
        return redirect(next_page or url_for('main.index'))

    return render_template('login.html', title='登录', form=form)

@bp.route('/register', methods=['GET', 'POST'])
def register():
    """用户注册"""
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(username=form.username.data, email=form.email.data)
        user.set_password(form.password.data)
        db.session.add(user)
        db.session.commit()
        flash('注册成功!')
        return redirect(url_for('auth.login'))

    return render_template('register.html', title='注册', form=form)

@bp.route('/logout')
def logout():
    """用户登出"""
    logout_user()
    return redirect(url_for('main.index'))

5. 主页路由(app/routes/main.py)

from flask import Blueprint, render_template
from app.models import Post

bp = Blueprint('main', __name__)

@bp.route('/')
@bp.route('/index')
def index():
    """首页"""
    posts = Post.query.order_by(Post.timestamp.desc()).all()
    return render_template('index.html', posts=posts)

@bp.route('/category/<int:id>')
def category(id):
    """分类页面"""
    category = Category.query.get_or_404(id)
    posts = category.posts.order_by(Post.timestamp.desc()).all()
    return render_template('post_list.html', posts=posts, title=category.name)

6. 文章路由(app/routes/post.py)

from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_required, current_user
from app import db
from app.models import Post, Comment, Tag
from app.forms import PostForm, CommentForm

bp = Blueprint('post', __name__)

@bp.route('/post')
def post_list():
    """文章列表"""
    page = request.args.get('page', 1, type=int)
    posts = Post.query.order_by(Post.timestamp.desc()).paginate(
        page, 10, False)

    return render_template('post_list.html', posts=posts)

@bp.route('/post/<int:id>', methods=['GET', 'POST'])
def post_detail(id):
    """文章详情"""
    post = Post.query.get_or_404(id)
    form = CommentForm()

    if form.validate_on_submit() and current_user.is_authenticated:
        comment = Comment(body=form.body.data,
                         author=current_user,
                         post=post)
        db.session.add(comment)
        db.session.commit()
        flash('评论已发表')
        return redirect(url_for('post.post_detail', id=post.id))

    comments = post.comments.order_by(Comment.timestamp.desc()).all()
    return render_template('post_detail.html', post=post, form=form, comments=comments)

@bp.route('/create', methods=['GET', 'POST'])
@login_required
def create_post():
    """创建文章"""
    form = PostForm()
    if form.validate_on_submit():
        post = Post(title=form.title.data,
                   body=form.body.data,
                   author=current_user,
                   category=form.category.data)

        # 处理标签
        tag_names = form.tags.data.split(',')
        for tag_name in tag_names:
            tag_name = tag_name.strip()
            if tag_name:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name)
                post.tags.append(tag)

        db.session.add(post)
        db.session.commit()
        flash('文章已发布')
        return redirect(url_for('post.post_detail', id=post.id))

    return render_template('create_post.html', title='创建文章', form=form)

@bp.route('/post/<int:id>/edit', methods=['GET', 'POST'])
@login_required
def edit_post(id):
    """编辑文章"""
    post = Post.query.get_or_404(id)
    if post.author != current_user:
        flash('您没有权限编辑此文章')
        return redirect(url_for('post.post_detail', id=id))

    form = PostForm()
    if form.validate_on_submit():
        post.title = form.title.data
        post.body = form.body.data
        post.category = form.category.data

        # 更新标签
        post.tags.clear()
        tag_names = form.tags.data.split(',')
        for tag_name in tag_names:
            tag_name = tag_name.strip()
            if tag_name:
                tag = Tag.query.filter_by(name=tag_name).first()
                if not tag:
                    tag = Tag(name=tag_name)
                post.tags.append(tag)

        db.session.commit()
        flash('文章已更新')
        return redirect(url_for('post.post_detail', id=post.id))

    elif request.method == 'GET':
        form.title.data = post.title
        form.body.data = post.body
        form.category.data = post.category
        form.tags.data = ', '.join([tag.name for tag in post.tags])

    return render_template('create_post.html', title='编辑文章', form=form)

@bp.route('/post/<int:id>/delete', methods=['POST'])
@login_required
def delete_post(id):
    """删除文章"""
    post = Post.query.get_or_404(id)
    if post.author != current_user:
        flash('您没有权限删除此文章')
        return redirect(url_for('post.post_detail', id=id))

    db.session.delete(post)
    db.session.commit()
    flash('文章已删除')
    return redirect(url_for('main.index'))

7. 表单类(app/forms.py)

from flask_wtf import FlaskForm
from wtforms import StringField, TextAreaField, PasswordField, BooleanField, SelectField, SubmitField
from wtforms.validators import DataRequired, Email, EqualTo, ValidationError
from app.models import User, Category

class LoginForm(FlaskForm):
    """登录表单"""
    username = StringField('用户名', validators=[DataRequired()])
    password = PasswordField('密码', validators=[DataRequired()])
    remember_me = BooleanField('记住我')
    submit = SubmitField('登录')

class RegistrationForm(FlaskForm):
    """注册表单"""
    username = StringField('用户名', validators=[DataRequired()])
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired()])
    password2 = PasswordField('重复密码', validators=[DataRequired(), EqualTo('password')])
    submit = SubmitField('注册')

    def validate_username(self, username):
        user = User.query.filter_by(username=username.data).first()
        if user is not None:
            raise ValidationError('用户名已被使用')

    def validate_email(self, email):
        user = User.query.filter_by(email=email.data).first()
        if user is not None:
            raise ValidationError('邮箱已被使用')

class PostForm(FlaskForm):
    """文章表单"""
    title = StringField('标题', validators=[DataRequired()])
    body = TextAreaField('内容', validators=[DataRequired()])
    category = SelectField('分类', coerce=int)
    tags = StringField('标签(用逗号分隔)')
    submit = SubmitField('发布')

    def __init__(self, *args, **kwargs):
        super(PostForm, self).__init__(*args, **kwargs)
        self.category.choices = [(c.id, c.name) for c in Category.query.all()]

class CommentForm(FlaskForm):
    """评论表单"""
    body = TextAreaField('评论', validators=[DataRequired()])
    submit = SubmitField('发表评论')

8. 启动文件(run.py)

from app import create_app, db
from app.models import User, Post, Category

app = create_app()

@app.shell_context_processor
def make_shell_context():
    """Shell 上下文"""
    return {'db': db, 'User': User, 'Post': Post, 'Category': Category}

if __name__ == '__main__':
    app.run(debug=True)

9. 依赖文件(requirements.txt)

Flask==2.3.0
Flask-SQLAlchemy==3.0.5
Flask-Login==0.6.2
Flask-WTF==1.1.1
email-validator==2.0.0

运行指南

1. 安装依赖

pip install -r requirements.txt

2. 初始化数据库

python
>>> from app import create_app, db
>>> from app.models import User, Post, Category
>>> app = create_app()
>>> with app.app_context():
...     db.create_all()
...     # 创建默认分类
...     tech = Category(name='技术')
...     life = Category(name='生活')
...     db.session.add(tech)
...     db.session.add(life)
...     db.session.commit()

3. 运行应用

python run.py

访问 http://127.0.0.1:5000

扩展功能建议

基础扩展

  1. 富文本编辑
  2. 集成 TinyMCE 或 CKEditor
  3. 支持图片上传

  4. 用户资料

  5. 头像上传
  6. 个人简介
  7. 文章归档

  8. 社交功能

  9. 文章点赞
  10. 用户关注
  11. 消息通知

进阶扩展

  1. 全文搜索
  2. 集成 Whoosh 或 Elasticsearch
  3. 支持高级搜索

  4. RSS 订阅

  5. 生成 RSS feed
  6. 支持 RSS 订阅

  7. API 接口

  8. RESTful API
  9. 支持移动应用

高级扩展

  1. 缓存系统
  2. Redis 缓存
  3. 提高性能

  4. 后台管理

  5. Flask-Admin
  6. 用户权限管理

  7. 部署优化

  8. Nginx + Gunicorn
  9. 数据库优化

常见问题

Q: 如何修改数据库?

A: 使用 Flask-Migrate 进行数据库迁移,避免直接修改数据库结构。

Q: 如何添加新的分类?

A: 可以在管理后台添加,或者通过 Flask shell 手动添加。

Q: 支持 Markdown 吗?

A: 可以集成 Flask-Markdown 或 Markdown 插件。

Q: 如何部署到生产环境?

A: 建议使用 Gunicorn + Nginx,并配置正确的环境变量。

项目总结

本项目实现了一个功能完整的博客系统,涵盖了:

  1. Web 开发:Flask 框架的应用
  2. 数据库设计:SQLAlchemy ORM
  3. 用户认证:Flask-Login 会话管理
  4. 表单处理:Flask-WTF 表单验证
  5. 模板渲染:Jinja2 模板引擎
  6. RESTful 设计:清晰的 API 结构

通过本项目,你将掌握: - Flask Web 开发 - 数据库设计和操作 - 用户认证和授权 - 前端模板设计 - RESTful API 设计

下一步

尝试添加更多功能,如: - 富文本编辑器 - 图片上传功能 - 文章点赞系统 - RSS 订阅 - 全文搜索 - 用户关注功能 - 消息通知系统


祝你开发愉快!