实战篇 / 第 24 章

第 24 章 · Web 应用全栈开发

需求设计：先把"要做什么"写清楚

工程的第一步不是装依赖，而是把需求、数据模型、接口契约三件事定下来。需求含糊会让后面每一次重构都在还债。

1.1 功能清单

1.2 核心实体关系

User 1 ──< Task >── 1 Project
 │               │
 │               └──< Tag >── Task      (多对多，通过 task_tags 关联表)
 │
 └──< Comment >── Task
 └──< Notification

+-----------------+        +------------------+
|      User       |        |      Task        |
+-----------------+        +------------------+
| id              |        | id               |
| username (uniq) |        | title            |
| email    (uniq) |        | description      |
| password_hash   |        | status  (todo/   |
| role            |        |  doing/done)     |
| is_active       |        | priority (low/   |
| created_at      |        |  mid/high/urgent)|
+-----------------+        | due_date         |
                         | project_id (FK)   |
                         | owner_id   (FK)   |
                         | created_at /      |
                         | updated_at        |
                         +------------------+

1.3 接口契约（先写死，前后端并行开发）

项目骨架：应用工厂与蓝图

2.1 目录结构

taskboard/
├── pyproject.toml
├── wsgi.py                  # 部署入口
├── migrations/              # Alembic 自动生成
│   ├── env.py
│   ├── alembic.ini
│   └── versions/
├── taskboard/
│   ├── __init__.py          # create_app() 工厂
│   ├── config.py            # 各类配置类
│   ├── extensions.py        # 延迟初始化的扩展单例
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py
│   │   ├── task.py
│   │   └── tag.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── tasks.py
│   │   ├── comments.py
│   │   ├── admin.py
│   │   └── errors.py        # 统一错误响应
│   ├── web/
│   │   ├── __init__.py
│   │   ├── views.py         # 服务端渲染页面
│   │   └── forms.py
│   ├── templates/
│   ├── static/
│   │   ├── css/app.css
│   │   └── js/board.js
│   ├── services/
│   │   ├── tasks.py         # 业务逻辑，与路由分离
│   │   └── notifications.py
│   └── utils/
│       ├── security.py      # 密码哈希、令牌签发
│       ├── pagination.py
│       └── responses.py
└── tests/
    ├── conftest.py
    ├── test_auth_api.py
    ├── test_tasks_api.py
    └── test_web_views.py

2.2 应用工厂：为什么不用全局 app

教程里常见的写法是 app = Flask(__name__) 写在模块顶层，这带来三个问题：① 测试时无法为每个用例创建干净实例；② 配置被硬编码，换环境要改代码；③ 蓝图和扩展无法按需注册。应用工厂（application factory）解决了全部三个。

"""taskboard/config.py —— 配置分环境，敏感信息走环境变量。"""
from __future__ import annotations

import os
from datetime import timedelta


class BaseConfig:
    """公共配置。所有环境共享。"""
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-only-insecure-key")
    JSON_SORT_KEYS = False
    JSONIFY_PRETTYPRINT_REGULAR = True

    # 会话 Cookie 安全
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = "Lax"
    REMEMBER_COOKIE_HTTPONLY = True

    # 分页默认值
    ITEMS_PER_PAGE = 20
    MAX_ITEMS_PER_PAGE = 100

    SQLALCHEMY_TRACK_MODIFICATIONS = False

    @staticmethod
    def init_app(app) -> None:
        pass


class DevelopmentConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "sqlite:///taskboard_dev.db")
    TEMPLATES_AUTO_RELOAD = True


class TestingConfig(BaseConfig):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = os.environ.get("TEST_DATABASE_URL", "sqlite://")
    WTF_CSRF_ENABLED = False           # 测试里关掉 CSRF
    JWT_SECRET_KEY = "test-secret-key"
    LOGIN_DISABLED = False


class ProductionConfig(BaseConfig):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get("DATABASE_URL")   # 无默认，强制外部注入
    SESSION_COOKIE_SECURE = True        # 生产必须 HTTPS
    JWT_ACCESS_TOKEN_EXPIRES = timedelta(minutes=30)
    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=14)


config_map = {
    "development": DevelopmentConfig,
    "testing": TestingConfig,
    "production": ProductionConfig,
    "default": DevelopmentConfig,
}


def get_config(name: str | None = None):
    key = name or os.environ.get("FLASK_ENV", "default")
    cfg = config_map.get(key)
    if cfg is None:
        raise ValueError(f"未知配置环境: {key}，可选: {list(config_map)}")
    if key == "production" and not os.environ.get("DATABASE_URL"):
        raise RuntimeError("生产环境必须设置 DATABASE_URL 环境变量")
    return cfg

"""taskboard/extensions.py —— 扩展对象在此声明，但在工厂里绑定。

Flask 扩展在模块导入时创建（不绑定 app），在 create_app() 里 init_app() 关联。
这样测试可以创建多个互相隔离的 app 实例。
"""
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from flask_jwt_extended import JWTManager

db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()
jwt = JWTManager()


@login_manager.user_loader
def load_user(user_id: str):
    """Flask-Login 要求提供：给我 user_id，我还你 user 对象。"""
    from .models.user import User
    return db.session.get(User, int(user_id))

"""taskboard/__init__.py —— 应用工厂。"""
from __future__ import annotations

import logging
import os

from flask import Flask, jsonify

from .config import get_config
from .extensions import csrf, db, jwt, login_manager, migrate


def create_app(config_name: str | None = None) -> Flask:
    """应用工厂：创建一个配置齐全的 Flask 应用。

    Returns:
        已注册扩展、蓝图与错误处理器的 Flask 实例
    """
    app = Flask(__name__, instance_relative_config=True)

    cfg = get_config(config_name)
    app.config.from_object(cfg)
    cfg.init_app(app)

    # 确保实例目录存在
    os.makedirs(app.instance_path, exist_ok=True)

    _init_extensions(app)
    _register_blueprints(app)
    _register_error_handlers(app)
    _register_commands(app)

    # 日志
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(
        "%(asctime)s %(levelname)s [%(name)s] %(message)s"))
    app.logger.handlers.clear()
    app.logger.addHandler(handler)
    app.logger.setLevel(logging.DEBUG if app.debug else logging.INFO)

    app.logger.info("应用已创建，环境=%s", config_name or os.environ.get("FLASK_ENV", "default"))
    return app


def _init_extensions(app: Flask) -> None:
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)
    csrf.init_app(app)
    jwt.init_app(app)

    # 确保模型被导入，SQLAlchemy 才能在 metadata 中注册表
    from . import models  # noqa: F401


def _register_blueprints(app: Flask) -& None:
    from .api.auth import bp as auth_api_bp
    from .api.tasks import bp as tasks_api_bp
    from .api.comments import bp as comments_api_bp
    from .api.admin import bp as admin_api_bp
    from .web.views import bp as web_bp

    app.register_blueprint(auth_api_bp, url_prefix="/api/v1/auth")
    app.register_blueprint(tasks_api_bp, url_prefix="/api/v1/tasks")
    app.register_blueprint(comments_api_bp, url_prefix="/api/v1")
    app.register_blueprint(admin_api_bp, url_prefix="/api/v1/admin")
    app.register_blueprint(web_bp)              # 页面路由不带前缀


def _register_error_handlers(app: Flask) -> None:
    """把 HTTP 异常统一转成 JSON（API 场景）或渲染页面（网页场景）。"""
    from werkzeug.exceptions import HTTPException

    @app.errorhandler(HTTPException)
    def handle_http_error(exc: HTTPException):
        from flask import request
        if request.path.startswith("/api/"):
            return jsonify({
                "error": {
                    "code": exc.code,
                    "name": exc.name,
                    "message": exc.description,
                }
            }), exc.code
        return exc

    @app.errorhandler(404)
    def not_found(exc):
        return jsonify({"error": {"code": 404, "message": "资源不存在"}}), 404

    @app.errorhandler(500)
    def server_error(exc):
        db.session.rollback()          # 关键：出错后回滚，否则连接池被占死
        app.logger.exception("未处理的服务端异常")
        return jsonify({"error": {"code": 500, "message": "服务器内部错误"}}), 500


def _register_commands(app: Flask) -> None:
    """注册 flask 自定义 CLI 命令。"""
    import click

    @app.cli.command("init-db")
    @click.option("--drop", is_flag=True, help="先删除所有表再创建")
    def init_db(drop: bool) -> None:
        """初始化数据库（开发用，生产请用迁移）。"""
        if drop:
            db.drop_all()
            click.echo("已删除所有表")
        db.create_all()
        click.echo("表已创建")

    @app.cli.command("seed")
    def seed() -> None:
        """填充演示数据。"""
        from .models import User, Task, Project
        from .utils.security import hash_password

        if db.session.execute(db.select(User).limit(1)).scalar_one_or_none():
            click.echo("数据库已有数据，跳过")
            return
        alice = User(username="alice", email="alice@example.com",
                     password_hash=hash_password("alice12345"))
        proj = Project(name="重构支付模块", owner_id=1, is_public=True)
        db.session.add_all([alice, proj])
        db.session.flush()
        for i in range(5):
            db.session.add(Task(title=f"任务 {i+1}", description="演示数据",
                                owner_id=alice.id, project_id=proj.id))
        db.session.commit()
        click.echo("演示数据已写入")

2.3 蓝图（blueprint）：把大应用切成模块

"""taskboard/api/__init__.py —— 蓝图工厂。"""
from __future__ import annotations

from flask import Blueprint


def make_api_bp(name: str, url_prefix: str | None = None) -> Blueprint:
    """创建 API 蓝图，统一配置 JSON 行为与蓝图级错误。"""
    bp = Blueprint(name, __name__, url_prefix=url_prefix)

    @bp.after_request
    def add_headers(resp):
        # 安全响应头
        resp.headers["X-Content-Type-Options"] = "nosniff"
        resp.headers["X-Frame-Options"] = "DENY"
        resp.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
        return resp

    return bp


# 各模块的公共 bp 实例
from .auth import bp as auth_bp          # noqa: E402
from .tasks import bp as tasks_bp        # noqa: E402

2.4 入口文件

"""wsgi.py —— 生产部署入口（Gunicorn 会找这个）。"""
from taskboard import create_app

app = create_app("production")


if __name__ == "__main__":
    # 仅本地调试用；生产请用 Gunicorn / uWSGI
    app.run(host="0.0.0.0", port=8000)

2.5 扩展选型表

数据模型：ORM 设计实战

Flask-SQLAlchemy 3.x 使用 SQLAlchemy 2.0 风格的 DeclarativeBase，配合 Mapped / mapped_column 获得完整类型提示。

3.1 扩展 User 模型

"""taskboard/models/user.py"""
from __future__ import annotations

from datetime import datetime, timezone

from flask_login import UserMixin
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy.orm import Mapped, mapped_column, relationship

from ..extensions import db


def utcnow() -> datetime:
    """统一时间源：带时区的 UTC。

    坑：MySQL 的 DATETIME 不存时区，如果这里存 aware datetime
    而读取时用 naive 比较会报 TypeError。统一存 UTC aware。
    """
    return datetime.now(timezone.utc)


class User(db.Model, UserMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(String(64), unique=True,
                                          nullable=False, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False,
                                        index=True)
    password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
    role: Mapped[str] = mapped_column(String(16), default="user", nullable=False)
    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
    avatar_url: Mapped[str | None] = mapped_column(String(512), nullable=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
                                                 default=utcnow, nullable=False)
    last_login_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True)

    # 关系：backref 双向，lazy 控制加载策略
    tasks: Mapped[list["Task"]] = relationship(  # noqa: F821
        back_populates="owner", cascade="all, delete-orphan",
        # selectin 一次性加载，避免 N+1 查询
        lazy="selectin",
    )
    projects: Mapped[list["Project"]] = relationship(  # noqa: F821
        back_populates="owner", cascade="all, delete-orphan", lazy="selectin")

    def set_password(self, raw: str) -> None:
        from ..utils.security import hash_password
        self.password_hash = hash_password(raw)

    def check_password(self, raw: str) -> bool:
        from ..utils.security import verify_password
        return verify_password(self.password_hash, raw)

    def is_admin(self) -> bool:
        return self.role == "admin"

    def to_dict(self) -> dict:
        """序列化的唯一出口。永远不要把 password_hash 放进响应。"""
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "role": self.role,
            "avatar_url": self.avatar_url,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }

    def __repr__(self) -> str:
        return f"<User {self.username}>"

3.2 Task 模型：枚举、外键、索引与多对多

"""taskboard/models/task.py"""
from __future__ import annotations

import enum
from datetime import date, datetime

from sqlalchemy import (
    CheckConstraint, Column, Date, DateTime, Enum as SAEnum, ForeignKey,
    Index, Integer, String, Table, Text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship

from ..extensions import db
from .user import utcnow


class TaskStatus(str, enum.Enum):
    """用 str + enum.Enum：既是枚举，又能直接当字符串存与比较。"""
    TODO = "todo"
    DOING = "doing"
    DONE = "done"
    CANCELLED = "cancelled"


class Priority(str, enum.Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    URGENT = "urgent"


# 多对多关联表：显式声明，好处是可加额外列（如 added_at、added_by）
task_tags = Table(
    "task_tags",
    db.metadata,
    Column("task_id", ForeignKey("tasks.id", ondelete="CASCADE"),
           primary_key=True),
    Column("tag_id", ForeignKey("tags.id", ondelete="CASCADE"),
           primary_key=True),
    Column("added_at", DateTime(timezone=True), default=utcnow),
    # 复合主键天然去重
    Index("ix_task_tags_tag", "tag_id"),
)


class Project(db.Model):
    __tablename__ = "projects"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(120), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)
    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id",
                                                      ondelete="CASCADE"),
                                           nullable=False, index=True)
    is_public: Mapped[bool] = mapped_column(default=False, nullable=False)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
                                                 default=utcnow, nullable=False)

    owner: Mapped["User"] = relationship(back_populates="projects")
    tasks: Mapped[list["Task"]] = relationship(
        back_populates="project", cascade="all, delete-orphan", lazy="selectin")

    def to_dict(self, with_tasks: bool = False) -> dict:
        data = {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "is_public": self.is_public,
            "owner": self.owner.username if self.owner else None,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }
        if with_tasks:
            data["tasks"] = [t.to_dict() for t in self.tasks]
        return data


class Task(db.Model):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    description: Mapped[str | None] = mapped_column(Text, nullable=True)

    # 用 SAEnum 存枚举值，数据库层面就有约束
    status: Mapped[TaskStatus] = mapped_column(
        SAEnum(TaskStatus, native_enum=False, length=16),
        default=TaskStatus.TODO, nullable=False, index=True)
    priority: Mapped[Priority] = mapped_column(
        SAEnum(Priority, native_enum=False, length=16),
        default=Priority.MEDIUM, nullable=False)

    progress: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
    due_date: Mapped[date | None] = mapped_column(Date, nullable=True)
    estimated_hours: Mapped[float | None] = mapped_column(nullable=True)
    position: Mapped[int] = mapped_column(Integer, default=0,
                                           nullable=False)   # 看板内排序

    owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"),
                                           nullable=False, index=True)
    project_id: Mapped[int | None] = mapped_column(
        ForeignKey("projects.id", ondelete="CASCADE"), nullable=True, index=True)

    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
                                                 default=utcnow, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True),
                                                 default=utcnow, onupdate=utcnow,
                                                 nullable=False)
    completed_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True)

    # 关系
    owner: Mapped["User"] = relationship(back_populates="tasks")
    project: Mapped["Project | None"] = relationship(back_populates="tasks")
    tags: Mapped[list["Tag"]] = relationship(secondary=task_tags,
                                            back_populates="tasks",
                                            lazy="selectin")
    comments: Mapped[list["Comment"]] = relationship(  # noqa: F821
        back_populates="task", cascade="all, delete-orphan",
        lazy="select", order_by="Comment.created_at")

    # 表级约束与索引：写在一处，可读性好
    __table_args__ = (
        CheckConstraint("progress >= 0 AND progress <= 100",
                        name="ck_tasks_progress_range"),
        CheckConstraint("title != ''", name="ck_tasks_title_nonempty"),
        Index("ix_tasks_owner_status", "owner_id", "status"),   # 复合索引
        Index("ix_tasks_due", "due_date"),
    )

    # 状态机：不允许从 done 直接跳回 todo
    ALLOWED_TRANSITIONS: dict[TaskStatus, set[TaskStatus]] = {
        TaskStatus.TODO: {TaskStatus.DOING, TaskStatus.CANCELLED, TaskStatus.DONE},
        TaskStatus.DOING: {TaskStatus.TODO, TaskStatus.DONE, TaskStatus.CANCELLED},
        TaskStatus.DONE: {TaskStatus.DOING},
        TaskStatus.CANCELLED: {TaskStatus.TODO},
    }

    def can_transition_to(self, new_status: TaskStatus) -> bool:
        return new_status in self.ALLOWED_TRANSITIONS.get(self.status, set())

    def mark_done(self) -> None:
        if not self.can_transition_to(TaskStatus.DONE):
            raise ValueError(f"不允许从 {self.status.value} 转为 done")
        self.status = TaskStatus.DONE
        self.progress = 100
        self.completed_at = utcnow()

    @property
    def is_overdue(self) -> bool:
        return (self.due_date is not None
                and self.status not in {TaskStatus.DONE, TaskStatus.CANCELLED}
                and self.due_date < date.today())

    def to_dict(self, detail: bool = False) -> dict:
        data = {
            "id": self.id,
            "title": self.title,
            "status": self.status.value,
            "priority": self.priority.value,
            "progress": self.progress,
            "owner": self.owner.username if self.owner else None,
            "due_date": self.due_date.isoformat() if self.due_date else None,
            "is_overdue": self.is_overdue,
            "tags": [{"id": t.id, "name": t.name} for t in self.tags],
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }
        if detail:
            data.update({
                "description": self.description,
                "project": self.project.to_dict() if self.project else None,
                "estimated_hours": self.estimated_hours,
                "position": self.position,
                "updated_at": self.updated_at.isoformat() if self.updated_at else None,
                "completed_at": (self.completed_at.isoformat()
                                 if self.completed_at else None),
                "comment_count": len(self.comments),
            })
        return data

    def __repr__(self) -> str:
        return f"<Task {self.id} {self.title[:20]!r} {self.status.value}>"

"""taskboard/models/tag.py"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import String, Table, Column, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship

from ..extensions import db
from .user import utcnow


class Tag(db.Model):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(40), unique=True,
                                      nullable=False, index=True)
    color: Mapped[str] = mapped_column(String(7), default="#6b7280")
    created_at: Mapped[datetime] = mapped_column(db.DateTime(timezone=True),
                                                 default=utcnow)

    tasks: Mapped[list["Task"]] = relationship(  # noqa: F821
        secondary="task_tags", back_populates="tags")

    def to_dict(self) -> dict:
        return {"id": self.id, "name": self.name, "color": self.color}

"""taskboard/models/__init__.py —— 统一导出，避免循环引用。"""
from .user import User, utcnow
from .task import Task, TaskStatus, Priority, Project, task_tags
from .tag import Tag
from .comment import Comment, Notification

__all__ = ["User", "Task", "TaskStatus", "Priority", "Project", "task_tags",
           "Tag", "Comment", "Notification", "utcnow"]

"""taskboard/models/comment.py"""
from __future__ import annotations

from datetime import datetime

from sqlalchemy import ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from ..extensions import db
from .user import utcnow


class Comment(db.Model):
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True)
    body: Mapped[str] = mapped_column(Text, nullable=False)
    is_edited: Mapped[bool] = mapped_column(default=False, nullable=False)

    task_id: Mapped[int] = mapped_column(ForeignKey("tasks.id",
                                                     ondelete="CASCADE"),
                                          nullable=False, index=True)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id",
                                                      ondelete="CASCADE"),
                                           nullable=False, index=True)
    created_at: Mapped[datetime] = mapped_column(db.DateTime(timezone=True),
                                                 default=utcnow, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(db.DateTime(timezone=True),
                                                 default=utcnow, onupdate=utcnow)

    task: Mapped["Task"] = relationship(back_populates="comments")  # noqa: F821
    author: Mapped["User"] = relationship()

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "body": self.body,
            "author": self.author.username if self.author else None,
            "is_edited": self.is_edited,
            "created_at": self.created_at.isoformat(),
        }


class Notification(db.Model):
    __tablename__ = "notifications"

    id: Mapped[int] = mapped_column(primary_key=True)
    message: Mapped[str] = mapped_column(String(255), nullable=False)
    is_read: Mapped[bool] = mapped_column(default=False, nullable=False)

    user_id: Mapped[int] = mapped_column(ForeignKey("users.id",
                                                     ondelete="CASCADE"),
                                          nullable=False, index=True)
    task_id: Mapped[int | None] = mapped_column(
        ForeignKey("tasks.id", ondelete="CASCADE"), nullable=True)

    created_at: Mapped[datetime] = mapped_column(db.DateTime(timezone=True),
                                                 default=utcnow, nullable=False)

    user: Mapped["User"] = relationship()

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "message": self.message,
            "is_read": self.is_read,
            "task_id": self.task_id,
            "created_at": self.created_at.isoformat(),
        }

3.3 模型设计要点表

数据库迁移：Alembic 正确用法

db.create_all() 只在建表时好用，一旦表上线再改字段就完全失效。Alembic 通过"版本化的迁移脚本"解决这个问题。

4.1 初始化与工作流

$ flask db init                # 生成 migrations/ 目录
$ flask db migrate -m "init: users, projects, tasks, tags, comments"
$ flask db upgrade               # 应用迁移到最新版本

# 改模型后：
$ flask db migrate -m "add estimated_hours to tasks"
$ flask db upgrade

# 出问题了回滚一步：
$ flask db downgrade -1

# 查看当前版本与历史：
$ flask db current
$ flask db history

4.2 手写迁移脚本（生成后一定要读一遍）

"""migrations/versions/0002_add_task_estimated_hours.py"""
from __future__ import annotations

from alembic import op
import sqlalchemy as sa

revision = "0002"
down_revision = "0001"


def upgrade() -> None:
    op.add_column("tasks",
                  sa.Column("estimated_hours", sa.Float(), nullable=True))
    # 新建索引：加字段顺手加索引，否则大表上很慢
    op.create_index("ix_tasks_estimated_hours", "tasks", ["estimated_hours"])
    # 数据回填：老数据给个默认值
    op.execute("UPDATE tasks SET estimated_hours = 0 WHERE estimated_hours IS NULL")


def downgrade() -> None:
    op.drop_index("ix_tasks_estimated_hours", table_name="tasks")
    op.drop_column("tasks", "estimated_hours")

4.3 迁移的红线

1. 永远读一遍自动生成的脚本：Alembic 猜的列类型、外键、约束经常不准确，尤其是复杂的关系表变更；
2. 不在一个迁移里既改 schema 又做危险数据操作，拆成两步，便于出问题时单独回滚；
3. 加列要有默认值或允许 NULL，否则大表上 ALTER TABLE 会锁表；
4. 删列分两次发布：第一次停止写入该列，第二次再 DROP；
5. 迁移脚本要与代码同步部署：先跑迁移再上新代码（或反之），但要保证任一时刻应用都不崩。

鉴权：密码、JWT 与访问控制

5.1 密码哈希

"""taskboard/utils/security.py"""
from __future__ import annotations

import secrets
import string
from datetime import datetime, timedelta, timezone
from typing import Any

import jwt
from flask import current_app
from werkzeug.security import check_password_hash, generate_password_hash

from ..models import utcnow

ALGORITHM = "HS256"


# ---- 密码 ----
def hash_password(raw: str, method: str = "scrypt:32768:8:1") -> str:
    """生成密码哈希。默认用 scrypt（werkzeug 内置，内存硬，抗 GPU 破解）。

    scrypt 的参数含义：N=32768 内存占用, r=8 块大小, p=1 并行度。
    method 里带着参数，所以升级算法只需给新密码用新方法，
    旧密码验证时仍按它自己哈希串里记录的方法来验。
    """
    return generate_password_hash(raw, method=method, salt_length=16)


def verify_password(stored_hash: str, raw: str) -> bool:
    try:
        return check_password_hash(stored_hash, raw)
    except (ValueError, TypeError):
        return False      # 哈希串损坏时安全降级


def password_strength(pwd: str) -> tuple[bool, list[str]]:
    """简单的强度检查，返回 (是否通过, 问题列表)。"""
    issues: list[str] = []
    if len(pwd) < 8:
        issues.append("长度至少 8 位")
    if not any(c.isdigit() for c in pwd):
        issues.append("需要包含数字")
    if not any(c.isalpha() for c in pwd):
        issues.append("需要包含字母")
    if pwd.lower() in {"password", "12345678", "qwerty", "admin123"}:
        issues.append("过于常见")
    return (len(issues) == 0), issues


# ---- 令牌 ----
def create_access_token(user_id: int, extra: dict[str, Any] | None = None) -> str:
    now = utcnow()
    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + timedelta(minutes=30),
        "type": "access",
        "jti": secrets.token_hex(8),      # 令牌唯一 ID，便于以后做黑名单
        **(extra or {}),
    }
    return jwt.encode(payload, current_app.config["SECRET_KEY"],
                      algorithm=ALGORITHM)


def create_refresh_token(user_id: int) -> str:
    now = utcnow()
    payload = {
        "sub": str(user_id),
        "iat": now,
        "exp": now + timedelta(days=14),
        "type": "refresh",
        "jti": secrets.token_hex(8),
    }
    return jwt.encode(payload, current_app.config["SECRET_KEY"],
                      algorithm=ALGORITHM)


def decode_token(token: str, expected_type: str = "access") -> dict:
    """解码并校验令牌。失败时抛统一的 InvalidTokenError。"""
    try:
        payload = jwt.decode(token, current_app.config["SECRET_KEY"],
                             algorithms=[ALGORITHM])
    except jwt.ExpiredSignatureError as exc:
        raise InvalidTokenError("令牌已过期，请重新登录") from exc
    except jwt.InvalidTokenError as exc:
        raise InvalidTokenError("无效的令牌") from exc
    if payload.get("type") != expected_type:
        raise InvalidTokenError(f"期望 {expected_type} 令牌")
    return payload


class InvalidTokenError(Exception):
    pass


def generate_api_key(length: int = 32) -> str:
    """生成 API Key（用于机器人/服务间调用）。"""
    alphabet = string.ascii_letters + string.digits
    return "".join(secrets.choice(alphabet) for _ in range(length))


if __name__ == "__main__":
    h = hash_password("alice12345")
    print("哈希:", h[:40], "...")
    print("长度:", len(h))
    print("验证正确密码:", verify_password(h, "alice12345"))    # True
    print("验证错误密码:", verify_password(h, "wrong"))         # False
    # 两次 hash 同一密码结果不同（因为 salt 不同）
    print("两次哈希是否相同:", h == hash_password("alice12345"))  # False

5.2 两种鉴权方案并存

真实项目常常两者都用：页面用 Session，API 用 JWT。

"""taskboard/utils/guards.py —— 访问控制装饰器。"""
from __future__ import annotations

from functools import wraps
from typing import Callable, TypeVar

from flask import abort, g, jsonify, redirect, request, url_for

from ..models import User

F = TypeVar("F", bound=Callable)


def api_auth(fn: F) -> F:
    """要求 API 请求带有效 Bearer 令牌，否则返回 401。"""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer "):
            return jsonify({"error": {"code": 401, "message": "缺少令牌"}}), 401
        token = auth[7:]
        try:
            payload = decode_token(token, expected_type="access")
        except InvalidTokenError as exc:
            return jsonify({"error": {"code": 401, "message": str(exc)}}), 401
        user = db.session.get(User, int(payload["sub"]))
        if user is None or not user.is_active:
            return jsonify({"error": {"code": 401, "message": "用户不存在或已封禁"}}), 401
        g.current_user = user
        return fn(*args, **kwargs)
    return wrapper  # type: ignore[return-value]


def admin_required(fn: F) -> F:
    """要求当前用户是管理员（页面版本：跳 403）。"""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        user = g.get("current_user") or getattr(request, "user", None)
        if user is None:
            abort(401, description="请先登录")
        if not user.is_admin():
            abort(403, description="需要管理员权限")
        return fn(*args, **kwargs)
    return wrapper  # type: ignore[return-value]


def owner_or_admin_required(attr: str = "owner_id") -> Callable:
    """要求当前用户是资源属主或管理员。用法：

        @app.route("/tasks/<int:tid>/delete")
        @api_auth
        @owner_or_admin_required()
        def delete_task(tid): ...
    """
    def decorator(fn: F) -> F:
        @wraps(fn)
        def wrapper(tid: int, *args, **kwargs):
            user = g.current_user
            task = db.session.get(Task, tid)
            if task is None:
                abort(404, description="任务不存在")
            if task.owner_id != user.id and not user.is_admin():
                abort(403, description="无权操作他人的任务")
            g.task = task
            return fn(tid, *args, **kwargs)
        return wrapper  # type: ignore[return-value]
    return decorator

RESTful API 设计

6.1 RESTful 六条原则

6.2 状态码选择表

6.3 统一响应封装

"""taskboard/utils/responses.py"""
from __future__ import annotations

import math
from typing import Any

from flask import jsonify, request, url_for


def ok(data: Any = None, **extra):
    """200 成功响应。列表带 meta，详情带 data。"""
    payload: dict[str, Any] = {}
    if isinstance(data, list):
        payload["data"] = data
    elif data is not None:
        payload["data"] = data
    payload.update(extra)
    return jsonify(payload), 200


def created(data: Any, location: str | None = None):
    """201 创建成功。Location 头指向新资源，让客户端知道去哪儿取。"""
    resp = jsonify({"data": data})
    if location:
        resp.headers["Location"] = location
    return resp, 201


def no_content():
    """204 删除/标记成功，无响应体。"""
    return "", 204


def error(code: int, message: str, *, detail: Any = None,
          field_errors: dict | None = None):
    """统一错误格式。所有 API 错误都走这个函数，前端只需处理一种结构。"""
    body: dict[str, Any] = {"error": {"code": code, "message": message}}
    if detail is not None:
        body["error"]["detail"] = detail
    if field_errors:
        body["error"]["field_errors"] = field_errors
    return jsonify(body), code


def paginated(items: list, page: int, per_page: int, total: int):
    """分页响应：items + meta。meta 里给出翻页所需的全部信息。"""
    return ok(
        items,
        meta={
            "page": page,
            "per_page": per_page,
            "total": total,
            "pages": math.ceil(total / per_page) if per_page else 0,
            "has_next": page * per_page < total,
            "has_prev": page > 1,
        },
    )

6.4 分页、排序与过滤

"""taskboard/utils/pagination.py —— 把查询字符串变成 SQLAlchemy 查询。"""
from __future__ import annotations

from dataclasses import dataclass

from flask import current_app, request
from sqlalchemy import Select, asc, desc, func, select

# 白名单：只有列表里的字段才允许排序，防止 SQL 注入与误用
SORTABLE = {"created_at", "updated_at", "due_date", "priority",
            "title", "status", "progress"}


@dataclass(frozen=True)
class PageParams:
    page: int
    per_page: int
    sort_by: str
    order: str


def get_page_params() -> PageParams:
    """从 ?page=&per_page=&sort=&order= 提取分页与排序参数，并做边界收敛。"""
    max_per = current_app.config.get("MAX_ITEMS_PER_PAGE", 100)
    default_per = current_app.config.get("ITEMS_PER_PAGE", 20)

    # page 至少为 1；非法值（0、负数、非数字）都退回 1
    raw_page = request.args.get("page", 1, type=int)
    page = max(1, raw_page or 1)

    # per_page 上限强制截断，防止 ?per_page=999999 拖垮数据库
    raw_per = request.args.get("per_page", default_per, type=int)
    per_page = min(max_per, max(1, raw_per or default_per))

    sort_by = request.args.get("sort", "created_at")
    if sort_by not in SORTABLE:
        sort_by = "created_at"
    order = request.args.get("order", "desc").lower()
    if order not in {"asc", "desc"}:
        order = "desc"

    return PageParams(page=page, per_page=per_page, sort_by=sort_by, order=order)


def apply_pagination(stmt: Select, params: PageParams) -> tuple[Select, int]:
    """给查询加分页，返回 (分页后的查询, 总数)。

    计数和取数据分两步——这是必须的，因为 LIMIT 会影响 COUNT 结果。
    """
    total = current_app.session.execute(
        select(func.count()).select_from(stmt.order_by(None).subquery())
    ).scalar_one()
    offset = (params.page - 1) * params.per_page
    sort_col = getattr(Task, params.sort_by)
    stmt = stmt.order_by(desc(sort_col) if params.order == "desc" else asc(sort_col))
    stmt = stmt.limit(params.per_page).offset(offset)
    return stmt, total

实现 API 端点

7.1 认证 API

"""taskboard/api/auth.py"""
from __future__ import annotations

from flask import Blueprint, current_app, g, jsonify, request
from flask_jwt_extended import (
    create_access_token, create_refresh_token, get_jwt_identity,
    jwt_required,
)
from sqlalchemy import or_, select

from ..extensions import db
from ..models import User
from ..utils.responses import created, error
from ..utils.security import (
    InvalidTokenError, decode_token, hash_password, password_strength,
    verify_password,
)

bp = Blueprint("auth_api", __name__)

USERNAME_RE = __import__("re").compile(r"^[a-zA-Z0-9_]{3,32}$")
EMAIL_RE = __import__("re").compile(r"^[^@\s]+@[^@\s]+\.[a-zA-Z]{2,}$")


def _validate_registration(data: dict) -> tuple[dict, dict]:
    """校验注册表单，返回 (清洗后的数据, 字段错误字典)。"""
    errors: dict[str, list[str]] = {}
    username = (data.get("username") or "").strip()
    email = (data.get("email") or "").strip().lower()
    password = data.get("password") or ""

    if not USERNAME_RE.match(username):
        errors["username"] = ["只能包含字母数字下划线，长度 3-32"]
    if not EMAIL_RE.match(email):
        errors["email"] = ["不是合法的邮箱地址"]
    ok_pwd, pwd_issues = password_strength(password)
    if not ok_pwd:
        errors["password"] = pwd_issues

    return {"username": username, "email": email, "password": password}, errors


@bp.post("/register")
def register():
    """注册新用户。

    201 Created
    Response: {"data": {"user": {...}, "access_token": "...", "refresh_token": "..."}}
    Error: 409 用户名或邮箱已存在
    """
    data = request.get_json(silent=True) or {}
    clean, errors = _validate_registration(data)
    if errors:
        return error(422, "注册信息校验失败", field_errors=errors)

    # 唯一性检查
    existing = db.session.execute(
        select(User).where(or_(User.username == clean["username"],
                               User.email == clean["email"]))
    ).scalar_one_or_none()
    if existing:
        field = "username" if existing.username == clean["username"] else "email"
        return error(409, f"{field} 已被使用",
                     field_errors={field: ["该值已被占用"]})

    user = User(username=clean["username"], email=clean["email"],
                password_hash=hash_password(clean["password"]))
    db.session.add(user)
    db.session.commit()      # 注意：必须 commit 才能拿到 user.id

    payload = {
        "user": user.to_dict(),
        "access_token": create_access_token(user.id),
        "refresh_token": create_refresh_token(user.id),
    }
    return created(payload, location=f"/api/v1/auth/users/{user.id}")


@bp.post("/login")
def login():
    """用户名 + 密码登录。

    200 OK / 401 凭证错误
    """
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    password = data.get("password") or ""

    user = db.session.execute(
        select(User).where(or_(User.username == username, User.email == username))
    ).scalar_one_or_none()

    # 统一错误信息：不告诉调用方"用户不存在"还是"密码错误"，防止账号枚举
    if user is None or not verify_password(user.password_hash, password):
        return error(401, "用户名或密码错误")
    if not user.is_active:
        return error(403, "账号已被封禁，请联系管理员")

    from ..models import utcnow
    user.last_login_at = utcnow()
    db.session.commit()

    return jsonify({
        "user": user.to_dict(),
        "access_token": create_access_token(user.id),
        "refresh_token": create_refresh_token(user.id),
        "token_type": "Bearer",
        "expires_in": 1800,       # 秒，前端据此提前刷新
    })


@bp.post("/refresh")
def refresh():
    """用刷新令牌换新的访问令牌。刷新令牌有效期长但只用于换发。"""
    data = request.get_json(silent=True) or {}
    token = data.get("refresh_token", "")
    try:
        payload = decode_token(token, expected_type="refresh")
    except InvalidTokenError as exc:
        return error(401, str(exc))
    user = db.session.get(User, int(payload["sub"]))
    if user is None or not user.is_active:
        return error(401, "用户不可用")
    return jsonify({
        "access_token": create_access_token(user.id),
        "token_type": "Bearer",
        "expires_in": 1800,
    })


@bp.get("/me")
def me():
    """获取当前登录用户信息。需要 Authorization: Bearer 。"""
    auth = request.headers.get("Authorization", "")
    if not auth.startswith("Bearer "):
        return error(401, "缺少令牌")
    try:
        payload = decode_token(auth[7:], expected_type="access")
    except InvalidTokenError as exc:
        return error(401, str(exc))
    user = db.session.get(User, int(payload["sub"]))
    if user is None:
        return error(401, "用户不存在")
    g.current_user = user
    return jsonify({"data": user.to_dict()})

7.2 任务 API（CRUD + 业务动作）

"""taskboard/api/tasks.py"""
from __future__ import annotations

from flask import Blueprint, current_app, g, request
from sqlalchemy import select
from sqlalchemy.orm import selectinload

from ..extensions import db
from ..models import Notification, Project, Tag, Task, TaskStatus, User, utcnow
from ..services import tasks as task_service
from ..utils.pagination import apply_pagination, get_page_params
from ..utils.responses import created, error, no_content, ok

bp = Blueprint("tasks_api", __name__)


@bp.get("")
def list_tasks():
    """列出任务。

    GET /api/v1/tasks?status=todo&priority=high&project_id=1&page=1&per_page=20&sort=due_date&order=asc

    200 OK -> {"data": [...], "meta": {"page":1,"total":42,...}}
    """
    params = get_page_params()

    stmt = select(Task).options(
        selectinload(Task.owner),
        selectinload(Task.tags),
        selectinload(Task.project),
    ).where(Task.owner_id == g.current_user.id)   # 端点已 api_auth 装饰

    # 过滤：每个条件独立判断，缺省不过滤
    if (status := request.args.get("status")):
        try:
            stmt = stmt.where(Task.status == TaskStatus(status))
        except ValueError:
            return error(400, f"非法状态值: {status}",
                         detail={"allowed": [s.value for s in TaskStatus]})
    if (priority := request.args.get("priority")):
        stmt = stmt.where(Task.priority == priority)
    if (project_id := request.args.get("project_id", type=int)):
        stmt = stmt.where(Task.project_id == project_id)
    if q := request.args.get("q"):
        # 模糊搜索：% 转义，避免用户输入 % 或 _ 变成通配符全表扫
        stmt = stmt.where(Task.title.ilike(f"%{q.replace('%', r'\\%').replace('_', r'\\_')}%",
                                           escape="\\"))
    if request.args.get("overdue") == "true":
        from datetime import date
        stmt = stmt.where(Task.due_date < date.today(),
                          Task.status.notin_([TaskStatus.DONE, TaskStatus.CANCELLED]))

    stmt, total = apply_pagination(stmt, params)
    tasks = db.session.execute(stmt).scalars().unique().all()
    return ok([t.to_dict() for t in tasks],
              meta={"page": params.page, "per_page": params.per_page,
                    "total": total, "has_next": params.page * params.per_page < total})


@bp.post("")
def create_task():
    """创建任务。

    201 Created，Location: /api/v1/tasks/{id}
    Body: {"title": "...", "description": "...", "priority": "high", "project_id": 1, "tags": ["bug","ui"]}
    """
    data = request.get_json(silent=True) or {}

    # 校验
    title = (data.get("title") or "").strip()
    if not title:
        return error(422, "标题不能为空", field_errors={"title": ["必填"]})
    if len(title) > 200:
        return error(422, "标题过长", field_errors={"title": ["最多 200 字符"]})

    task = Task(
        title=title,
        description=(data.get("description") or "").strip() or None,
        priority=data.get("priority", "medium"),
        due_date=task_service.parse_date(data.get("due_date")),
        project_id=data.get("project_id"),
        owner_id=g.current_user.id,
    )
    db.session.add(task)
    db.session.flush()   # flush 拿到 task.id，但还没提交

    # 标签：已存在则复用，不存在则创建
    tag_names = data.get("tags") or []
    if tag_names:
        task.tags = task_service.get_or_create_tags(db.session, tag_names)
    db.session.commit()

    return created(task.to_dict(detail=True),
                   location=f"/api/v1/tasks/{task.id}")


def _load_task_or_error(task_id: int) -> tuple[Task | None, tuple | None]:
    """取任务并检查归属。返回 (task, error_response)，二者必有其一为 None。"""
    task = db.session.get(Task, task_id)
    if task is None:
        return None, error(404, "任务不存在")
    if task.owner_id != g.current_user.id and not g.current_user.is_admin():
        return None, error(403, "无权访问该任务")
    return task, None


@bp.get("/<int:task_id>")
def get_task(task_id: int):
    """获取单个任务详情。

    200 OK
    """
    task, err = _load_task_or_error(task_id)
    if err:
        return err
    return ok(task.to_dict(detail=True))


@bp.patch("/<int:task_id>")
def update_task(task_id: int):
    """部分更新任务。只传要改的字段。"""
    task, err = _load_task_or_error(task_id)
    if err:
        return err

    data = request.get_json(silent=True) or {}
    if not data:
        return error(400, "请求体为空，没有要更新的字段")

    # 白名单：只允许修改这几个字段，防止越权改 owner_id
    allowed = {"title", "description", "status", "priority",
               "progress", "due_date", "project_id", "position"}
    unknown = set(data) - allowed
    if unknown:
        return error(400, f"包含不可修改的字段: {sorted(unknown)}")

    if "title" in data:
        new_title = (data["title"] or "").strip()
        if not new_title:
            return error(422, "标题不能为空")
        task.title = new_title

    if "status" in data:
        try:
            new_status = TaskStatus(data["status"])
        except ValueError:
            return error(400, f"非法状态: {data['status']}")
        if not task.can_transition_to(new_status):
            return error(409, f"不允许从 {task.status.value} 转为 {new_status.value}",
                         detail={"allowed_from": task.status.value})
        task.status = new_status
        if new_status == TaskStatus.DONE:
            task.completed_at = utcnow()
            task.progress = 100

    if "progress" in data:
        try:
            task.progress = max(0, min(100, int(data["progress"])))
        except (TypeError, ValueError):
            return error(422, "progress 必须是 0-100 的整数")

    if "description" in data:
        task.description = (data["description"] or "").strip() or None

    if "due_date" in data:
        task.due_date = task_service.parse_date(data["due_date"])

    db.session.commit()
    return ok(task.to_dict(detail=True))


@bp.delete("/<int:task_id>")
def delete_task(task_id: int):
    """删除任务。

    204 No Content
    """
    task, err = _load_task_or_error(task_id)
    if err:
        return err
    db.session.delete(task)
    db.session.commit()
    return no_content()


@bp.post("/<int:task_id>/complete")
def complete_task(task_id: int):
    """业务动作：标记完成。

    为什么单独开一个端点而不是 PATCH？
    因为"完成"不是简单改字段——它触发状态机校验、更新 completed_at、
    可能发通知、进统计。这些副作用有明确的业务名字，适合作为独立动作。
    """
    task, err = _load_task_or_error(task_id)
    if err:
        return err
    try:
        task.mark_done()
    except ValueError as exc:
        return error(409, str(exc))
    db.session.commit()
    return ok(task.to_dict(detail=True))


@bp.get("/stats")
def task_stats():
    """按状态和优先级统计。看板顶部的数字卡片。"""
    from sqlalchemy import func
    rows = db.session.execute(
        select(Task.status, func.count())
        .where(Task.owner_id == g.current_user.id)
        .group_by(Task.status)
    ).all()
    by_status = {s.value if hasattr(s, "value") else s: c for s, c in rows}
    return ok({
        "by_status": {st.value: by_status.get(st.value, 0) for st in TaskStatus},
        "total": sum(by_status.values()),
    })

7.3 业务逻辑层：路由不该写逻辑

"""taskboard/services/tasks.py —— 业务逻辑，可被 API、CLI、定时任务复用。"""
from __future__ import annotations

from datetime import date, datetime, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session

from ..models import Notification, Tag


def parse_date(raw: str | None) -> date | None:
    """把字符串安全地解析成 date。非法输入返回 None，不抛异常。"""
    if not raw:
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(raw.strip(), fmt).date()
        except ValueError:
            continue
    raise ValueError(f"无法解析日期: {raw!r}，应为 YYYY-MM-DD")


def get_or_create_tags(session: Session, names: list[str]) -> list[Tag]:
    """按名字批量获取标签，不存在则创建。

    不存在重复：先查已有 → 缺失的批量插入 → 再查一次拿全部。
    这样只发 2 条 SQL，而不是 N 条。
    """
    cleaned = []
    for n in names:
        n = n.strip()
        if n and n not in cleaned:
            cleaned.append(n)
    if not cleaned:
        return []

    existing = session.execute(
        select(Tag).where(Tag.name.in_(cleaned))
    ).scalars().all()
    have = {t.name: t for t in existing}

    missing = [n for n in cleaned if n not in have]
    if missing:
        # 这里可能有并发竞态（两个请求同时创建同名标签），
        # 需要数据库唯一约束兜底 + IntegrityError 重试。生产要处理。
        session.add_all([Tag(name=n) for n in missing])
        session.flush()
        existing = session.execute(
            select(Tag).where(Tag.name.in_(cleaned))
        ).scalars().all()

    return list(existing)


def notify(session: Session, user_id: int, message: str,
           task_id: int | None = None) -> Notification:
    """创建通知。"""
    n = Notification(user_id=user_id, message=message, task_id=task_id)
    session.add(n)
    return n


def send_overdue_reminders(session: Session) -> int:
    """扫描过期任务，给属主发提醒。返回发送的通知数。

    这是典型的"每日定时任务"——用 Flask CLI 命令 + 系统 cron，
    或 Celery Beat 触发。
    """
    from ..models import Task, TaskStatus
    today = date.today()
    overdue = session.execute(
        select(Task).where(Task.due_date < today,
                           Task.status.in_([TaskStatus.TODO, TaskStatus.DOING]))
    ).scalars().all()
    for task in overdue:
        notify(session, task.owner_id,
               f"任务「{task.title}」已逾期（{task.due_date}）", task.id)
    session.commit()
    return len(overdue)

前端：模板渲染与 SPA 交互

8.1 服务端渲染页面

"""taskboard/web/views.py"""
from __future__ import annotations

from flask import Blueprint, abort, flash, redirect, render_template, request, url_for
from flask_login import current_user, login_required, login_user, logout_user

from ..extensions import db, login_manager
from ..models import Project, Task, TaskStatus, User
from ..utils.pagination import get_page_params

bp = Blueprint("web", __name__)


@bp.route("/")
def index():
    """首页：登录用户看自己的看板，未登录看公开项目。"""
    if current_user.is_authenticated:
        params = get_page_params()
        tasks = (db.session.query(Task)
                 .filter_by(owner_id=current_user.id)
                 .order_by(Task.created_at.desc())
                 .limit(params.per_page).offset((params.page - 1) * params.per_page)
                 .all())
        projects = db.session.query(Project).filter_by(
            owner_id=current_user.id).all()
        return render_template("dashboard.html", tasks=tasks, projects=projects)

    public = db.session.query(Project).filter_by(is_public=True).limit(10).all()
    return render_template("landing.html", projects=public)


@bp.route("/login", methods=["GET", "POST"])
def login():
    if current_user.is_authenticated:
        return redirect(url_for("web.index"))
    error_msg = None
    if request.method == "POST":
        username = request.form.get("username", "").strip()
        password = request.form.get("password", "")
        user = User.query.filter_by(username=username).first()
        if user and user.check_password(password):
            login_user(user, remember=request.form.get("remember") == "on")
            next_url = request.args.get("next")
            # 安全：只允许站内相对跳转，防止开放重定向
            if not next_url or not next_url.startswith("/") or next_url.startswith("//"):
                next_url = url_for("web.index")
            return redirect(next_url)
        error_msg = "用户名或密码错误"
    return render_template("login.html", error=error_msg)


@bp.route("/logout")
@login_required
def logout():
    logout_user()
    flash("已退出登录", "info")
    return redirect(url_for("web.index"))


@bp.route("/tasks/<int:task_id>", methods=["GET", "POST"])
@login_required
def task_detail(task_id: int):
    task = db.session.get(Task, task_id)
    if task is None or (task.owner_id != current_user.id
                        and not current_user.is_admin()):
        abort(404)
    if request.method == "POST":
        action = request.form.get("action")
        if action == "toggle":
            task.status = (TaskStatus.DONE if task.status != TaskStatus.DONE
                           else TaskStatus.TODO)
            task.progress = 100 if task.status == TaskStatus.DONE else 0
            db.session.commit()
            flash("状态已更新", "success")
        elif action == "comment":
            body = request.form.get("body", "").strip()
            if body:
                from ..models import Comment
                db.session.add(Comment(body=body, task_id=task.id,
                                       author_id=current_user.id))
                db.session.commit()
        return redirect(url_for("web.task_detail", task_id=task.id))
    return render_template("task_detail.html", task=task)


@bp.route("/profile")
@login_required
def profile():
    stats = {
        "total": Task.query.filter_by(owner_id=current_user.id).count(),
        "done": Task.query.filter_by(owner_id=current_user.id,
                                     status=TaskStatus.DONE).count(),
    }
    return render_template("profile.html", user=current_user, stats=stats)

<!-- templates/dashboard.html -->
<!DOCTYPE html>
<html lang="zh-CN">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>任务看板</title>
  <link rel="stylesheet" href="{{ url_for('static', filename='css/app.css') }}">
</head>
<body>
  <header class="topbar">
    <span class="brand">TaskBoard</span>
    <nav>
      <a href="{{ url_for('web.index') }}">看板</a>
      <a href="{{ url_for('web.profile') }}">我的</a>
      <a href="{{ url_for('web.logout') }}">退出</a>
    </nav>
  </header>

  <main class="board">
    <!-- 按状态分列看板 -->
    {% for st in statuses %}
    <section class="column" data-status="{{ st.value }}">
      <h2>{{ st.label }} <span class="count">{{ counts.get(st.value, 0) }}</span></h2>
      <ul class="task-list">
        {% for task in grouped.get(st.value, []) %}
        <li class="task-card priority-{{ task.priority.value }}"
            draggable="true" data-task-id="{{ task.id }}">
          <a href="{{ url_for('web.task_detail', task_id=task.id) }}">
            <h3>{{ task.title }}</h3>
          </a>
          <div class="meta">
            <span class="tag-project">{{ task.project.name if task.project }}</span>
            {% if task.is_overdue %}<span class="overdue">已逾期</span>{% endif %}
            {% for t in task.tags %}
            <span class="chip" style="background: {{ t.color }}">{{ t.name }}</span>
            {% endfor %}
          </div>
        </li>
        {% endfor %}
      </ul>
    </section>
    {% endfor %}
  </main>

  <script src="{{ url_for('static', filename='js/board.js') }}"></script>
</body>
</html>

8.2 前端调用 API：fetch + JWT 拦截

<!-- templates/dashboard.html 底部内联脚本 -->
<script>
// 统一的 API 客户端：自动带上令牌、自动处理 401 刷新、自动解析错误
const API = {
  base: '/api/v1',
  tokens: { access: null, refresh: null },

  setTokens(t) {
    this.tokens.access = t.access_token;
    this.tokens.refresh = t.refresh_token;
    // 生产环境不建议把 refresh token 放 localStorage（易被 XSS 窃取），
    // 更安全是放 HttpOnly Cookie 由服务端下发。
  },

  async request(path, options = {}) {
    const headers = { 'Content-Type': 'application/json', ...(options.headers || {}) };
    if (this.tokens.access) headers['Authorization'] = `Bearer ${this.tokens.access}`;

    let resp = await fetch(this.base + path, { ...options, headers });

    // 401 尝试自动刷新一次
    if (resp.status == 401 && this.tokens.refresh) {
      const refreshed = await this.refresh();
      if (refreshed) {
        headers['Authorization'] = `Bearer ${this.tokens.access}`;
        resp = await fetch(this.base + path, { ...options, headers });
      } else {
        window.location.href = '/login';
        return null;
      }
    }

    if (resp.status === 204) return null;   // 204 无 body，不能调 json()

    const data = await resp.json();
    if (!resp.ok) {
      // 统一的错误处理：抛出带字段信息的错误
      const err = new Error(data.error?.message || `HTTP ${resp.status}`);
      err.status = resp.status;
      err.fieldErrors = data.error?.field_errors;
      throw err;
    }
    return data;
  },

  async refresh() {
    try {
      const resp = await fetch(this.base + '/auth/refresh', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ refresh_token: this.tokens.refresh }),
      });
      if (!resp.ok) return false;
      const data = await resp.json();
      this.tokens.access = data.access_token;
      return true;
    } catch {
      return false;
    }
  },

  get(path) { return this.request(path); },
  post(path, body) { return this.request(path, { method: 'POST', body: JSON.stringify(body) }); },
  patch(path, body) { return this.request(path, { method: 'PATCH', body: JSON.stringify(body) }); },
  del(path) { return this.request(path, { method: 'DELETE' }); },
};

// 拖拽更新状态：看板列即状态
document.querySelectorAll('.column').forEach(col => {
  col.addEventListener('dragover', e => { e.preventDefault(); col.classList.add('drag-over'); });
  col.addEventListener('dragleave', () => col.classList.remove('drag-over'));
  col.addEventListener('drop', async e => {
    e.preventDefault();
    col.classList.remove('drag-over');
    const card = e.dataTransfer.getData('text/plain');
    const newStatus = col.dataset.status;
    try {
      await API.patch(`/tasks/${card}`, { status: newStatus });
      card.classList.add('just-updated');
      // 乐观更新：先动 DOM，失败再回滚
    } catch (err) {
      alert('更新失败：' + err.message);
      location.reload();
    }
  });
});

// 新建任务
document.getElementById('new-task-form')?.addEventListener('submit', async e => {
  e.preventDefault();
  const form = e.target;
  const title = form.title.value.trim();
  if (!title) return;
  try {
    await API.post('/tasks', { title });
    location.reload();
  } catch (err) {
    document.getElementById('form-error').textContent = err.message;
  }
});
</script>

测试与部署

9.1 API 测试

"""tests/conftest.py"""
from __future__ import annotations

import pytest

from taskboard import create_app
from taskboard.extensions import db
from taskboard.models import Task, TaskStatus, User, Project
from taskboard.utils.security import hash_password


@pytest.fixture
def app():
    app = create_app("testing")
    with app.app_context():
        # 在内存 SQLite 里建表
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()


@pytest.fixture
def client(app):
    return app.test_client()


@pytest.fixture
def user(app):
    u = User(username="alice", email="alice@example.com",
              password_hash=hash_password("alice12345"))
    db.session.add(u)
    db.session.commit()
    return u


@pytest.fixture
def other_user(app):
    u = User(username="bob", email="bob@example.com",
              password_hash=hash_password("bob12345"))
    db.session.add(u)
    db.session.commit()
    return u


@pytest.fixture
def project(app, user):
    p = Project(name="测试项目", owner_id=user.id, is_public=True)
    db.session.add(p)
    db.session.commit()
    return p


@pytest.fixture
def auth_headers(client, user):
    """登录并返回带 Bearer 令牌的请求头。"""
    resp = client.post("/api/v1/auth/login",
                       json={"username": "alice", "password": "alice12345"})
    assert resp.status_code == 200
    token = resp.get_json()["access_token"]
    return {"Authorization": f"Bearer {token}"}


@pytest.fixture
def sample_task(app, user, project):
    t = Task(title="写测试", description="覆盖 API", status=TaskStatus.TODO,
             owner_id=user.id, project_id=project.id)
    db.session.add(t)
    db.session.commit()
    return t

"""tests/test_tasks_api.py"""
from taskboard.models import Task, TaskStatus


class TestAuth:
    def test_register_creates_user(self, client):
        resp = client.post("/api/v1/auth/register", json={
            "username": "charlie", "email": "c@example.com", "password": "charlie123"})
        assert resp.status_code == 201
        assert resp.get_json()["data"]["user"]["username"] == "charlie"

    def test_register_weak_password_rejected(self, client):
        resp = client.post("/api/v1/auth/register", json={
            "username": "charlie", "email": "c@example.com", "password": "123"})
        assert resp.status_code == 422
        assert "password" in resp.get_json()["error"]["field_errors"]

    def test_duplicate_username_returns_409(self, client, user):
        resp = client.post("/api/v1/auth/register", json={
            "username": "alice", "email": "new@example.com", "password": "alice12345"})
        assert resp.status_code == 409

    def test_login_wrong_password_401(self, client, user):
        resp = client.post("/api/v1/auth/login",
                           json={"username": "alice", "password": "wrong"})
        assert resp.status_code == 401
        # 错误信息不能泄露"用户存在"
        assert "密码" in resp.get_json()["error"]["message"]

    def test_me_requires_token(self, client):
        assert client.get("/api/v1/auth/me").status_code == 401


class TestTasks:
    def test_list_requires_auth(self, client):
        assert client.get("/api/v1/tasks").status_code == 401

    def test_create_task(self, client, auth_headers, project):
        resp = client.post("/api/v1/tasks", headers=auth_headers,
                           json={"title": "写文档", "priority": "high",
                                 "project_id": project.id})
        assert resp.status_code == 201
        assert resp.get_json()["data"]["title"] == "写文档"
        # Location 头应指向新资源
        assert f"/api/v1/tasks/{resp.get_json()['data']['id']}" in resp.headers["Location"]

    def test_create_task_empty_title_422(self, client, auth_headers):
        resp = client.post("/api/v1/tasks", headers=auth_headers, json={"title": "  "})
        assert resp.status_code == 422

    def test_list_pagination(self, client, auth_headers, user, project):
        from taskboard.extensions import db
        for i in range(25):
            db.session.add(Task(title=f"任务{i}", owner_id=user.id,
                                project_id=project.id))
        db.session.commit()
        resp = client.get("/api/v1/tasks?page=2&per_page=10", headers=auth_headers)
        data = resp.get_json()
        assert resp.status_code == 200
        assert len(data["data"]) == 10
        assert data["meta"]["total"] == 25
        assert data["meta"]["has_next"] is True
        assert data["meta"]["has_prev"] is True

    def test_list_filters_by_status(self, client, auth_headers, sample_task, app, user):
        from taskboard.extensions import db
        db.session.add(Task(title="另一个", owner_id=user.id,
                            status=TaskStatus.DONE))
        db.session.commit()
        resp = client.get("/api/v1/tasks?status=todo", headers=auth_headers)
        titles = [t["title"] for t in resp.get_json()["data"]]
        assert "写测试" in titles and "另一个" not in titles

    def test_cannot_view_others_task_403(self, client, auth_headers, app, other_user):
        from taskboard.extensions import db
        from taskboard.models import Task
        t = Task(title="私有的", owner_id=other_user.id)
        db.session.add(t); db.session.commit()
        resp = client.get(f"/api/v1/tasks/{t.id}", headers=auth_headers)
        assert resp.status_code == 403

    def test_patch_cannot_change_owner(self, client, auth_headers, sample_task, other_user):
        resp = client.patch(f"/api/v1/tasks/{sample_task.id}",
                            headers=auth_headers, json={"owner_id": other_user.id})
        assert resp.status_code == 400    # 不在白名单

    def test_complete_sets_progress_100(self, client, auth_headers, sample_task):
        resp = client.post(f"/api/v1/tasks/{sample_task.id}/complete",
                           headers=auth_headers)
        assert resp.status_code == 200
        data = resp.get_json()["data"]
        assert data["status"] == "done"
        assert data["progress"] == 100

    def test_invalid_status_transition_409(self, client, auth_headers, sample_task):
        sample_task.status = TaskStatus.DONE
        from taskboard.extensions import db
        db.session.commit()
        # done -> todo 不允许
        resp = client.patch(f"/api/v1/tasks/{sample_task.id}",
                            headers=auth_headers, json={"status": "todo"})
        assert resp.status_code == 409

    def test_delete_returns_204(self, client, auth_headers, sample_task):
        resp = client.delete(f"/api/v1/tasks/{sample_task.id}", headers=auth_headers)
        assert resp.status_code == 204
        assert resp.data == b""          # 无响应体
        assert client.get(f"/api/v1/tasks/{sample_task.id}",
                           headers=auth_headers).status_code == 404

9.2 Docker 部署

# Dockerfile
FROM python:3.12-slim AS base

ENV PYTHONUNBUFFERED=1 \\
    PYTHONDONTWRITEBYTECODE=1 \\
    PIP_NO_CACHE_DIR=1

WORKDIR /app

# 依赖单独一层，代码改动时不必重装依赖（利用 Docker 缓存）
COPY pyproject.toml README.md ./
COPY taskboard/__init__.py taskboard/__init__.py
RUN pip install --upgrade pip && pip install .

COPY . .

# 非 root 运行，降低容器逃逸后的影响
RUN useradd -m appuser && chown -R appuser /app
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=3s CMD \\
    python -c "import urllib.request;urllib.request.urlopen('http://127.0.0.1:8000/healthz')"

CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", \\
     "--threads", "2", "--timeout", "60", "--access-logfile", "-", \\
     "wsgi:app"]

# docker-compose.yml（应用 + Postgres）
version: "3.9"
services:
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: taskboard
      POSTGRES_PASSWORD: ${DB_PASSWORD:?必须设置}
      POSTGRES_DB: taskboard
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U taskboard"]
      interval: 5s
      retries: 5

  web:
    build: .
    ports: ["8000:8000"]
    environment:
      DATABASE_URL: postgresql+psycopg://taskboard:${DB_PASSWORD}@db:5432/taskboard
      SECRET_KEY: ${SECRET_KEY:?必须设置}
      FLASK_ENV: production
    depends_on:
      db:
        condition: service_healthy
    # 启动前自动跑迁移
    command: >
      sh -c "flask db upgrade &&
             gunicorn --bind 0.0.0.0:8000 --workers 4 wsgi:app"

volumes:
  pgdata:

9.3 生产部署检查清单

"""taskboard/web/views.py 里追加健康检查与限流。"""
from flask import jsonify, request
import time
from collections import defaultdict

# 简易滑动窗口限流（生产建议用 Redis）
_hits: dict[str, list[float]] = defaultdict(list)


def rate_limit(max_per_minute: int = 60):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            ip = request.remote_addr or "unknown"
            now = time.time()
            # 清理 60 秒前的记录
            _hits[ip] = [t for t in _hits[ip] if now - t < 60]
            if len(_hits[ip]) >= max_per_minute:
                return jsonify({"error": {"code": 429, "message": "请求过于频繁"}}), 429
            _hits[ip].append(now)
            return fn(*args, **kwargs)
        return wrapper
    return decorator


@bp.get("/healthz")
def healthz():
    """容器健康检查：只查进程活着，不查数据库（DB 慢时会误杀容器）。"""
    return jsonify({"status": "ok", "time": time.time()}), 200


@bp.get("/readyz")
def readyz():
    """就绪检查：查数据库连通性。"""
    from ..extensions import db
    from sqlalchemy import text
    try:
        db.session.execute(text("SELECT 1"))
        return jsonify({"status": "ready"}), 200
    except Exception:
        return jsonify({"status": "not-ready"}), 503

进阶设计：性能、安全与可维护性

10.1 性能优化清单

"""批量操作：避免逐条 INSERT 触发 N 次往返。"""
from sqlalchemy import insert


def bulk_create_tasks(session, rows: list[dict]) -> list[int]:
    """批量插入，返回新 id 列表。

    1000 条逐条 insert 约 1.2s；bulk 约 30ms。
    注意：PostgreSQL 要用 RETURNING 拿 id；MySQL/SQLite 需另查。
    """
    if not rows:
        return []
    result = session.execute(insert(Task), rows)
    return list(result.inserted_primary_key)


def export_tasks_csv(session, owner_id: int) -> str:
    """大数据量导出用流式迭代，不要一次加载全部到内存。"""
    import csv, io
    from sqlalchemy import select
    buf = io.StringIO()
    w = csv.writer(buf)
    w.writerow(["id", "title", "status", "priority", "created_at"])
    # yield_per 只在 server-side cursor 下生效，适合大结果集
    result = session.execute(
        select(Task).where(Task.owner_id == owner_id).execution_options(yield_per=1000)
    )
    for t in result.scalars():
        w.writerow([t.id, t.title, t.status.value, t.priority.value,
                    t.created_at])
    return buf.getvalue()

10.2 安全清单

• SQL 注入：一律用 ORM 的参数化查询；动态排序走白名单；ilike 里转义 % 与 _；
• XSS：Jinja2 默认转义，慎用 | safe；API 响应设 Content-Type: application/json；
• CSRF：页面表单用 Flask-WTF 的 CSRF token；纯 Bearer 令牌的 API 不受 CSRF 影响（因为浏览器不会自动带 Authorization 头）；
• 越权（IDOR）：每个端点都要校验「当前用户是否有权操作这个资源」，不能只在列表查询里过滤；
• 开放重定向：登录跳转的 next 参数必须校验是站内相对路径；
• 密码与令牌：不写日志、不进 URL、不放 localStorage、哈希用 scrypt/bcrypt；
• 限流：登录、注册、密码重置、发消息接口必须限流；
• 上传：校验类型、大小、文件名，禁止用用户提供的文件名拼路径，防目录穿越。

"""输入校验的三个层次。"""
# 层次 1：类型与形状校验（pydantic 风格的显式校验）
def validate_task_payload(data: dict) -> tuple[dict, dict]:
    errors, clean = {}, {}
    if "title" not in data or not str(data["title"]).strip():
        errors["title"] = "必填且不能为空白"
    elif len(str(data["title"])) > 200:
        errors["title"] = "最多 200 字符"
    else:
        clean["title"] = str(data["title"]).strip()

    if "priority" in data:
        try:
            clean["priority"] = Priority(data["priority"])
        except ValueError:
            errors["priority"] = f"必须是 {[p.value for p in Priority]} 之一"
    return clean, errors


# 层次 2：业务规则校验（跨字段、跨状态）
def validate_due_date(task: Task, new_due) -> str | None:
    if new_due and task.completed_at and new_due < task.completed_at.date():
        return "已完成的任务的截止日期不能早于完成日期"


# 层次 3：数据库约束（最后防线，防并发绕过应用层）
#   - 唯一索引：username UNIQUE
#   - CHECK 约束：progress BETWEEN 0 AND 100
#   - 外键约束：project_id REFERENCES projects(id)
# 应用层校验可被绕过（并发、脚本），数据库约束是最后一道闸门。

10.3 测试金字塔与接口契约

接口契约（OpenAPI）保证前后端并行开发不打架：

"""用代码描述 OpenAPI 契约，自动生成文档与前端类型。"""
from flask_smorest import Api, Blueprint


def create_api(app):
    api = Api(app, spec_kwargs={"title": "TaskBoard API", "version": "1.0.0",
                               "openapi_version": "3.0.3"})

    @api.spec
    def spec():
        return {
            "tags": [{"name": "auth", "description": "认证"},
                     {"name": "tasks", "description": "任务管理"}],
            "components": {
                "securitySchemes": {
                    "bearerAuth": {"type": "http", "scheme": "bearer",
                                   "bearerFormat": "JWT"}
                }
            },
        }

    @api.route("/tasks")
    class TaskList(Resource):
        @api.doc(tags=["tasks"], security="bearerAuth")
        @api.marshal_with(task_list_schema, code=200)
        @api.expect(task_create_schema)
        def get(self):
            """分页列出任务"""
            ...

        @api.doc(tags=["tasks"], security="bearerAuth")
        @api.expect(task_create_schema)
        @api.response(201, "创建成功", task_schema)
        @api.response(422, "校验失败")
        def post(self):
            """创建任务"""
            ...

    return api

生成 /api-docs/openapi.json 后，前端可以用 openapi-typescript 自动生成 TypeScript 类型，或用 Swagger UI 直接联调，避免"接口文档和实现对不上"。

把整件事串起来：一次完整开发流程

1. 定契约：写接口表格（第 1.3 节），先定 URL、方法、请求体、状态码、响应结构；
2. 定模型：画 ER 图，写 models/；
3. 建工厂：create_app() + 扩展 + 蓝图占位，先让服务能启动；
4. 迁移：flask db init && flask db migrate && flask db upgrade；
5. 先写测试：给每个端点写 3 类测试——成功、校验失败、权限不足；
6. 实现端点：路由薄、业务在 services/、响应统一走 responses.py；
7. 接前端：模板渲染页面，或 fetch 调 API；
8. 安全加固：限流、CORS 白名单、响应头、输入校验、审计日志；
9. 部署：Docker 构建 + Compose 编排 + Nginx 反代 + 健康检查；
10. 运维：结构化日志、错误上报、监控告警、备份与迁移演练。

# 一条命令跑起整个开发环境
$ docker compose up --build
# 打开 http://localhost:8000，注册 alice / alice12345 试试

# 本地开发的热重载
$ FLASK_ENV=development FLASK_DEBUG=1 flask run --port 8000

本章小结

• 先定契约再写代码：功能清单、ER 图、接口表格三样东西提前定稿，能避免后续大规模返工。
• 应用工厂（create_app）优于全局 app：便于多实例测试、多环境配置、按需注册蓝图；扩展对象在模块级声明、在工厂里 init_app 绑定。
• 蓝图（blueprint） 把 API 与页面路由切成独立模块，用 url_prefix 区分职责。
• ORM 模型设计要点：外键级联删除、枚举用 native_enum=False 存字符串、统一 UTC 时间、复合索引匹配常用查询、lazy="selectin" 规避 N+1、每个模型一个 to_dict() 作为序列化出口。
• Alembic 迁移：flask db migrate/upgrade 是线上改表的唯一正路；自动生成的脚本一定要人工审阅；加列要有默认值或可空。
• 鉴权双轨：页面用 Session Cookie，API 用 JWT；密码用 scrypt 慢哈希，令牌用 HS256 签名且载荷不放敏感数据；错误处理器里统一 db.session.rollback()。
• RESTful 规范：名词复数 URL、方法表达语义、错误用状态码而非 body；401 未登录、403 无权限、409 状态冲突、422 校验失败；统一响应封装 {"data":..., "meta":...} 与 {"error":{...}}。
• 分页排序过滤必须走白名单并对 per_page 强制上限；计数与取数分两步；模糊搜索要转义通配符。
• 逻辑与路由分离：业务放 services/，可被 API、CLI、定时任务复用；模型更新用字段白名单防越权改 owner_id。
• 前端两条路线：服务端渲染（Jinja2 自动转义防 XSS）适合内容型；API + fetch 适合交互型，注意乐观更新与令牌安全。
• 部署要点：生产用 PostgreSQL + Gunicorn + Nginx；Docker 多阶段构建、非 root 运行、健康检查（/healthz 只探活、/readyz 查 DB）；迁移纳入部署流程。

练习题

1. 把全局 app 改造成应用工厂并写测试。用 create_app(config_name) 重构现有项目，支持 development/testing/production 三套配置。写 conftest.py 保证每个测试用例拿到独立的 app 与干净的内存数据库，验证两个测试之间数据互不污染。
   参考答案要点app = create_app("testing") 在 fixture 里；with app.app_context(): db.create_all(); yield app; db.session.remove(); db.drop_all()。注意 SQLALCHEMY_DATABASE_URI = "sqlite://" 是内存库，每次连接独立，需要在 app context 内 create_all。可用 app.config["SQLALCHEMY_DATABASE_URI"] 断言配置正确切换。
2. 设计并迁移一个完整的数据模型。为项目加「里程碑（Milestone）」实体：任务归属里程碑、里程碑有名称/截止日期/状态。用 flask db migrate 生成迁移脚本，人工检查并补充 ondelete 与索引，再写 3 个测试验证级联行为。
   参考答案要点新表关系 milestones = relationship(back_populates="tasks")，Task 侧加 milestone_id 外键 ondelete="SET NULL"（删里程碑不该删任务，用 CASCADE 就错了）。迁移脚本里确认 nullable=True、给 due_date 建索引。级联测试：删里程碑后断言任务仍在但 milestone_id is None。
3. 实现完整的 JWT 鉴权链路。写注册、登录、刷新、/me 四个端点。访问令牌 30 分钟、刷新令牌 14 天；刷新接口只接受 refresh 类型的令牌（拿 access 令牌去刷新应返回 401）。写 6 个测试覆盖：正常流程、错误密码、令牌过期、令牌类型错、缺失 Authorization 头、封禁用户。
   参考答案要点载荷里放 type 字段，decode_token(token, expected_type="refresh") 校验；过期测试用 jwt.encode 手工造一个 exp 在过去的令牌，或 freezegun / monkeypatch 时间。登录失败统一返回"用户名或密码错误"，不要区分用户是否存在。
4. 实现一个带完整查询能力的任务列表接口。要求支持 status、priority、project_id、q（模糊搜索）、overdue、分页、排序六个维度。写至少 12 个测试，重点：非法 status 返回 400 且响应里带合法枚举列表、per_page=99999 被截断到配置上限、sort=nonexistent 静默回退到默认排序、搜索关键词含 % 时不被当成通配符。
   参考答案要点sort 用白名单 SORTABLE；per_page = min(MAX_ITEMS_PER_PAGE, ...)；ilike(f"%{q}%", escape="\\") 并先 q.replace("%", "\\%")。测试非法排序：resp = client.get("?sort=password_hash") 断言仍返回 200 且顺序是 created_at desc。
5. 实现字段级白名单的部分更新与状态机。PATCH /tasks/<id> 允许改 title/description/status/priority/progress/due_date，其它字段（尤其 owner_id）必须拒绝。实现 ALLOWED_TRANSITIONS 状态机，非法转换返回 409 且响应里带 allowed_from。写 10 个测试。
   参考答案要点unknown = set(data) - allowed 后返回 400；状态转换 if not task.can_transition_to(new_status): return error(409, ...)。测试矩阵：todo→done 允许、done→todo 拒绝（409）、cancelled→todo 允许、非法枚举字符串 400、progress=150 被夹到 100 或报 422（按你的设计写断言）。
6. 把路由里的逻辑抽到 service 层并加通知。把创建/更新/完成任务时的副作用搬到 services/tasks.py：完成任务时给项目其他成员发 Notification。写测试验证：完成任务后通知被创建且消息内容正确；不重复通知同一人。
   参考答案要点service 函数签名如 complete_task(session, task, actor) -> list[Notification]，只操作 session 不碰 request/flask.g，这样单元测试可以传一个真实 session 而不需要 request context。通知去重用 set() 收集 user_id 后遍历。
7. 做一次性能优化：消灭 N+1。打开 SQLAlchemy 的 echo=True，用 20 个任务、每任务 5 个评论的种子数据跑列表接口，数一下发了几条 SQL。改成 eager load 或直接查询后 SQL 条数降到常数级，前后对比并写进报告。
   参考答案要点关系设 lazy="selectin" 或查询时 selectinload(Task.comments)；to_dict() 里若访问 task.owner.username 而关系是 lazy="select"，每条记录就多一条查询。另一种优化是列表页的 to_dict() 不输出关联对象，只输出 id 与名称。用 len(session.get_debug_queries()) 计数做断言。
8. 完整 Docker 化并部署到测试环境。写 Dockerfile（多阶段、非 root、健康检查）、docker-compose.yml（应用 + Postgres）、.dockerignore。要求 docker compose up 一条命令跑起来，自动执行迁移。写一个 /readyz 探活端点，并用 docker compose ps 与 curl 验证容器健康。
   参考答案要点Dockerfile 里先 COPY pyproject.toml 再 pip install . 利用缓存层；.dockerignore 至少含 .git、__pycache__、.venv、*.db、tests；Compose 用 depends_on: condition: service_healthy 等数据库就绪；command: sh -c "flask db upgrade && gunicorn ..." 把迁移串进启动；HEALTHCHECK 用 curl 打 /healthz。
9. （选做）给 TaskBoard 加实时协作。用 Flask-SocketIO 或 SSE（Server-Sent Events）实现：任一用户修改任务后，其他在线用户的看板自动更新该卡片。要求：连接时鉴权；断线自动重连；同一项目内的房间隔离。
   参考答案要点WebSocket 方案：socketio = SocketIO(cors_allowed_origins=[...])，在 factory 里 socketio.init_app(app)（注意多个 init 会报错，需 init_app 而非 SocketIO(app)）；连接事件里校验 JWT 并 join_room(f"project_{pid}")；任何写操作后 socketio.emit("task_updated", data, room=...)。SSE 方案更简单（服务端单向推），用 Response(generator(), mimetype="text/event-stream")，但注意 Nginx 需要 proxy_buffering off。多进程部署时要用消息队列（如 Redis pub/sub）同步事件，否则各 worker 收不到彼此的事件。