文章摘要

🌈 嗨！欢迎来到 FastAPI 的魔法世界！ 🌈

🌈 首先，我们要做的是安装 FastAPI 和 Uvicorn。 🎉

💡 安装依赖

pip install fastapi uvicorn

🎉 太棒了！现在我们有了构建 API 的超级工具！ 🎉

🌈 接下来，让我们一起创建一个 FastAPI 的小应用吧！ 🌈

🎯 目标： 我们要创建一个可以处理 GET 和 POST 请求的小应用！

🚀 步骤一： 引入我们的库！

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

🌟 理解它们：

• Query & Body：帮我们处理 GET 和 POST 请求的参数。
• BaseModel：来自 Pydantic，用于定义数据模型。
• uvicorn：启动服务器。

🚀 步骤二： 创建 FastAPI 实例。

app = FastAPI()

🌟 理解它们：

• FastAPI()：这就像召唤出你的应用精灵！

🚀 步骤三： 定义数据模型。

class Item(BaseModel):
    name: str
    description: str = None # 注意，这里的默认值为None，也就是可选参数～
    price: float
    tax: float = None

🌟 理解它们：

• Item：一个数据模型，像一本魔法书，记录着商品的信息。

🚀 步骤四： 定义 GET 请求处理函数。

@app.get("/items/")

async def read_items(q: str = Query(None, max_length=50)):
    return {"q": q}

🌟 理解它们：

• @app.get(“/items/”)：这就像设定一个魔法门，只允许 GET 请求进入。
• Query：确保传入的参数符合规则。
• q 是参数的名称，表示这个函数接受一个名为 q 的参数。
• str 是参数的类型,表示它应该是一个字符串。
• None 表示如果客户端没有提供 q 参数,则默认值为 None。
• max_length=50 表示 q 参数的最大长度为 50 个字符

🚀 步骤五： 定义 POST 请求处理函数。

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item}

🌟 理解它们：

• @app.post(“/items/”)：这是另一个魔法门，但这次只对 POST 请求开放。
• Item：接收的数据必须符合我们之前定义的魔法书（模型）。

🚀 步骤六： 运行我们的应用！

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

🌟 理解它们：

• uvicorn.run()：启动服务器，让我们的应用活起来！
• 注意哦：为什么这里使用 host="0.0.0.0" 而不是 localhost 或是127.0.0.1 ？
  • 使用 host="0.0.0.0" 可以让 Uvicorn 服务器监听所有可用的网络接口，这意味着你的 FastAPI 应用程序可以从任何设备上访问,而不仅仅是从本地机器上访问。

🌈 下面我们要学习一些额外魔法咯：

🌟 一：数据校验 - 给数据穿上安全的盔甲！🛡️

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

# 🌟 定义一个数据模型，确保数据的安全和完整
class Item(BaseModel):
    name: str  # 必需的
    description: str = None  # 可选的
    price: float  # 必需的
    tax: float = None  # 可选的

# 🎯 GET 请求的校验，确保 q 参数长度在规定范围内
@app.get("/items/")
async def read_items(q: str = Query(..., min_length=3, max_length=50)):
    return {"q": q}

# 🎯 POST 请求的数据校验，确保数据符合 Item 模型
@app.post("/items/")
async def create_item(item: Item):
    return {"item": item}

🌈 二：表单数据 - 接收用户传来的数据！📝

from fastapi import FastAPI, Form

app = FastAPI()

# 🎯 接受表单数据，从 POST 请求中读取 username 和 password
@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    return {"username": username}  # 返回用户名作为确认

🌈 三：声明模型 - 用 Pydantic 构建数据的完美框架！🛠️

from pydantic import BaseModel

# 🌟 定义一个用户模型，包括必填和可选字段
class User(BaseModel):
    username: str  # 用户名，必需的
    email: str  # 邮箱地址，必需的
    full_name: str = None  # 完整姓名，可选的

# 🎯 使用这个模型可以确保数据的格式正确
# user = User(username="john.doe", email="john.doe@example.com")
# print(user)  # 输出包含数据的 User 对象

🌈 魔法小贴士： 数据校验和模型声明是 FastAPI 的强大之处，它们确保了你的数据在传输过程中的安全性和一致性。就像魔法世界的守护者一样，它们会阻止任何不符合规则的数据入侵你的程序。

🌟 四：响应体 - 给客户端一个满意的答复！🎉

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 🌟 定义一个数据模型，用于响应体
class ItemResponse(BaseModel):
    name: str
    description: str = None
    price: float

# 🎯 创建商品，返回 ItemResponse 模型的实例
@app.post("/items/", response_model=ItemResponse)
async def create_item(name: str, description: str = None, price: float):
    return {"name": name, "description": description, "price": price}

🌈 魔法小贴士： 响应体是你和客户端沟通的桥梁，通过它你可以告诉客户端操作是否成功，或者返回一些有用的数据。

🌟 五：状态码 - 用 HTTP 语言说话！🗣️

from fastapi import FastAPI, status

app = FastAPI()

# 🎯 创建商品，使用 201 CREATED 状态码表示资源已创建
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(name: str, price: float):
    return {"name": name, "price": price}

# 🎯 错误处理，使用 400 BAD REQUEST 状态码表示客户端错误
@app.post("/items/error")
async def error_item(name: str):
    if name == "bad":
        raise HTTPException(status_code=400, detail="Bad item name")
    return {"name": name}

🌈 魔法小贴士： 状态码是 HTTP 协议的一部分，它们告诉客户端发生了什么。2xx 是成功，4xx 是客户端错误，5xx 是服务器错误。

🌟 六：模型转换 - 变身大师！🧙‍♂️

from pydantic import BaseModel

# 🌟 定义一个数据模型
class Item(BaseModel):
    name: str
    price: float

# 🎯 创建一个 Item 实例
item = Item(name="Magic Wand", price=999.99)

# 🎯 将 Item 实例转换为字典
item_dict = item.dict()

# 🎯 将 Item 实例转换为 JSON 字符串
item_json = item.json()

🌈 魔法小贴士： 模型转换非常方便，当你需要将模型数据转换为其他格式，如字典或 JSON 时，它会派上大用场。

🌟 七：文件上传 - 接收用户的大礼包！🎁

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

# 🎯 单文件上传
@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

# 🎯 多文件上传
@app.post("/uploadfiles/")
async def upload_files(files: list[UploadFile] = File(...)):
    return {"filenames": [file.filename for file in files]}

# for file in files ：它遍历 files 列表中的每个 UploadFile 对象,并将其赋值给 file 变量。
# file.filename ：这是访问每个 UploadFile 对象的 filename 属性。
# 遍历 files 列表中的每个 UploadFile 对象，然后获取它的 filename 属性

🌈 魔法小贴士： 文件上传是许多应用的重要功能，FastAPI 提供了简单的方式来处理文件，无论是单个还是多个。

🌟 八：Cookie - 记录用户的足迹！👣

from fastapi import FastAPI, Cookie
from fastapi.responses import JSONResponse

app = FastAPI()

# 🎯 设置 Cookie
@app.get("/setcookie/")
async def set_cookie():
    response = JSONResponse(content={"message": "Cookie set"})
    response.set_cookie(key="user", value="John Doe")
    return response

# 🎯 获取 Cookie
@app.get("/getcookie/")
async def get_cookie(user: str = Cookie(None)):
    return {"user": user}

# 除了 key 和 value 之外,set_cookie() 方法还支持以下选项:
max_age: Cookie 的最大生存时间(以秒为单位)。
expires: Cookie 的过期时间(以 UTC 时间格式表示)。
path: Cookie 的作用路径。
domain: Cookie 的作用域(域名)。
secure: 是否只能通过 HTTPS 协议访问 Cookie。
httponly: 是否只能通过 HTTP 协议访问 Cookie(不能被 JavaScript 访问)。
samesite: 跨站点 Cookie 的策略(Lax、Strict 或 None)。

• 要删除一个 Cookie,可以设置它的 max_age 为 0,并将 expires 设置为过去的时间:

@app.get("/deletecookie/")
async def delete_cookie():
    response = JSONResponse(content={"message": "Cookie deleted"})
    response.delete_cookie(key="user")
    return response

• 可以在一个响应中设置多个 Cookie:

@app.get("/setcookies/")
async def set_cookies():
    response = JSONResponse(content={"message": "Cookies set"})
    response.set_cookie(key="user", value="John Doe")
    response.set_cookie(key="session_id", value="abc123")
    return response

• 获取所有 Cookie：可以使用 request.cookies 字典获取所有 Cookie:

from fastapi import FastAPI, Request

@app.get("/getcookies/")

# 如果想要获取到request，必须冒号实例化一下才可以使用
async def get_cookies(request: Request):
    return {"cookies": request.cookies}

🌈 魔法小贴士： Cookie 是存储在客户端的小片段信息，常用于跟踪用户会话。在 FastAPI 中，设置和获取 Cookie 都很容易。

🌟 九：跨域 - 穿越防火墙的超级英雄！🦸‍♂️

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 🎯 设置 CORS 中间件，允许所有来源访问
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

🌈 魔法小贴士： 跨域资源共享（CORS）是解决前端和后端跨域请求问题的关键技术。在 FastAPI 中，你可以轻松配置 CORS 中间件来控制哪些来源可以访问你的 API。

🌟 十：OAuth2 - 登录认证的守护神！🔐

from fastapi import Depends, FastAPI
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()

# 🌟 OAuth2 的守护者
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 🎯 登录路由，模拟生成令牌
@app.post("/token")
async def login():
    return {"access_token": "magic-token", "token_type": "bearer"}

# 🎯 保护路由，需要 OAuth2 的令牌
@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

1. 获取当前用户:

   • 你可以编写一个 get_current_user 函数,根据访问令牌查找并返回当前登录的用户:

   async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
       # 根据令牌查找并返回当前用户
       return User(username="johndoe", email="johndoe@example.com", is_admin=True)
   

2. 权限控制:

   • 你可以根据用户的角色或权限,对特定的路由进行访问控制:

   from fastapi import Depends, FastAPI, HTTPException, status
   from fastapi.security import OAuth2PasswordBearer
   
   oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", scopes={"admin": "Access admin features", "user": "Access user features"})
   
   async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
       # 根据令牌查找并返回当前用户
       return User(username="johndoe", email="johndoe@example.com", is_admin=True)
   
   async def get_current_active_user(current_user: User = Depends(get_current_user)):
       if not current_user.is_active:
           raise HTTPException(status_code=400, detail="Inactive user")
       return current_user
   
   async def get_current_admin_user(current_user: User = Depends(get_current_active_user)):
       if not current_user.is_admin:
           raise HTTPException(status_code=403, detail="Not enough permissions")
       return current_user
   
   @app.get("/users/me", response_model=User)
   async def read_users_me(current_user: User = Depends(get_current_active_user)):
       return current_user
   
   @app.get("/admin/users", response_model=List[User])
   async def read_users(current_user: User = Depends(get_current_admin_user)):
       # 只有管理员才能访问这个路由
       return [
           User(username="johndoe", email="johndoe@example.com", is_admin=True),
           User(username="janesmith", email="janesmith@example.com", is_admin=False)
       ]
   

🌈 魔法小贴士： OAuth2 是一种流行的授权协议，它允许应用程序安全地获取对资源的访问权限。在 FastAPI 中，你可以轻松集成 OAuth2 来保护你的 API。

🌟 十一：JWT 认证 - 令牌的魔法！🪙

import jwt
from datetime import datetime, timedelta
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 🌟 JWT 的秘密配方
def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, "secret", algorithm="HS256")
    return encoded_jwt

# 🎯 登录并生成 JWT
@app.post("/token")
async def login():
    access_token = create_access_token(data={"sub": "john.doe@example.com"})
    return {"access_token": access_token, "token_type": "bearer"}

# 🎯 验证 JWT 的保护路由
@app.get("/users/me")
async def read_users_me(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, "secret", algorithms=["HS256"])
        email: str = payload.get("sub")
        if email is None:
            raise HTTPException(status_code=401, detail="Invalid token")
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"email": email}

• 为了提高安全性,你可以引入刷新令牌机制,允许用户在访问令牌过期时获取新的访问令牌:

from datetime import datetime, timedelta
from uuid import uuid4

class TokenData(BaseModel):
    email: str
    access_token: str
    refresh_token: str
    expires_at: datetime

@app.post("/token", response_model=TokenData)
async def login():
    access_token = create_access_token(data={"sub": "john.doe@example.com"})
    refresh_token = str(uuid4())
    expires_at = datetime.utcnow() + timedelta(minutes=15)
    return TokenData(
        email="john.doe@example.com",
        access_token=access_token,
        refresh_token=refresh_token,
        expires_at=expires_at
    )

@app.post("/token/refresh", response_model=TokenData)
async def refresh_token(refresh_token: str):
    # 根据刷新令牌生成新的访问令牌
    access_token = create_access_token(data={"sub": "john.doe@example.com"})
    expires_at = datetime.utcnow() + timedelta(minutes=15)
    return TokenData(
        email="john.doe@example.com",
        access_token=access_token,
        refresh_token=refresh_token,
        expires_at=expires_at
    )

🌈 魔法小贴士： JSON Web Tokens (JWTs) 是一种用于在各方之间安全地传输信息的方式。JWTs 可以在没有服务器的情况下被验证和解析，因此非常适合移动应用和单页应用。

🌟 十二：中间件 - 在请求的旅程中播撒魔法！🔮

from fastapi import FastAPI, Request

app = FastAPI()

# 🌟 中间件，记录请求的时间
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

🌈 魔法小贴士： 中间件是在请求到达处理函数之前和之后执行的代码。它可以用来添加额外的头信息，处理日志，甚至实现身份验证。

🌟 十三：依赖注入 - 为你的应用添加超能力！🦸‍♂️

from fastapi import FastAPI, Depends
from typing import Optional

app = FastAPI()

async def get_db_session():
    # 创建数据库会话
    session = create_session()
    try:
        yield session
    finally:
        session.close()

@app.get("/items/{item_id}")
async def read_item(item_id: int, session: Session = Depends(get_db_session)):
    # 使用 session 对象查询数据库
    item = session.query(Item).get(item_id)
    return item

在这个例子中,get_db_session 函数是一个依赖项,它返回一个数据库会话对象。在路由函数 read_item 中,我们使用 Depends(get_db_session) 来注入这个依赖项。

当客户端调用 /items/{item_id} 时,FastAPI 会自动调用 get_db_session 函数,并将返回的会话对象注入到 read_item 函数中。这样,我们就可以在路由函数中使用数据库会话对象来查询数据。

🌈 魔法小贴士： 依赖注入是一种设计模式，它使代码更具可测试性和可重用性。在 FastAPI 中，你可以使用依赖注入来管理复杂的依赖关系，例如数据库会话或外部服务。

🌟 十四：数据库使用 - 打开数据的宝库！🗝️

1. MongoDB:

   • 安装motor库,它是 MongoDB 的异步 Python 驱动程序:

     pip install motor
     

   • 创建一个 MongoDB 连接:

     from motor.motor_asyncio import AsyncIOMotorClient
     
     async def get_db():
         client = AsyncIOMotorClient("mongodb://localhost:27017")
         db = client.your_database_name
         yield db
         await client.close()
     

   • 在路由函数中使用 MongoDB 连接:

     from fastapi import FastAPI, Depends
     from .models import Item
     from .database import get_db
     
     app = FastAPI()
     
     @app.post("/items/")
     async def create_item(item: Item, db=Depends(get_db)):
         new_item = await db.items.insert_one(item.dict())
         return {"id": str(new_item.inserted_id)}
     

2. MySQL:

   • 安装aiomysql库,它是 MySQL 的异步 Python 驱动程序:

     pip install aiomysql
     

   • 创建一个 MySQL 连接:

     import aiomysql
     
     async def get_db():
         conn = await aiomysql.connect(host="localhost", user="your_username", password="your_password", db="your_database_name")
         cursor = await conn.cursor()
         try:
             yield cursor
         finally:
             await cursor.close()
             await conn.close()
     

   • 在路由函数中使用 MySQL 连接:

     from fastapi import FastAPI, Depends
     from .models import Item
     from .database import get_db
     
     app = FastAPI()
     
     @app.post("/items/")
     async def create_item(item: Item, cursor=Depends(get_db)):
         await cursor.execute("INSERT INTO items (name, description) VALUES (%s, %s)", (item.name, item.description))
         await cursor.connection.commit()
         return {"id": cursor.lastrowid}
     

3. SQLite:

   • 使用 Python 标准库中的 sqlite3 模块:

   • 创建一个 SQLite 连接:

     import sqlite3
     
     async def get_db():
         conn = sqlite3.connect("your_database.db")
         cursor = conn.cursor()
         try:
             yield cursor
         finally:
             cursor.close()
             conn.close()
     

   • 在路由函数中使用 SQLite 连接:

     from fastapi import FastAPI, Depends
     from .models import Item
     from .database import get_db
     
     app = FastAPI()
     
     @app.post("/items/")
     async def create_item(item: Item, cursor=Depends(get_db)):
         cursor.execute("INSERT INTO items (name, description) VALUES (?, ?)", (item.name, item.description))
         cursor.connection.commit()
         return {"id": cursor.lastrowid}
     

🌈 魔法小贴士： FastAPI 可以与多种数据库系统一起工作，包括 MySQL、MongoDB、SQLite 等。你可以选择最适合你项目需求的数据库。

🌟 十五：项目拆分 - 建造你的城堡！🏰

your_project/
├── app/ # 这是你的 FastAPI 应用程序的主目录
│   ├── api/ # 这个目录包含你的 API 路由
│   │   ├── v1/
│   │   │   ├── endpoints/
│   │   │   │   ├── items.py
│   │   │   │   └── users.py
│   │   │   └── __init__.py
│   │   └── __init__.py
│   ├── core/ # 这个目录包含应用程序的核心配置和功能,如配置文件
│   │   ├── config.py
│   │   ├── security.py
│   │   └── __init__.py
│   ├── db/ # 这个目录包含数据库相关的代码,如 ORM 模型定义、数据库会话管理等
│   │   ├── base.py
│   │   ├── session.py
│   │   └── __init__.py
│   ├── models/ # 这个目录包含 Pydantic 模型,用于请求和响应数据的序列化和验证
│   │   ├── items.py
│   │   ├── users.py
│   │   └── __init__.py
│   ├── schemas/
│   │   ├── items.py
│   │   ├── users.py
│   │   └── __init__.py
│   ├── tests/
│   │   ├── api/
│   │   │   └── test_items.py
│   │   └── __init__.py
│   └── __init__.py
├── requirements.txt
├── main.py
└── README.md

🌈 魔法小贴士： 将项目拆分成多个文件和模块可以帮助你保持代码的组织性和可维护性。每个文件应该负责一个特定的任务，如模型定义、模式验证、业务逻辑等。

🌟 十六：异步编程 - 加速你的魔法！🚀

import asyncio
from fastapi import FastAPI

app = FastAPI()

# 🎯 异步处理请求
@app.get("/")
async def read_root():
    await asyncio.sleep(1)
    return {"message": "Hello World!"}

🌈 魔法小贴士： 异步编程允许你的应用程序在等待 I/O 操作完成的同时继续执行其他任务，从而提高了效率和响应速度。

🌟 十七：WebSocket - 实时通信的魔法！⚡

from fastapi import FastAPI, WebSocket

app = FastAPI()

# 🎯 WebSocket 路由
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

🌈 魔法小贴士： WebSocket 提供了一种在客户端和服务器之间进行全双工通信的方法。它非常适合实时应用，如聊天应用或实时数据分析。

🌟 十八：事件处理 - 见证魔法的诞生和消逝！💫

from fastapi import FastAPI

app = FastAPI()

# 🎯 启动时
@app.on_event("startup")
async def startup_event():
    print("Magical startup!")

# 🎯 关闭时
@app.on_event("shutdown")
async def shutdown_event():
    print("Magical shutdown!")

🌈 魔法小贴士： FastAPI 允许你定义在应用启动和关闭时执行的事件处理器。这可以用来初始化资源或执行清理操作。