from fastapi import FastAPI
import uvicorn

#创建FastAPI实例
app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World666888"}

@app.get("/hello/{name}")
async def hello(name: str):
    return {"message": f"Hello {name}"}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000)

1.如何运行FastAPI项目？

run项目

uvicorn main:app --reload

2.怎么访问FastAPI文档

网址后面加/docs：http://127.0.01:8000/docs

基础知识

1.request and response

get request method：

在 FastAPI 中，GET 请求是最基础、最常用的 HTTP 方法。它的核心作用是查询和获取服务器上的资源（例如：获取书籍列表、查询用户详情等）

1.基础 GET 请求（获取全部数据）:

@app.get("/books")
async def read_all_books():
    return BOOKS

2.路径参数 (Path Parameter)：精确匹配单个资源

# 假设通过索引或 ID 获取某本书
@app.get("/books/{book_id}")
async def read_book(book_id: int):
    # 注意边界处理，防止越界报错
    if book_id < len(BOOKS):
        return BOOKS[book_id]
    return {"detail": "书籍不存在"}

3.查询参数 (Query Parameter)：过滤与分页

from fastapi import Query
from typing import Optional<websource>source_group_web_3</websource>

@app.get("/books/search")
async def search_books(
    category: str = Query(..., description="按分类搜索"), 
    author: Optional[str] = Query(None, description="可选的作者名称")
):
    results = [book for book in BOOKS if book.get("category") == category]
    if author:
        results = [book for book in results if book.get("author") == author]
    return results

post request method：

在 FastAPI 中，POST 请求的核心作用是向服务器提交或新增数据（如用户注册、发布文章等）1。与 GET 请求将参数暴露在 URL 中不同，POST 请求的数据通常隐藏在请求体（Request Body）中，因此非常适合传递复杂结构或密码等敏感信息

1.JSON 请求体：提交结构化数据（最常用 ⭐）

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 1. 定义数据结构模型
class BookCreate(BaseModel):
    title: str
    author: str
    category: str

# 2. 定义 POST 路由
@app.post("/books")
async def create_book(book: BookCreate):
    # FastAPI 会自动解析 JSON 并转换为 BookCreate 对象
    return {"message": "书籍创建成功", "book_data": book}

2.使用 Body() 显式声明

你的接口只需要接收简单的、非嵌套的字段（比如一个简单的登录接口），或者你想对单个基础数据类型做额外的元数据描述时，可以使用 fastapi.Body 来显式指定参数来源于请求体

from fastapi import Body

@app.post("/login")
def login(
    username: str = Body(..., description="用户名"), 
    password: str = Body(..., description="密码")
):
    return {"user": username}

put request method:

在 FastAPI 中，PUT 请求的核心作用是完整地更新服务器上已存在的资源。你可以把它想象成“用一份全新的文件去替换掉旧的文件”

from fastapi import FastAPI, Path, Query, Body
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

# 1. 定义请求体模型
class BookUpdateRequest(BaseModel):
    title: str
    author: str
    category: Optional[str] = None

@app.put("/books/{book_id}")
def update_book(
    book_id: int = Path(..., description="要更新的书籍ID"),
    notify: bool = Query(False, description="更新后是否发送通知"),
    book: BookUpdateRequest = Body(...)
):
    # 模拟更新逻辑
    return {
        "message": f"书籍 {book_id} 更新成功",
        "data": book.dict(),
        "notify_sent": notify
    }

delete request method:

在 FastAPI 和 RESTful API 设计中，DELETE 请求的核心作用是删除服务器上指定的资源（例如：下架商品、注销账户、删除评论等）

from fastapi import FastAPI, Path

app = FastAPI()

# 假设的初始书籍数据
BOOKS = [
    {'title': 'FastAPI指南', 'author': '刘杰', 'category': '编程'},
]

# 定义 DELETE 路由
@app.delete("/books/{book_id}")
async def delete_book(book_id: int):
    # 1. 检查要删除的书籍是否存在
    if book_id < len(BOOKS):
        removed_book = BOOKS.pop(book_id)  # 从列表中移除
        return {"message": f"ID为 {book_id} 的书籍已删除", "data": removed_book}
    
    return {"detail": "书籍不存在"}

2.建立一个Books项目

from fastapi import FastAPI
import uvicorn

app = FastAPI()

class Book:
    id: int
    title: str
    author: str
    description: str
    rating: int

    def __init__(self,id,title,author,description,rating):
        self.id = id
        self.title = title
        self.author = author
        self.description = description
        self.rating = rating


BOOKS = [
    Book(1,'cs','liujie','very nice',5),
    Book(2,'cs fastapi','liujie','very nice',6),
    Book(3,'cs python','liujie','very nice',9),
    Book(4,'cs llm','liujie','very nice',10)
]


@app.get("/books")
async def read_books():
    return BOOKS

2.1.使用Body

需要 from fastapi import Body

在 FastAPI 中，Body（请求体）是用于接收客户端通过 HTTP 请求（通常是 POST、PUT、PATCH 等）发送的复杂结构化数据（如 JSON 对象或数组）的核心机制 

@app.post("/create-books")
async def create_books(book_request = Body()):
    BOOKS.append(book_request)

2.2 Pydantics介绍

在 FastAPI 中，Pydantic 是不可或缺的核心基石。简单来说，Pydantic = Python 类型提示 + 自动数据校验 + 数据转换。它的主要作用是充当数据的“守门员”，帮你自动检查、清洗和转换外部传入的复杂或“脏”数据，让你能专注于业务逻辑

pydantic允许我们对数据进行验证，basemodel能够验证对象本身中的变亮的对象

进行请求体数据验证：

BookRequest and Book Conversation

from fastapi import FastAPI, Body, HTTPException
from pydantic import BaseModel
import uvicorn


class Book:
    id: int
    title: str
    author: str
    description: str
    rating: int

    def __init__(self,id,title,author,description,rating):
        self.id = id
        self.title = title
        self.author = author
        self.description = description
        self.rating = rating

class BookRequest(BaseModel):
    id: int
    title: str
    author: str
    description: str
    rating: int



@app.post("/create-books")
async def create_books(book_request: BookRequest ):
    new_book = Book(**book_request.dict())
    BOOKS.append(new_book)

为pydantic每个字段都引入数据验证

class BookRequest(BaseModel):
    id: Optional[int] = None
    title: str = Field(min_length=3)
    author: str
    description: str
    rating: int = Field(gt=0, lt= 6)

在 FastAPI 中接收前端数据、进行转换并存储到内存列表（模拟数据库）的完整流程。它主要包含两个部分：一个处理 POST 请求的路由函数，和一个用于生成自增 ID 的辅助函数。

@app.post("/create-books")
async def create_books(book_request: BookRequest ):
    new_book = Book(**book_request.dict())
    BOOKS.append(find_book_id(new_book))

def find_book_id(book : Book):
    book_id = 1 if len(BOOKS) == 0 else BOOKS[-1].id + 1
    return book

根据用户提供的评分（rating），从书籍列表中筛选并返回所有符合该评分的书籍。

@app.get("books/")
async def read_books_by_rating(book_rating: int):
    book_to_return = []
    for book in BOOKS:
        if book.rating == book_rating:
            book_to_return.append(book)
        return book_to_return

添加一个PUT request方法，允许我们在fastapi应用程序中更新对象或者数据。

@app.put("/book/update_book")
async def update_book(book: BookRequest):
    for i in range(len(BOOKS)):
        if BOOKS[i].id == book.id:
            BOOKS[i] = book

添加一个delete方法用于删除书籍

@app.delete("/book/{book_id}")
async def delete_book(book_id: int):
    for i in range(len(BOOKS)):
        if BOOKS[i].id == book_id:
            BOOKS.pop(i)
            break

以上是这个项目对put、delete、get、post request method的添加。

使用Path来进行数据验证（在路径参数中使用）

导入包path：from fastapi import Path

@app.get("/books/{book_id}")
async def read_book(book_id: int = Path(gt = 0, lt = 5)):
    return BOOKS[book_id]
    for book in BOOKS:
        if book.id == book_id:
            return book

使用Query来进行数据验证（在查询参数中使用）

@app.get("books/")
async def read_books_by_rating(book_rating: int = Query(gt = 0, lt = 5)):
    book_to_return = []
    for book in BOOKS:
        if book.rating == book_rating:
            book_to_return.append(book)
        return book_to_return

2.3 异常捕获与status code

使用HTTPExcerption捕获异常

了解常见status codes

@app.delete("/book/{book_id}")
async def delete_book(book_id: int):
    book_changed = False
    for i in range(len(BOOKS)):
        if BOOKS[i].id == book_id:
            BOOKS.pop(i)
            book_changed = True
            break
    if not book_changed:
        raise HTTPException(status_code=404, detail="Book not found")

2.4 正确返回时候，Explicit status code

导包：from starlette import status

@app.get("/books", status_code=status.HTTP_200_OK)
async def read_books():
    return BOOKS

2.5 完整代码

from typing import Optional
from fastapi import FastAPI, Body, HTTPException, Path,Query
from pydantic import BaseModel,Field
import uvicorn
from starlette import status

app = FastAPI()

class Book:
    id: int
    title: str
    author: str
    description: str
    rating: int
    published_date: int

    def __init__(self,id,title,author,description,rating,published_date):
        self.id = id
        self.title = title
        self.author = author
        self.description = description
        self.rating = rating
        self.published_date = published_date

class BookRequest(BaseModel):
    id: Optional[int]
    title: str = Field(min_length=3)
    author: str
    description: str
    rating: int = Field(gt=0, lt= 6)
    published_date: int = Field(gt=1999, lt=2031)


class Config:
    schema_extra = {
        "example": {
            "title": "cs python pro",
            "author": "liujie",
            "description": "cs python",
            "rating": 3,
            "published_date": 1999,
        }
    }

BOOKS = [
    Book(1,'cs','liujie','very nice',5,published_date=1999),
    Book(2,'cs fastapi','liujie','very nice',6,published_date=1999),
    Book(3,'cs python','liujie','very nice',9,published_date=2001),
    Book(4,'cs llm','liujie','very nice',10,published_date=1993),
]


@app.get("/books", status_code=status.HTTP_200_OK)
async def read_books():
    return BOOKS

@app.get("/books/{book_id}",status_code=status.HTTP_200_OK)
async def read_book(book_id: int = Path(gt = 0, lt = 5)):
    for book in BOOKS:
        if book.id == book_id:
            return book
    raise HTTPException(status_code=404, detail="Book not found")

@app.get("books/",status_code=status.HTTP_200_OK)
async def read_books_by_rating(book_rating: int = Query(gt = 0, lt = 5)):
    book_to_return = []
    for book in BOOKS:
        if book.rating == book_rating:
            book_to_return.append(book)
        return book_to_return

@app.get("/books/publish/",status_code=status.HTTP_200_OK)
async def read_books_by_rating(published_date: int):
    book_to_return = []
    for book in BOOKS:
        if book.published_date == published_date:
            book_to_return.append(book)

    return book_to_return

@app.post("/create-books",status_code=status.HTTP_201_CREATED)
async def create_books(book_request: BookRequest ):
    new_book = Book(**book_request.dict())
    BOOKS.append(find_book_id(new_book))

def find_book_id(book : Book):
    book_id = 1 if len(BOOKS) == 0 else BOOKS[-1].id + 1
    return book

@app.put("/book/update_book",status_code=status.HTTP_202_ACCEPTED)
async def update_book(book: BookRequest):
    for i in range(len(BOOKS)):
        if BOOKS[i].id == book.id:
            BOOKS[i] = book


@app.delete("/book/{book_id}",status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: int):
    book_changed = False
    for i in range(len(BOOKS)):
        if BOOKS[i].id == book_id:
            BOOKS.pop(i)
            book_changed = True
            break
    if not book_changed:
        raise HTTPException(status_code=404, detail="Book not found")




if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8003)

3.建立一个RESTful APIs项目

3.1 建立一个database

终端安装sql：pip install sqlalchemy 

创建数据库连接、创建会话(Session)、定义ORM模型基类(Base)。

3.1.1 导入模块

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

create_engine

用于创建数据库引擎（Engine），Engine负责和数据库通信。

可以理解为：

FastAPI
   ↓
SQLAlchemy
   ↓
Engine
   ↓
SQLite/MySQL/PostgreSQL

sessionmaker

用于创建数据库会话(Session)。

declarative_base

用于创建 ORM 模型基类。

3.1.2 数据库连接地址

SQLALCHEMY_DATABASE_URI = 'sqlite:///todos.db'

表示使用SQLite数据库

如果是mysql：

mysql+pymysql://root:123456@localhost/test

3.1.3 创建数据库引擎

engine = create_engine(
    SQLALCHEMY_DATABASE_URI,
    connect_args={"check_same_thread": False}
)

3.1.4 创建Session工厂

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

bind=engine

绑定数据库引擎。

3.1.5 创建ORM基类

Base = declarative_base()

总代码：

database.py文件如下：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URI = "sqlite:///todos.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URI,
    connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

Base = declarative_base()

3.2 models.py文件

定义数据库表结构（Table Structure）以及 Python 类与数据库表之间的映射关系（ORM Mapping）。

from database import Base
from sqlalchemy import Column, Integer, String, Boolean

class Todos(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    completed = Column(Boolean,default=False)
    

实际上等价于数据库中的：

CREATE TABLE todos (
    id INTEGER PRIMARY KEY,
    title VARCHAR,
    description VARCHAR,
    priority INTEGER,
    completed BOOLEAN DEFAULT FALSE
);

3.2.1. 继承 Base

class Todos(Base):

来自你刚才的 database.py

作用是告诉 SQLAlchemy：这是一个ORM模型，需要映射成数据库表

3.2.2 创建表

class Todos(Base):
    __tablename__ = "todos"

    id = Column(Integer, primary_key=True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    completed = Column(Boolean,default=False)

3.3 main.py文件

3.3.1 创建FastAPI应用

app = FastAPI()

3.3.2 创建数据库表

models.Base.metadata.create_all(bind=engine)

from fastapi import FastAPI
import models
from database import engine

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

3.4 数据库中加入元素

终端输入：sqlite3 todos.db（这里横线输入已经创建好的数据库名字）  

再输入：.schema

插入数据：insert into 数据库名字（feature1，feature2，...） values (值1，值2，...)

3.5 创建依赖项

3.5.1：get_db()

def get_db():
    db = SessionLocal() #这里真正创建数据库会话
    try:
        yield db
    finally:
        db.close()

这是一个 FastAPI 依赖函数。

请求进来
    ↓
SessionLocal()
    ↓
yield db
    ↓
read_all()
    ↓
finally
    ↓
db.close()

3.5.2 数据库连接依赖

db_dependency = Annotated[Session,Depends(get_db)]

3.5.3 定义接口

完整的main.py如下：

from fastapi import FastAPI,Depends,HTTPException,status
from sqlalchemy.sql.annotation import Annotated
from typing import Annotated
from sqlalchemy.orm import Session
import models
from database import engine
from models import Todos
from todoapp.database import SessionLocal
import uvicorn

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

db_dependency = Annotated[Session,Depends(get_db)]

@app.get("/")
async def read_all(db:db_dependency):
    return db.query(Todos).all()


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8004)

3.6 GET\POST\PUT\DELETE Request method in Database

3.6.1 GET Request method

根据 URL 中的 todo_id 查询单条 Todo 数据，如果找不到则返回 404。

@app.get("/todos/{todo_id}",status_code=status.HTTP_200_OK)
async def read_todo(db: db_dependency,todo_id:int = Path(gt=0)):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
    if todo_model is not None:
        return todo_model
    raise HTTPException(status_code=404, detail="Not found")

这是 SQLAlchemy ORM 查询。

todo_model = db.query(Todos)\
               .filter(Todos.id == todo_id)\
               .first()

注意⚠️：这个是传统的SQLAlchemy ORM 查询方式

2.0的写法如下：

result = await db.execute(
    select(Todos)
    .where(Todos.id == todo_id)
)

todo = result.scalars().first()

3.6.2 POST Request method

接收前端提交的 Todo 数据 → 校验数据 → 创建数据库记录 → 保存到数据库。

也就是一个典型的 新增（Create）接口。

class TodoRequest(BaseModel): #定义请求体
    title: str = Field(min_length=3)
    description: str = Field(min_length=3,max_length=100)
    priority: int
    completed: bool


@app.post("/todos",status_code=status.HTTP_201_CREATED) #定义post接口
async def create_todo(db: db_dependency,todo_request: TodoRequest):   #接收参数                            
    todo_model = Todos(**todo_request.dict()) #创建ORM对象，**表示字典解包

    db.add(todo_model) #把对象放进session中
    db.commit()。#提交事物

3.6.3 PUT Request method

根据 todo_id 找到数据库中的 Todo，然后用请求体中的新数据覆盖原来的数据。

整体流程：

PUT /todos/1
    ↓
查询 id=1 的 Todo
    ↓
存在？
 ├─ 否 → 返回404
 └─ 是
      ↓
更新字段
      ↓
commit()
      ↓
返回200

@app.put("/todos/{todo_id}",status_code=status.HTTP_200_OK)
async def update_todo(db: db_dependency, todo_id:int, todo_request: TodoRequest):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first() #查询数据库
    if todo_model is None:
        raise HTTPException(status_code=404, detail="Not found")

    #修改字段
    todo_model.title = todo_request.title
    todo_model.description = todo_request.description
    todo_model.priority = todo_request.priority
    todo_model.completed = todo_request.completed

    db.add(todo_model)
    db.commit()

3.6.4 DELETE Request method

根据 todo_id 删除指定的 Todo。

整体流程：

DELETE /todos/1
      ↓
查询 id=1 是否存在
      ↓
存在？
 ├─ 否 → 返回404
 └─ 是
      ↓
删除
      ↓
commit()
      ↓
返回204

@app.delete("/todos/{todo_id}",status_code=status.HTTP_204_NO_CONTENT)
async def delete_todo(db: db_dependency, todo_id:int):
    todo_model = db.query(Todos).filter(Todos.id == todo_id).first()
    if todo_model is None:
        raise HTTPException(status_code=404, detail="Not found")
    
    db.query(Todos).filter(Todos.id == todo_id).delete()
    db.commit()

3.7 Authentication and Authorization

from routers import auth

app.include_router(auto.router)

新建一个routers包，一个文件auth.py

一.路由

路由就是URL地址和处理函数之间的映射关系，他决定了当用户访问某个特定网址时候，服务器应该执行那段代码返回结果

FastAPI的路由定义基于python的装饰器模式

访问 /hello 响应结果 msg：你好 FastAPI

from fastapi import FastAPI
import uvicorn

app = FastAPI()

@app.get("/hello")
async def get_hello():
    return {"msg": "你好 FastAPI"}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

练习：

访问路径/user/hello，响应结果是{“msg”：“I'm learning FastAPI”}

from fastapi import FastAPI
import uvicorn

app = FastAPI()



@app.get("/user/hello")
async def get_hello():
    return {"msg": "i'm learning FastAPI"}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

二.参数简介和路径参数

1.参数介绍

同一段接口逻辑，根据参数不同返回不同的数据

参数就是客户端发送请求时附带的额外信息和指令

参数的作用是让同一个接口能根据不同的输入，返回不同的输出，实现动态交互。

2.参数分类

2.2.1路径参数

位置：URL路径一部分   /book/{id} 这里id就填路径参数的名字

作用：指向唯一的、特定的资源

方法：GET

这里id，根据用户的输入不同，响应不同的结果。

这个路径参数是必填的！！！

练习：

以用户ID为路径参数设计URL，要求响应结果包含用户id和名称。

from fastapi import FastAPI
import uvicorn

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {
        "id": user_id,
        "name": f"用户{user_id}"
    }

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

如何为路径参数类型注解？

FastAPI运行为参数声明额外的信息和校验

导入FastAPI的Path函数

from fastapi import FastAPI, Path
import uvicorn

app = FastAPI()


@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., gt=0, lt=100,description="取值范围1-100")):
    return {
        "id": user_id,
        "name": f"用户{user_id}"
    }

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

练习：

查找书籍的作者，路径参数name，长度范围2-10

from fastapi import FastAPI, Path
import uvicorn


app = FastAPI()


@app.get("/users/{name}")
async def get_user(name: str = Path(..., min_length=2, max_length=10, description="取值范围2-10")):
    return {
        "id": name,
        "name": f"用户{name}"
    }

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

练习：

定义两个接口，携带路径参数，并使用path注解

接口1:以新闻分类id为参数设计URL，id范围为1-100

接口2:以新闻分类名称name 作为参数设计URL，分类名称2-10

2.2.2 查询参数

声明的参数不是路径参数时候，路径操作函数会把参数自动解释为查询参数

位置：URL？之后

k1=v1&k2=v2

作用：对资源集合进行过滤、排序、分页等操作

方法：GET

如何为查询参数添加类型注解？

使用Query注解

实现：查询新闻->分页，skip：跳过的记录数，limit：返回的记录数 

from fastapi import FastAPI, Path, Query
import uvicorn


app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/news/news_list")
async def get_news_list(skip: int = Query(0,description="跳过的记录数",lt=100),
                        limit: int = Query(10,description="返回的记录数")):
    return {"skip": skip, "limit": limit}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

练习：

设计接口查询图书，要求携带两个查询参数：图书分类和价格

图书分类：默认值为python开发，长度限制5-255

价格：限制大小50-100

from fastapi import FastAPI, Path, Query
import uvicorn


app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/news/news_list")
async def get_news_list(category: str = Query(0,description="图书分类",min_length=5,max_length=255),
                        price: int = Query(10,description="价格",gt=50,lt=100)):
    return {"category": category, "price": price}

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

2.2.3 请求体参数

位置：HTTP请求的消息体（body）中

作用：创建、更新资源、携带大量数据。如：JSON

方法：POST、PUT

在HTTP协议中，一个完整的请求由三部分组成：

1.请求行：包含方法、URL、协议版本

2.请求头：元数据信息

3.请求体：实际要发送的数据内容

第一步：定义类型

第二步：类型注解

练习：

设计接口新增图书，图书信息包含：书名、作者、出版社、售价

from fastapi import FastAPI, Path, Query
import uvicorn
from pydantic import BaseModel

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}


#注册：用户名和密码->str
class Book(BaseModel):
    bookname: str
    author: str
    price: float
    publisher: str

@app.post("/bookinfo")
async def register(book: Book):
    return book

如何为请求体参数添加类型注解？

使用Field函数

导入pydantic的Field函数

from fastapi import FastAPI, Path, Query
import uvicorn
from pydantic import BaseModel,Field

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}


#注册：用户名和密码->str
class Book(BaseModel):
    bookname: str = Field(..., description="Book name",min_length=2,max_length=10)
    price: float = Field(..., description="Book price",gt=10)
    author: str
    price: float
    publisher: str

@app.post("/bookinfo")
async def register(book: Book):
    return book

三.响应类型

默认情况下，FastAPI会自动将路径操作函数返回python对象（字典、列表、pydantic模型等），经由jsonable_encoder转化为json兼容格式，并包装jsonResponse返回。这省去了手动序列化的步骤，让开发者更专注于业务逻辑。如果需要返回非json数据（html、文件流），fastAPI提供了丰富的响应类型来返回不同数据。

1.响应类型设置方式介绍

装饰器中指定响应类、返回响应对象

3.1.1 装饰器中指定响应类

如果你需要返回非 JSON 格式的内容（例如 HTML 页面、纯文本），可以在路径操作装饰器中通过 response_class 参数指定 FastAPI 内置的响应类

1.响应HTML格式

设置响应类为HTMLResponse，当前接口即可返回HTML内容

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import uvicorn

app = FastAPI()

@app.get("/", response_class=HTMLResponse)
def index():
    return "<h1>Hello FastAPI</h1>"

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

2.直接返回响应对象：

FileResponse是FastAPI提供的专门用于高校返回文件内容（如图片、pdf等）的响应类。它能够智能处理文件路径、媒体类型推断、范围请求和缓存头部，是服务静态文件的推荐方式。

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.responses import FileResponse
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

@app.get("/file")
async def file():
    path = "./data/pic1.png"
    return FileResponse(path)


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8001)

3.1.2 自定义响应数据格式

response_model是路径操作装饰器（如@app.get或@app.post）的关键参数，他是通过一个Pydantic模型来严格定义和约束API端点的输出格式。这一机制在提供自动数据验证和序列化的同时，更是保障数据安全性的第一道防线。

通过 response_model 参数配合 Pydantic 模型，可以严格定义接口返回的数据结构。FastAPI 会自动将返回值转换为该模型的格式，并过滤掉未在模型中声明的字段4。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 1. 定义期望的响应数据结构
class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    # 假设数据库里有 password，但我们不在模型里写，它就不会返回给前端

@app.get("/user", response_model=UserResponse)
def get_user():
    # 2. 即使这里返回了额外的字段（如 password），也会被自动过滤
    return {
        "id": 1, 
        "name": "Alice", 
        "email": "alice@example.com", 
        "password": "secret123" 
    }

1.如何自定义响应数据的格式？

以响应类型为JSONResponse为例：

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from fastapi.responses import FileResponse
from pydantic import BaseModel
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

class News(BaseModel):
    title: str
    content: str
    id: int


@app.get("/news/{id}",response_model=News)
async def get_news(id: int):
    return {
        "id": id,
        "title": f"这是第{id}本书",
        "content": "this is a good book",
    }


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

2.异常处理

对于客户端引发的错误（4xx，如资源未找到、认证失败），应使用fastapi.HTTPException来中断正常处理流程，并返回标准错误响应。

from fastapi import FastAPI
from fastapi import HTTPException
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

@app.get("/news/{id}")
async def get(id: int):
    id_list = [1,2,3,4,5,6]
    if id not in id_list:
        raise HTTPException(status_code=404, detail="Not Found")

    return {"id":id}



if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

四.中间件

1.中间价介绍

使用中间件为每个请求前后添加统一的处理逻辑

中间件是一个在每次请求进入FastAPI应用时都会被执行的函数。他在请求到达十几的路径操作（路由处理函数）之前运行，并且在响应返回给客户端之前再运行一次

2.中间件写法

中间件：函数的顶部使用装饰器@app.middleware("http")

中间件执行：按代码顺序，自下向上

from fastapi import FastAPI
from fastapi import HTTPException
import uvicorn

app = FastAPI()

@app.middleware("http")
async def middleware(request, call_next):
    print("中间件1 start")
    response = await call_next(request)
    print("中间件1 end")
    return response

@app.middleware("http")
async def middleware2(request, call_next):
    print("中间件2 start")
    response = await call_next(request)
    print("中间件2 end")
    return response

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}



if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

五.依赖注入系统

5.1 介绍

使用依赖注入系统来共享通用逻辑，减少代码重复

依赖项：可重用的组件（函数/类），负责提供某种功能或数据

注入：FastAPI自动帮你调用依赖项，并将结果注入到路径操作函数中。

优点：

代码复用：一次编写，多次使用

解耦：业务逻辑与基础设施代码分离

易于测试：轻松地用模拟依赖替换真实依赖进行测试

5.2 依赖注入系统应用场景

5.3 依赖注入系统的使用

步骤一：创建依赖项

步骤二：导入Depends

步骤三：声明依赖项

from fastapi import FastAPI, Query, Depends
from fastapi import HTTPException
import uvicorn

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

#分页参数逻辑共用：新闻列表和用户
#步骤一：依赖项
async def common_parameters(
        skip: int = Query(0, ge=0),
        limit: int = Query(10, le=60),
):
    return {"skip": skip, "limit": limit}

#步骤二：声明依赖项->依赖注入
@app.get("/news/news_list")
async def get_news_list(
        commons = Depends(common_parameters),
):
    return commons


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

六.ORM对象关系映射

6.1 ORM简介

ORM（对象关系映射）是一种编程技术，用于在面向对象编程语言和关系数据库之间建立映射。它允许开发者通过操作对象的方式与数据库进行交互，而无需直接编写复杂的sql语句

优势：

减少重复的sql代码；代码更简洁易读；自动处理数据库连接和事物；自动防止sql注入攻击

在pycharm终端输入以下：（mac版本）

pip install "sqlalchemy[asyncio]" aiomysql 

6.2 ORM的建表

语句：create database xxx；（创建数据库）

1.创建数据库引擎：使用create_async_engine创建异步引擎

2.定义模型类

基类，继承DeclarativeBase（包含通用属性和字段的映射）

定义数据库表对应的模型类

3.建表

从连接池获取异步连接，开启事物，执行ORM操作

FastAPI应用启动时，创建数据库表

from fastapi import FastAPI
import uvicorn
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy import String, DateTime, func, Float
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

#1.创建异步引擎
ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./book.db"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True, #可选，输出sql日志
    pool_size=10, #设置连接池活跃的连接数
    max_overflow=20, #允许额外的连接数
)

#2.定义模型类：基类+表对应的模型类
#基类：创建时间、更新时间；书集表：id、书名、作者
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, comment ="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, onupdate=func.now())


class Book(Base):
    __tablename__ = "book"
    id:Mapped[int] = mapped_column(primary_key=True, comment="book_id")
    bookname:Mapped[str] = mapped_column(String(255),comment="book_name")
    author: Mapped[str] = mapped_column(String(255),comment="book_author")


#3.建表：定义函数建表，FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事物异步引擎
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all) #Base模型类的元数据创建

@app.on_event("startup")
async def startup_event():
    await create_tables()



if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

6.3 在路由匹配中使用ORM

核心：创建依赖项，使用Depends注入到处理函数

from fastapi import FastAPI,Depends,HTTPException
import uvicorn
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy import String, DateTime, func, Float, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello FastAPI"}

#1.创建异步引擎
ASYNC_DATABASE_URL = "sqlite+aiosqlite:///./book.db"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True, #可选，输出sql日志
    pool_size=10, #设置连接池活跃的连接数
    max_overflow=20, #允许额外的连接数
)

#2.定义模型类：基类+表对应的模型类
#基类：创建时间、更新时间；书集表：id、书名、作者、价格、出版商
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, comment ="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now, onupdate=func.now())


class Book(Base):
    __tablename__ = "book"
    id:Mapped[int] = mapped_column(primary_key=True, comment="book_id")
    bookname:Mapped[str] = mapped_column(String(255),comment="book_name")
    author: Mapped[str] = mapped_column(String(255),comment="book_author")



#3.建表：定义函数建表，FastAPI启动的时候调用建表的函数
async def create_tables():
    #获取异步引擎，创建事物异步引擎
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all) #Base模型类的元数据创建

@app.on_event("startup")
async def startup_event():
    await create_tables()


#需求：查询功能的接口、查询图书->依赖注入：创建依赖项获取数据库绘画+Depends注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine, #绑定数据库引擎
    class_=AsyncSession, #指定会话类
    expire_on_commit=False #提交后回话不过期，不会重新查询数据库
)


#依赖项
async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session #返回数据库绘画给路由处理函数
            await session.commit()  #提交事物
        except Exception:
            await session.rollback() #有异常就会滚
            raise
        finally:
            await session.close() #关闭会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    result = await db.execute(select(Book))
    book = result.scalars().all()
    return book


if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8002)

6.4 ORM操作数据

6.4.1 查询

核心语句：await db.execute(select(模型类))，返回一个ORM对象

1.获取所有数据

scalars().all()

2.获取单条数据

scalars().first()

get(模型类，主健值)

#查询数据
@app.get("/book/books")
async def get_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book)) #获取一个orm对象
    book_all = result.scalars().all() #获取所有
    book_first = result.scalars().first() #获取第一条数据
    book_first2 = await db.get(Book,5)  #获取一条数据
    return book_all,book_first,book_first2

6.4.2 条件查询

select(Book).where(条件1,条件2,...)

比较判断

#条件查询-比较判断
@app.get("/book/books/{book_id}")
async def get_book(db: AsyncSession = Depends(get_database), book_id: int):
    result = await db.execute(select(Book).where(Book.id == book_id)) 
    book = result.one_or_none()
    return book

#条件查询：价格大于等于200
@app.get("/book/books")
async def get_book_price(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.price >= 200) 
    books = result.scalars().all()
    return books

模糊查询+与非查询

#模糊查询 作者以刘开头并且价格大于200
@app.get("/book/search_books")
async def get_book_search(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.author.like("刘%")) & (Book.price > 200))
    books2 = result.scalars().all()
    return books2

6.4.3 聚合查询

result = await db.execute( select ( func.方法(模型类.属性) ) )

num = result.sclar() //用来提取一个数值

@app.get("/book/count")
async def get_book_count(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(func.count(Book.id)))
    #result = await db.execute(select(func.max(Book.price)))
    book_count = result.scalars()
    return book_count

6.4.4 分页查询

select().offset().limit()

offset：跳过的记录数

limit：返回的记录数

offset值 = （当前页码 - 1）*每页数量 limit

@app.get("/book/get_books")
async def get_books(
        page: int = Query(1, description = "页数"),
        page_size: int = Query(3,description="页面数量"),
        db: AsyncSession = Depends(get_database)
                    ):
    skip = (page - 1) * page_size
    stmt = select(Book).offset(skip).limit(page_size)
    result = await db.execute(stmt)
    books = result.scalars().all()
    return {"books":books}

ORM-查询-总结

6.4.5 数据库操作--新增数据

核心步骤：定义ORM对象->添加对象到事物：add(对象)->commit提交到数据库

class BookBase(BaseModel):
    id:int
    bookname:str
    author:str

#用户输入图书信息（id，书名；作者）
@app.post("/book/add_books")
async def add_books(book: BookBase, db: AsyncSession = Depends(get_database)):
    #orm对象->add->commit
    Book_obj(**book.__dict__)
    db.add(Book_obj)
    await db.commit()
    return book

6.4.6 数据库操作--更新数据

核心步骤：查询get->属性重新赋值->commit提交到数据库

#修改图书信息
#路径参数id；作用是查找；请求体参数：作用是新数据（书名、作者、价格）

class Bookupdate(BaseModel):
    id:int
    author:str
    price:float
    
    
@app.put("/book/update_books/{book_id}")
async def update_book(book_id: int, data : Bookupdate, db:AsyncSession = Depends(get_database))
    db_book = await db.get(Book, book_id)
    if db_book is None:
        raise HTTPException(status_code=404, detail="Book not found")
    
    #重新赋值
    db_book.bookname = data.bookname
    db_book.author = data.author
    db_book.price = data.price
    
    await db.commit()
    return db_book
    

6.4.7 数据库操作--删除数据

核心步骤：查询get->delete删除->commit提交到数据库