第 6 章 · 数据库与 ORM

import pymysql

def get_article_detail(article_id):
    conn = pymysql.connect(
        host="127.0.0.1", user="root", password="123456",
        database="hiwen", charset="utf8mb4",
    )
    cursor = conn.cursor()
    # 危险①：字符串拼接，article_id 若来自 URL 未校验，可被注入
    sql = f"SELECT * FROM article WHERE id = {article_id}"
    cursor.execute(sql)
    row = cursor.fetchone()
    conn.close()

    if row is None:
        return None
    # 危险②：靠下标取字段，加一列就可能全错位
    return {"id": row[0], "title": row[1], "author": row[2], "views": row[8]}

from app.models import Article

def get_article_detail(article_id):
    article = Article.query.get(article_id)   # 或 db.session.get(Article, article_id)
    if article is None:
        return None
    return {
        "id": article.id,
        "title": article.title,
        "author": article.author,
        "views": article.views,
    }

from flask_sqlalchemy import SQLAlchemy

# 全局唯一的 db 对象：所有模型都从它派生
db = SQLAlchemy()

from flask import Flask
from config import config_map
from .extensions import db

def create_app(env: str = "development") -> Flask:
    app = Flask(__name__)
    app.config.from_object(config_map[env])

    # 连接串来自 config.py 的 SQLALCHEMY_DATABASE_URI
    # mysql+pymysql://用户:密码@主机:端口/库名?charset=utf8mb4
    db.init_app(app)

    return app

from datetime import datetime
from .extensions import db

class User(db.Model):
    """用户表，对应 hiwen 库的 `user`"""
    __tablename__ = "user"

    id         = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username   = db.Column(db.String(50), nullable=False, unique=True)
    password   = db.Column(db.String(255), nullable=False)   # 存哈希，绝不存明文
    nickname   = db.Column(db.String(50))
    email      = db.Column(db.String(100), unique=True)
    avatar     = db.Column(db.String(255), default="")
    intro      = db.Column(db.String(255))
    gender     = db.Column(db.SmallInteger, default=0)       # 0未知 1男 2女
    status     = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at = db.Column(db.DateTime, nullable=False,
                           default=datetime.now, onupdate=datetime.now)

class Category(db.Model):
    """文学分类表，对应 `category`（唐诗/宋词/元曲/散文/小说）"""
    __tablename__ = "category"

    id          = db.Column(db.Integer, primary_key=True)
    name        = db.Column(db.String(30), nullable=False, unique=True)
    description = db.Column(db.String(255))
    sort        = db.Column(db.Integer, nullable=False, default=0)
    status      = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at  = db.Column(db.DateTime, nullable=False, default=datetime.now)


class Article(db.Model):
    """文学作品表，对应 `article`（静夜思 / 春望 / 念奴娇·赤壁怀古 ……）"""
    __tablename__ = "article"

    id           = db.Column(db.Integer, primary_key=True)
    title        = db.Column(db.String(150), nullable=False)
    author       = db.Column(db.String(50))
    dynasty      = db.Column(db.String(20))
    category_id  = db.Column(db.Integer, db.ForeignKey("category.id"))  # 外键
    cover        = db.Column(db.String(255), default="")
    summary      = db.Column(db.String(500))
    content      = db.Column(db.Text)
    views        = db.Column(db.Integer, nullable=False, default=0)
    likes        = db.Column(db.Integer, nullable=False, default=0)
    is_recommend = db.Column(db.SmallInteger, nullable=False, default=0)
    status       = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at   = db.Column(db.DateTime, nullable=False, default=datetime.now)
    updated_at   = db.Column(db.DateTime, nullable=False,
                             default=datetime.now, onupdate=datetime.now)

class Comment(db.Model):
    """书评表，对应 `comment`；parent_id 自关联实现回复"""
    __tablename__ = "comment"

    id         = db.Column(db.Integer, primary_key=True)
    user_id    = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)
    article_id = db.Column(db.Integer, db.ForeignKey("article.id"), nullable=False)
    parent_id  = db.Column(db.Integer, db.ForeignKey("comment.id"))
    content    = db.Column(db.String(1000), nullable=False)
    likes      = db.Column(db.Integer, nullable=False, default=0)
    status     = db.Column(db.SmallInteger, nullable=False, default=1)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.now)

class Article(db.Model):
    __tablename__ = "article"
    # …… 上面 14 个 Column 不变 ……

    # 一对多：一篇文章 ↔ 多条书评
    # backref="article" 会自动给 Comment 生成 .article 属性
    comments = db.relationship("Comment", backref="article", lazy="dynamic")


class Comment(db.Model):
    __tablename__ = "comment"
    # …… 上面 8 个 Column 不变 ……

    # 自关联：一条书评 ↔ 它的父书评 / 子回复
    parent = db.relationship(
        "Comment",
        backref=db.backref("children", lazy="dynamic"),
        remote_side=[id],          # 指向「本表的主键」，表示这是一对多的自关联
    )


class Category(db.Model):
    __tablename__ = "category"
    # ……

    # 一个分类 ↔ 多篇作品
    articles = db.relationship("Article", backref="category", lazy="dynamic")


class User(db.Model):
    __tablename__ = "user"
    # ……

    comments = db.relationship("Comment", backref="user", lazy="dynamic")

from app.models import Article, Comment

# ① 按主键取一条：查 id=1 的作品（推荐用 db.session.get，见下方讲透）
a = db.session.get(Article, 1)
# 老写法 Article.query.get(1) 也能用，但新代码不推荐

# ② filter_by：只能写等值，写法最简洁
tang = Article.query.filter_by(dynasty="唐", status=1).all()

# ③ filter：能写 > < >=、like、in_ 等完整条件
hot = Article.query.filter(Article.views > 300).order_by(Article.views.desc()).all()

# ④ 模糊搜索：标题或摘要里含「月」
kw = Article.query.filter(Article.title.like("%月%")).first()

# ⑤ 多条件组合：唐朝 + 已发布 + 浏览量降序
rows = (Article.query
        .filter_by(dynasty="唐", status=1)
        .order_by(Article.views.desc())
        .all())

# ⑥ 关联查询：某篇作品的所有书评（用上节的 relationship）
comments = Article.query.get(3).comments.filter_by(status=1).all()

from app.models import Comment, Article

# 增：造一个对象，add 后 commit
c = Comment(user_id=2, article_id=1, content="同感！", likes=0)
db.session.add(c)
db.session.commit()
print("新书评 id =", c.id)          # commit 后主键才会被回填

# 改：取出对象，改属性，再 commit（ORM 会自动生成 UPDATE）
a = db.session.get(Article, 1)
a.views = a.views + 1               # 浏览量 +1
db.session.commit()

# 批量增：add_all
db.session.add_all([
    Comment(user_id=1, article_id=3, content="豪放派代表！"),
    Comment(user_id=2, article_id=3, content="周郎赤壁，千古风流。"),
])
db.session.commit()

# 删：先取出对象，再 delete
bad = Comment.query.filter_by(content="垃圾广告").first()
if bad:
    db.session.delete(bad)
    db.session.commit()

from app.models import Article

# 方法一：Model.query.paginate(...)
pager = (Article.query
         .filter_by(status=1)
         .order_by(Article.views.desc())
         .paginate(page=1, per_page=3, error_out=False))

print("本页条数：", len(pager.items))
print("总条数：  ", pager.total)
print("总页数：  ", pager.pages)
print("有下一页：", pager.has_next)

# 方法二（Flask-SQLAlchemy 3.x 新的推荐写法）：
# from sqlalchemy import select
# pager = db.paginate(
#     select(Article).where(Article.status == 1).order_by(Article.views.desc()),
#     page=1, per_page=3, error_out=False)

for a in pager.items:
    print(a.id, a.title, a.views)

pip install flask-migrate
flask db init

flask db migrate -m "init user/category/article/comment"

flask db upgrade

from sqlalchemy import text

# ① 简单统计：ORM 用 func + group_by
from sqlalchemy import func
rows = (db.session.query(Article.dynasty,
                         func.count(Article.id),
                         func.avg(Article.views))
        .group_by(Article.dynasty)
        .all())

# ② 同样的结果，原生 SQL 更直观
sql = text("""
    SELECT dynasty, COUNT(*) AS cnt, ROUND(AVG(views), 1) AS avg_views
    FROM article
    WHERE status = :status
    GROUP BY dynasty
    HAVING cnt > 0
    ORDER BY cnt DESC
""")
rows = db.session.execute(sql, {"status": 1}).all()

for dynasty, cnt, avg_views in rows:
    print(dynasty, cnt, avg_views)