引言
FastAPI是一个现代、快速且高效的Web框架,专为构建API和异步Web应用而设计。它结合了Python的类型提示和异步编程的特性,使得开发者能够快速开发出高性能、易维护的Web应用程序。本文将深入探讨FastAPI的实战技巧和案例解析,帮助开发者更好地理解和应用FastAPI。
FastAPI基础
安装FastAPI
要开始使用FastAPI,首先需要安装FastAPI及其依赖库。可以通过以下命令进行安装:
pip install fastapi uvicorn
创建FastAPI应用
创建一个FastAPI应用非常简单,只需导入FastAPI类并创建其实例:
from fastapi import FastAPI
app = FastAPI()
定义路由
在FastAPI中,可以使用装饰器来定义路由。以下是一个简单的例子:
@app.get("/")
async def root():
return {"message": "Hello World"}
在这个例子中,@app.get("/")
装饰器定义了一个处理GET请求的路由,当访问根路径时,会调用 root()
函数。
实战技巧
类型安全
FastAPI利用Python的类型提示来实现类型安全。在定义函数参数时,可以指定参数的类型,FastAPI会自动进行数据验证:
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
异步处理
FastAPI支持异步处理,这意味着它可以同时处理多个请求,而不会阻塞服务器:
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
依赖注入
FastAPI的依赖注入系统允许你将依赖项注入到路由处理函数中,从而解耦你的代码:
from fastapi import Depends, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas
def get_db():
db = Session()
try:
yield db
finally:
db.close()
@app.get("/items/{item_id}", response_model=schemas.Item)
def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(models.Item).filter(models.Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
案例解析
案例一:Todo应用
以下是一个简单的Todo应用示例,使用FastAPI和SQLite数据库:
from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import models, schemas
app = FastAPI()
def get_db():
db = Session()
try:
yield db
finally:
db.close()
@app.post("/todo/")
def create_todo(todo: schemas.Todo, db: Session = Depends(get_db)):
db_todo = models.Todo(name=todo.name)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
@app.get("/todo/")
def read_todos(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
todos = db.query(models.Todo).offset(skip).limit(limit).all()
return todos
案例二:用户认证
以下是一个用户认证的例子,使用FastAPI和Passlib库:
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from . import models, schemas
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def authenticate_user(db: Session, username: str, password: str):
user = db.query(models.User).filter(models.User.username == username).first()
if not user or not verify_password(password, user.hashed_password):
return False
return user
@app.post("/token")
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(db, form_data.username, 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"}
@app.get("/users/me")
async def read_users_me(current_user: models.User = Depends(authenticate_user)):
return current_user
总结
FastAPI是一个功能强大且易于使用的Web框架,它可以帮助开发者快速构建高性能的Web服务。通过本文的实战技巧和案例解析,相信读者能够更好地理解和应用FastAPI。