正文

新手必看：FastAPI框架入门教程与实战案例解析

/0 浏览量
0718

FastAPI简介

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 API。它基于 Python 3.6+，使用标准 Python 类型提示功能。FastAPI 非常易于上手，且提供了丰富的功能。以下是关于 FastAPI 的一些基本概念和特点。

FastAPI特点

  1. 快速性能：FastAPI 是基于 Starlette 和 Pydantic 构建，提供了出色的性能。
  2. 自动文档：FastAPI 提供了一个交互式 API 文档，使用 Swagger UI。
  3. 数据验证：通过 Pydantic 对请求和响应进行验证。
  4. 异步支持：支持异步处理，适用于高并发场景。
  5. 易于扩展：FastAPI 可以轻松扩展，支持自定义中间件和依赖注入。

FastAPI入门教程

安装FastAPI

首先，我们需要安装 FastAPI 和 Uvicorn，这是一个 ASGI 服务器。

pip install fastapi uvicorn

创建FastAPI应用

创建一个名为 main.py 的 Python 文件，并添加以下代码：

from fastapi import FastAPI

app = FastAPI()

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

这段代码创建了一个 FastAPI 应用，并定义了一个 / 路径的路由。当访问该路径时，会返回 “Hello, world!“。

运行FastAPI应用

在终端中运行以下命令来启动服务器：

uvicorn main:app --reload

然后，在浏览器中访问 http://127.0.0.1:8000/，你会看到 “Hello, world!“。

使用Pydantic进行数据验证

假设我们想要对 /items/ 路径进行数据验证。首先，我们需要定义一个 Pydantic 模型：

from pydantic import BaseModel

class Item(BaseModel):
    id: int
    name: str

然后，在 main.py 中修改 /items/ 路由：

from fastapi import FastAPI, HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int, item: Item = Depends()):
    return item

现在，当访问 /items/1 时，FastAPI 会要求传入一个名为 item 的对象，它将根据 Item 模型进行验证。如果验证失败，FastAPI 会返回一个 422 状态码。

异步处理

FastAPI 支持异步处理，这使得它非常适合高并发场景。下面是一个异步路由的例子：

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # 模拟异步操作
    await asyncio.sleep(1)
    return {"item_id": item_id}

在这个例子中，我们使用 asyncio.sleep(1) 来模拟一个异步操作。当访问 /items/1 时，你会在 1 秒后收到响应。

FastAPI实战案例解析

实战案例一：用户认证

在这个案例中，我们将使用 OAuth2 密码（密码流）进行用户认证。

  1. 安装 python-multipart 和 httpx：
pip install python-multipart httpx
  1. 创建一个名为 auth.py 的文件，并添加以下代码：
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(username=form_data.username, password=form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

def authenticate_user(username: str, password: str):
    # 这里应该有数据库查询和用户认证逻辑
    pass

def create_access_token(data: dict):
    # 这里应该有 JWT 生成逻辑
    pass
  1. 修改 main.py，添加一个需要认证的路由：
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

@app.get("/users/me", response_model=TokenData)
async def read_users_me(token: str = Depends(oauth2_scheme)):
    # 这里应该有用户信息查询逻辑
    return {"username": "test"}

现在，当你访问 /users/me 路由时，你需要提供一个有效的 access token。

实战案例二：数据库操作

在这个案例中，我们将使用 SQLite 数据库进行用户管理。

  1. 安装 databases 包：
pip install databases
  1. 创建一个名为 database.py 的文件，并添加以下代码：
from databases import Database
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

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

database = Database(DATABASE_URL)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
  1. 修改 main.py，添加用户表和数据库操作：
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

from .database import get_db
from . import models, schemas

app = FastAPI()

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

@app.post("/users/")
async def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    db_user = models.User(name=user.name, email=user.email, hashed_password=user.hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

在这个例子中，我们定义了一个 /users/ 路由，用于创建新用户。我们使用 SQLAlchemy 进行数据库操作。

通过以上教程和案例，相信你已经对 FastAPI 框架有了初步的了解。FastAPI 是一个非常强大且易于使用的 Web 框架，适合构建高性能的 API。祝你在 FastAPI 的道路上越走越远！

-- 展开阅读全文 --