[FastAPI-25]博客接口-526互联

import typing

from fastapi import FastAPI, Query, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI(title="Blog CRUD")

blogs = {
    1: {
        "id": 1,
        "title": "blog1",
        "body": "This is blog1",
        "desc": "desc"
    },
    2: {
        "id": 2,
        "title": "blog2",
        "body": "This is blog2",
        "desc": "desc"
    }
}


class Blog(BaseModel):
    title: str
    body: str
    desc: str

class BlogUpdate(Blog):
    # 博客编辑的时候不是必选项
    title: typing.Optional[str] = None
    body: typing.Optional[str] = None
    desc: typing.Optional[str] = None

@app.get("/blogs",
         tags=["博客"],
         summary="博客列表接口",
         description="page页码，size每页显示的数量"
         )
def get_blogs(page: typing.Optional[int] = Query(description="页码", default=1),
              size: int = 10):
    _blogs_list = list(blogs.values())
    return _blogs_list[(page - 1) * size:page * size]


@app.get("/blog",
         tags=["博客"],
         summary="博客详情页")
def get_blog_id(blog_id: int):
    # 字典的get方法效率优于循环遍历
    blog = blogs.get(blog_id)
    if not blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id :{blog_id}"
        )
    return blog


@app.post("/blog",
          tags=["博客"],
          summary="创建新的博客使用POST")
def create_blog(blog: Blog):
    blog_id = len(blogs) + 1
    # **blog.dict()
    # 序列化数据nable_encoder
    blogs[blog_id] = {"id": blog_id, **jsonable_encoder(blog)}
    return blogs[blog_id]


@app.put("/blog",
         tags=["博客"],
         summary="整体更新博客使用PUT")
def update_blog(blog_id: int, blog: Blog):
    to_update_blog = blogs.get(blog_id)
    if not to_update_blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id :{blog_id}"
        )
    to_update_blog.update(jsonable_encoder(blog))
    blogs[blog_id] = to_update_blog
    return to_update_blog

@app.patch("/blog",
           tags=["博客"],
           summary="局部更新博客使用Patch")
def part_update_blog(blog_id: int, blog: BlogUpdate):
    to_update_blog = blogs.get(blog_id)
    if not to_update_blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id :{blog_id}"
        )

    to_update_blog.update(**jsonable_encoder(blog,exclude_unset=True))
    blogs[blog_id] = to_update_blog
    return to_update_blog

@app.delete("/blog",
            tags=["博客"],
            summary="删除博客使用Delete"
            )
def delete_blog(blog_id:int):
    to_delete_blog = blogs.get(blog_id)
    if not to_delete_blog:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Not found the blog with id :{blog_id}"
        )
    return blogs.pop(blog_id)