diff --git a/models/blog.py b/models/blog.py index a498148..74ab4cb 100644 --- a/models/blog.py +++ b/models/blog.py @@ -1,100 +1,129 @@ -from sqlalchemy import ( - BigInteger, - Boolean, - Column, - Integer, - DECIMAL, - SMALLINT, - String, - LargeBinary -) +from sqlalchemy import (BigInteger, Boolean, Column, Integer, DECIMAL, String, LargeBinary) from sqlalchemy.orm import relationship from models.base import AuditBase, IdBase class Blog(AuditBase): - title = Column(String(255), nullable=True, comment="博客标题") - top_value = Column(Integer, nullable=True, comment="置顶值 越大越靠前") - is_great = Column(Boolean, nullable=True, comment="是否是精品 0否1是") - category_id = Column(BigInteger, nullable=True, comment="博客类别") - summary = Column(String(255), nullable=True, comment="博客内容概要") - content_id = Column(BigInteger, nullable=True, comment="博客内容") - word_count = Column(Integer, nullable=True, comment="字数统计") - read_duration = Column(DECIMAL(10, 2), nullable=True, comment="阅读时长") - is_approved = Column(SMALLINT, nullable=True, comment="是否发布") + title = Column(String(255), nullable=False, comment="博客标题") + top_value = Column(Integer, nullable=False, default=0, comment="置顶值 越大越靠前") + is_great = Column(Boolean, nullable=False, default=False, comment="是否是精品") + category_id = Column(BigInteger, nullable=False, comment="博客类别") + summary = Column(String(255), nullable=False, comment="博客内容概要") + content_id = Column(BigInteger, nullable=False, comment="博客内容") + word_count = Column(Integer, nullable=False, default=0, comment="字数统计") + read_duration = Column(DECIMAL(10, 2), nullable=False, default=0.00, comment="阅读时长") + is_approved = Column(Boolean, nullable=False, default=False, comment="是否发布") + # 分类关系（多对一） category = relationship( "BlogCategory", + # 与BlogCategory的blogs属性建立双向关系 back_populates="blogs", + # 只级联保存和合并操作，不级联删除（删除博客不应删除分类 + cascade="save-update, merge", + # 明确指定连接条件 + # 如果数据库设置了外键可以省略 primaryjoin="foreign(Blog.category_id) == BlogCategory.id" ) + # 内容关系（一对一） content = relationship( "BlogContent", + # 与BlogContent的blog属性建立双向关系 back_populates="blog", + # 完全级联操作：保存、合并、刷新、删除等所有操作都会级联 + cascade="all, delete-orphan", + # 设置为False表示一对一关系，返回单个对象而不是列表 + uselist=False, + # 确保内容只有一个父博客，与delete-orphan配合使用 + single_parent=True, + # 明确指定连接条件 primaryjoin="foreign(Blog.content_id) == BlogContent.id" ) + # 访问记录（一对多） visits = relationship( "BlogVisit", + # 与BlogVisit的blog属性建立双向关系 back_populates="blog", + # 完全级联操作：博客删除时自动删除所有访问记录 + cascade="all, delete-orphan", + # 明确指定连接条件 primaryjoin="Blog.id == foreign(BlogVisit.blog_id)" ) + # 4. 评论（一对多） comments = relationship( "BlogComment", + # 与BlogComment的blog属性建立双向关系 back_populates="blog", + # 完全级联操作：博客删除时自动删除所有评论 + cascade="all, delete-orphan", + # 明确指定连接条件 primaryjoin="Blog.id == foreign(BlogComment.blog_id)" ) class BlogCategory(AuditBase): - name = Column(String(45), nullable=True, comment="类别名称") + name = Column(String(45), nullable=False, comment="类别名称") blogs = relationship( "Blog", + # 与Blog的category属性建立双向关系 back_populates="category", + # 完全级联操作：分类删除时自动删除所有关联的博客 + # 警告：这会级联删除分类下的所有博客，包括博客的内容、访问记录和评论 + cascade="all, delete-orphan", + # 明确指定连接条件 primaryjoin="BlogCategory.id == foreign(Blog.category_id)" ) class BlogContent(IdBase): - content = Column(LargeBinary, comment="博客内容") + content = Column(LargeBinary, nullable=False, comment="博客内容") blog = relationship( "Blog", + # 与Blog的content属性建立双向关系 back_populates="content", + # 设置为False表示一对一关系 + uselist=False, + # 明确指定连接条件 primaryjoin="BlogContent.id == foreign(Blog.content_id)" ) class BlogVisit(AuditBase): - ip = Column(String(255), nullable=True, comment="ip地址") - os = Column(String(255), nullable=True, comment="操作系统") - browser = Column(String(255), nullable=True, comment="浏览器") - uri = Column(String(255), nullable=True, comment="路径") - blog_id = Column(BigInteger, nullable=True, comment="博客id") + ip = Column(String(255), nullable=False, comment="IP地址") + os = Column(String(255), nullable=False, comment="操作系统") + browser = Column(String(255), nullable=False, comment="浏览器") + uri = Column(String(255), nullable=False, comment="路径") + blog_id = Column(BigInteger, nullable=True, comment="博客ID") blog = relationship( "Blog", + # 与Blog的visits属性建立双向关系 back_populates="visits", + # 明确指定连接条件 primaryjoin="foreign(BlogVisit.blog_id) == Blog.id" ) class BlogComment(AuditBase): - blog_id = Column(BigInteger, nullable=True, comment="博客id") - parent_id = Column(BigInteger, nullable=True, comment="父评论id") - name = Column(String(255), nullable=True, comment="评论人昵称") + blog_id = Column(BigInteger, nullable=False, comment="博客ID") + parent_id = Column(BigInteger, nullable=False, comment="父评论ID") + name = Column(String(255), nullable=False, comment="评论人昵称") website = Column(String(255), nullable=True, comment="评论人网站") - ip_address = Column(String(45), nullable=True, comment="评论人ip") - user_agent = Column(String(255), nullable=True, comment="评论人浏览器信息") - content = Column(String(512), nullable=True, comment="评论内容") - is_approved = Column(SMALLINT, nullable=True, comment="是否通过") + ip_address = Column(String(45), nullable=False, comment="评论人IP") + user_agent = Column(String(255), nullable=False, comment="评论人浏览器信息") + content = Column(String(255), nullable=False, comment="评论内容") + is_approved = Column(Boolean, nullable=False, default=False, comment="是否通过") blog = relationship( "Blog", + # 与Blog的comments属性建立双向关系 back_populates="comments", + # 明确指定连接条件 primaryjoin="foreign(BlogComment.blog_id) == Blog.id" ) diff --git a/schemas/blog.py b/schemas/blog.py index 8501981..7a872d6 100644 --- a/schemas/blog.py +++ b/schemas/blog.py @@ -1,25 +1,42 @@ from typing import Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field, field_validator from datetime import datetime class BlogQuery(BaseModel): - # 类别 - category: Optional[str] = None - # 标题 - title: Optional[str] = None - # 年份 - year: Optional[int] = None + """博客查询参数""" + category: Optional[str] = Field(None, description="分类名称") + title: Optional[str] = Field(None, description="标题关键词") + year: Optional[int] = Field(None, description="发布年份", ge=2000, le=datetime.now().year) + + @field_validator('year') + def validate_year(cls, v): + if v is not None and v > datetime.now().year: + raise ValueError('年份不能超过当前年份') + return v class BlogBase(BaseModel): - title: str - topValue: int - isGreat: bool - category: str - content: Optional[str] = None - isApproved: int + """博客基础模型""" + title: str = Field(..., min_length=1, max_length=255, description="博客标题") + top_value: int = Field(default=0, ge=0, description="置顶值，越大越靠前", alias="topValue") + is_great: bool = Field(default=False, description="是否是精品", alias="isGreat") + category: str = Field(..., min_length=1, max_length=45, description="分类名称") + content: Optional[str] = Field(None, description="博客内容") + is_approved: bool = Field(default=False, description="是否已发布", alias="isApproved") + + @field_validator('title') + def title_not_empty(cls, v): + if not v or not v.strip(): + raise ValueError('标题不能为空') + return v.strip() + + @field_validator('category') + def category_not_empty(cls, v): + if not v or not v.strip(): + raise ValueError('分类不能为空') + return v.strip() class BlogCreate(BlogBase): @@ -31,16 +48,17 @@ class BlogUpdate(BlogBase): class BlogResponse(BlogBase): - id: Optional[int] = None - summary: Optional[str] = None - wordCount: Optional[int] = None - readDuration: Optional[float] = None - visitCount: Optional[int] = None - createTime: Optional[datetime] = None - updateTime: Optional[datetime] = None + id: int = Field(..., description="博客ID") + summary: Optional[str] = Field(None, description="内容摘要") + word_count: Optional[int] = Field(None, description="字数统计", alias="wordCount") + read_duration: Optional[float] = Field(None, description="阅读时长", alias="readDuration") + visit_count: Optional[int] = Field(0, description="访问次数", alias="visitCount") + create_time: datetime = Field(..., description="创建时间", alias="createTime") + update_time: datetime = Field(..., description="更新时间", alias="updateTime") class Config: from_attributes = True + populate_by_name = True json_encoders = { # 自定义 datetime 类型的序列化格式 datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S') @@ -48,65 +66,82 @@ class BlogResponse(BlogBase): class BlogCategoryResponse(BaseModel): - name: str - count: int + name: str = Field(..., description="分类名称") + count: int = Field(..., description="博客数量") class Config: from_attributes = True + populate_by_name = True class BlogStatsResponse(BaseModel): - blogCount: int - categoryCount: int - wordCount: int + blog_count: int = Field(..., description="博客总数", alias="blogCount") + category_count: int = Field(..., description="分类总数", alias="categoryCount") + word_count: int = Field(..., description="总字数", alias="wordCount") class BlogVisitResponse(BaseModel): - ip: str - os: str - browser: str - uri: str - title: Optional[str] = None - visitTime: datetime + ip: str = Field(..., description="IP地址") + os: str = Field(..., description="操作系统") + browser: str = Field(..., description="浏览器") + uri: str = Field(..., description="访问路径") + title: str = Field(None, description="博客标题") + visit_time: datetime = Field(..., description="访问时间", alias="visitTime") class Config: from_attributes = True + populate_by_name = True json_encoders = { datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S') } class BlogLatestResponse(BaseModel): - id: int - title: str + id: int = Field(..., description="博客ID") + title: str = Field(..., description="博客标题") class Config: from_attributes = True + populate_by_name = True class BlogAdjacentResponse(BaseModel): - id: int - title: str + id: int = Field(..., description="博客ID") + title: str = Field(..., description="博客标题") class Config: from_attributes = True + populate_by_name = True class BlogCommentCreate(BaseModel): - parentId: int - name: str - website: Optional[str] = None - content: str + parent_id: int = Field(default=0, ge=0, description="父评论ID，0表示顶级评论", alias="parentId") + name: str = Field(..., min_length=1, max_length=50, description="评论人昵称") + website: Optional[str] = Field(None, description="评论人网站") + content: str = Field(..., min_length=1, max_length=1000, description="评论内容") + + @field_validator('name') + def name_not_empty(cls, v): + if not v or not v.strip(): + raise ValueError('昵称不能为空') + return v.strip() + + @field_validator('content') + def content_not_empty(cls, v): + if not v or not v.strip(): + raise ValueError('评论内容不能为空') + return v.strip() class BlogCommentResponse(BlogCommentCreate): - id: int - ipAddress: str - userAgent: str - createTime: datetime + id: int = Field(..., description="评论ID") + ip_address: str = Field(..., description="IP地址", alias="ipAddress") + user_agent: str = Field(..., description="浏览器信息", alias="userAgent") + create_time: datetime = Field(..., description="创建时间", alias="createTime") class Config: from_attributes = True + populate_by_name = True json_encoders = { # 自定义 datetime 类型的序列化格式 datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S') diff --git a/service/blog_service.py b/service/blog_service.py index a480603..47f6bed 100644 --- a/service/blog_service.py +++ b/service/blog_service.py @@ -1,7 +1,7 @@ from typing import List from fastapi import Request -from sqlalchemy import select, func, desc, and_, asc, delete, distinct +from sqlalchemy import select, func, desc, and_, asc from sqlalchemy.orm import Session from models.blog import Blog, BlogCategory, BlogVisit, BlogContent, BlogComment @@ -22,16 +22,16 @@ def query_blog_by_page(db: Session, current_page: int = 1, page_size: int = 10, stmt = (select( Blog.id, Blog.title, - Blog.top_value.label("topValue"), - Blog.is_great.label("isGreat"), + Blog.top_value, + Blog.is_great, BlogCategory.name.label("category"), Blog.summary, - Blog.word_count.label("wordCount"), - Blog.read_duration.label("readDuration"), + Blog.word_count, + Blog.read_duration, func.count(BlogVisit.id).label("visitCount"), - Blog.is_approved.label("isApproved"), - Blog.create_time.label("createTime"), - Blog.update_time.label("updateTime") + Blog.is_approved, + Blog.create_time, + Blog.update_time ).outerjoin(BlogCategory, Blog.category_id == BlogCategory.id) .outerjoin(BlogVisit, Blog.id == BlogVisit.blog_id) .group_by(Blog.id, BlogCategory.name) @@ -53,17 +53,17 @@ def get_query_blog_by_condition_stmt(blog_query: BlogQuery): stmt = (select( Blog.id, Blog.title, - Blog.top_value.label("topValue"), - Blog.is_great.label("isGreat"), + Blog.top_value, + Blog.is_great, BlogCategory.name.label("category"), Blog.summary, - BlogContent.content.label("content"), - Blog.word_count.label("wordCount"), - Blog.read_duration.label("readDuration"), + BlogContent.content, + Blog.word_count, + Blog.read_duration, func.count(BlogVisit.id).label("visitCount"), - Blog.is_approved.label("isApproved"), - Blog.create_time.label("createTime"), - Blog.update_time.label("updateTime") + Blog.is_approved, + Blog.create_time, + Blog.update_time ).where(Blog.is_approved == 1) .outerjoin(BlogCategory, Blog.category_id == BlogCategory.id) .outerjoin(BlogContent, Blog.content_id == BlogContent.id) @@ -99,17 +99,17 @@ def query_unapproved_blog(db: Session) -> List[BlogResponse]: stmt = ((select( Blog.id, Blog.title, - Blog.top_value.label("topValue"), - Blog.is_great.label("isGreat"), + Blog.top_value, + Blog.is_great, BlogCategory.name.label("category"), Blog.summary, - BlogContent.content.label("content"), - Blog.word_count.label("wordCount"), - Blog.read_duration.label("readDuration"), - func.count(BlogVisit.id).label("visitCount"), - Blog.is_approved.label("isApproved"), - Blog.create_time.label("createTime"), - Blog.update_time.label("updateTime") + BlogContent.content, + Blog.word_count, + Blog.read_duration, + func.count(BlogVisit.id), + Blog.is_approved, + Blog.create_time, + Blog.update_time ).where(Blog.is_approved == 0) .outerjoin(BlogCategory, Blog.category_id == BlogCategory.id) .outerjoin(BlogContent, Blog.content_id == BlogContent.id)) @@ -126,19 +126,19 @@ def query_blog_by_id(db: Session, blog_id: int) -> BlogResponse: check_blog_exist(db, blog_id) stmt = (select( - Blog.id.label("id"), - Blog.title.label("title"), - Blog.top_value.label("topValue"), - Blog.is_great.label("isGreat"), + Blog.id, + Blog.title, + Blog.top_value, + Blog.is_great, BlogCategory.name.label("category"), - Blog.summary.label("summary"), - BlogContent.content.label("content"), - Blog.word_count.label("wordCount"), - Blog.read_duration.label("readDuration"), + Blog.summary, + BlogContent.content, + Blog.word_count, + Blog.read_duration, func.count(BlogVisit.id).label("visitCount"), - Blog.is_approved.label("isApproved"), - Blog.create_time.label("createTime"), - Blog.update_time.label("updateTime") + Blog.is_approved, + Blog.create_time, + Blog.update_time ).where(Blog.id == blog_id) .outerjoin(BlogCategory, Blog.category_id == BlogCategory.id) .outerjoin(BlogContent, Blog.content_id == BlogContent.id) @@ -146,7 +146,7 @@ def query_blog_by_id(db: Session, blog_id: int) -> BlogResponse: blog = db.execute(stmt).first() - return BlogResponse.from_orm(blog) + return BlogResponse.model_validate(blog) def add_blog(db: Session, blog: BlogCreate) -> bool: @@ -154,16 +154,18 @@ def add_blog(db: Session, blog: BlogCreate) -> bool: db_blog = Blog( title=blog.title, - top_value=blog.topValue, - is_great=blog.isGreat, + top_value=blog.top_value, + is_great=blog.is_great, category_id=add_blog_category(db, blog.category), - content_id=add_blog_content(db, blog.content), summary=get_blog_summary(blog.content), word_count=word_count, read_duration=get_read_duration(word_count), - is_approved=blog.isApproved + is_approved=blog.is_approved ) + # 级联新增 + db_blog.content = BlogContent(content=blog.content.encode('utf-8')) + db.add(db_blog) db.commit() db.refresh(db_blog) @@ -194,26 +196,17 @@ def add_blog_category(db: Session, category: str) -> int: return category_id -def add_blog_content(db: Session, content: str) -> int: - db_blog_content = BlogContent(content=content.encode('utf-8')) - db.add(db_blog_content) - db.commit() - - return db_blog_content.id - - def update_blog(db: Session, blog_id: int, blog: BlogUpdate) -> bool: db_blog = check_blog_exist(db, blog_id) - update_data = blog.model_dump(exclude_unset=True) - - update_blog_content(db, db_blog.content_id, blog.content) - - db_blog.title = update_data["title"] - db_blog.top_value = update_data["topValue"] - db_blog.is_great = update_data["isGreat"] + db_blog.title = blog.title + db_blog.top_value = blog.topValue + db_blog.is_great = blog.isGreat db_blog.category_id = update_blog_category(db, blog.category) + # 级联更新 + db_blog.content = BlogContent(content=blog.content.encode('utf-8')) + word_count = get_word_count(blog.content) db_blog.summary = get_blog_summary(blog.content), db_blog.word_count = word_count @@ -235,30 +228,11 @@ def update_blog_category(db: Session, category: str) -> int: return db_blog_category.id -def update_blog_content(db: Session, blog_content_id: int, blog_content: str) -> bool: - db_blog_content = check_blog_content_exist(db, blog_content_id) - - db_blog_content.content = blog_content.encode('utf-8') - db.commit() - db.refresh(db_blog_content) - - return True - - def delete_blog(db: Session, blog_id: int) -> bool: db_blog = check_blog_exist(db, blog_id) - delete_blog_content(db, db_blog.content_id) - db.execute(delete(Blog).where(Blog.id == blog_id)) - db.commit() - - return True - - -def delete_blog_content(db: Session, blog_content_id: int) -> bool: - check_blog_content_exist(db, blog_content_id) - - db.execute(delete(BlogContent).where(BlogContent.id == blog_content_id)) + # 级联删除 + db.delete(db_blog) db.commit() return True @@ -341,15 +315,15 @@ def query_blog_visit(db: Session, current_page: int = 1, page_size: int = 10) -> def query_blog_comment(db: Session, blog_id: int) -> List[BlogCommentResponse]: stmt = select( BlogComment.id, - BlogComment.blog_id.label("blogId"), - BlogComment.parent_id.label("parentId"), + BlogComment.blog_id, + BlogComment.parent_id, BlogComment.name, BlogComment.website, - BlogComment.ip_address.label("ipAddress"), - BlogComment.user_agent.label("userAgent"), + BlogComment.ip_address, + BlogComment.user_agent, BlogComment.content, - BlogComment.is_approved.label("isApproved"), - BlogComment.create_time.label("createTime") + BlogComment.is_approved, + BlogComment.create_time ).where(BlogComment.blog_id == blog_id, BlogComment.is_approved == 1).order_by(desc(BlogComment.create_time)) results = db.execute(stmt).fetchall() @@ -377,7 +351,7 @@ def add_blog_comment(db: Session, request: Request, blog_id: int, blog_comment: content=blog_comment.content, ip_address=request.client.host, user_agent=request.headers.get("user-agent"), - is_approved=1 + is_approved=True ) db.add(db_comment)