打造高效API：FastAPI认证与授权全方位指南

在开发现代Web应用时，API（应用程序编程接口）已成为不可或缺的一部分。FastAPI是一个现代、快速（高性能）的Web框架，用于构建API，其简洁性和强大的功能使其成为开发者的热门选择。本文将深入探讨如何在FastAPI中实现高效的认证与授权。

认证：确保用户身份

认证是确保用户身份的过程。在FastAPI中，我们可以使用多种方法来实现认证，如JWT（JSON Web Tokens）、OAuth2等。

JWT认证

JWT是一种开放标准（RFC 7519），用于在各方之间安全地传输信息作为JSON对象。下面是如何在FastAPI中实现JWT认证的示例：

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

app = FastAPI()

SECRET_KEY = "your_secret_key"

fake_db = {
    "john_doe": {"username": "john_doe", "password": "secret"}
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class TokenData(BaseModel):
    username: str

@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=30)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "bearer"}

def create_access_token(data: dict, expires_delta: timedelta | None = 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_KEY, algorithm="HS256")
    return encoded_jwt

@app.get("/users/me", response_model=User)
def read_users_me(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    return verify_token(token, credentials_exception)

def verify_token(token: str, credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return username
    except jwt.ExpiredSignatureError:
        raise credentials_exception
    except jwt.exceptions.InvalidTokenError:
        raise credentials_exception

OAuth2认证

OAuth2是一种授权框架，允许第三方应用代表用户访问受保护的资源。在FastAPI中，我们可以使用python-oauthlib库来实现OAuth2认证。

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

app = FastAPI()

SECRET_KEY = "your_secret_key"

fake_db = {
    "john_doe": {"username": "john_doe", "password": "secret"}
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class TokenData(BaseModel):
    username: str

@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=30)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "bearer"}

def create_access_token(data: dict, expires_delta: timedelta | None = 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_KEY, algorithm="HS256")
    return encoded_jwt

@app.get("/users/me", response_model=User)
def read_users_me(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    return verify_token(token, credentials_exception)

def verify_token(token: str, credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return username
    except jwt.ExpiredSignatureError:
        raise credentials_exception
    except jwt.exceptions.InvalidTokenError:
        raise credentials_exception

授权：控制对资源的访问

授权是控制对受保护资源的访问的过程。在FastAPI中，我们可以使用角色基础访问控制（RBAC）来实现授权。

角色基础访问控制（RBAC）

RBAC是一种基于角色的访问控制机制，用于限制对特定资源的访问。以下是如何在FastAPI中实现RBAC的示例：

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

app = FastAPI()

SECRET_KEY = "your_secret_key"

fake_db = {
    "john_doe": {"username": "john_doe", "password": "secret", "role": "admin"},
    "jane_doe": {"username": "jane_doe", "password": "secret", "role": "user"}
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

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

class TokenData(BaseModel):
    username: str

@app.post("/token", response_model=Token)
def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_db.get(form_data.username)
    if not user or user["password"] != form_data.password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=30)
    access_token = create_access_token(data={"sub": user["username"]}, expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "bearer"}

def create_access_token(data: dict, expires_delta: timedelta | None = 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_KEY, algorithm="HS256")
    return encoded_jwt

@app.get("/users/me", response_model=User)
def read_users_me(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    return verify_token(token, credentials_exception)

def verify_token(token: str, credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        return username
    except jwt.ExpiredSignatureError:
        raise credentials_exception
    except jwt.exceptions.InvalidTokenError:
        raise credentials_exception

def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    return verify_token(token, credentials_exception)

def get_current_active_user(current_user: str = Depends(get_current_user)):
    user = fake_db.get(current_user)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Active user not found",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

@app.get("/admin")
def read_admin_data(current_user: dict = Depends(get_current_active_user)):
    if current_user["role"] != "admin":
        raise HTTPException(status_code=403, detail="Not enough privileges")
    return {"message": "Admin data"}

在上述示例中，我们为john_doe和jane_doe创建了两个用户，其中john_doe是管理员，而jane_doe是普通用户。当访问/admin端点时，只有管理员才能访问。

总结

在FastAPI中实现高效的认证与授权是构建安全、可靠的API的关键。通过使用JWT、OAuth2和RBAC等技术，我们可以确保只有授权的用户才能访问受保护的资源。希望本文能帮助您在FastAPI中实现高效的认证与授权。

正文

打造高效API：FastAPI认证与授权全方位指南

认证：确保用户身份

JWT认证

OAuth2认证

授权：控制对资源的访问

角色基础访问控制（RBAC）

总结

相关阅读

Android开发：ASM与Javassist利器大揭秘，实战技巧助力高效编程

掌握MySQL监控，保障数据库稳定运行：五大关键配置技巧解析

Rocky Linux系统软件包管理：高效安装、更新与维护指南

团队代码重构：告别混乱，迈向高效协作的编程之道

Docker实战指南：从入门到精通，构建高效容器化应用

掌握CSS文字宽度限制，轻松实现网页排版美观与实用

企业级开发如何高效运用XPath：实战技巧与案例分析

如何应对社会风险：揭秘全球最佳防范策略与实战案例

HTML5网页布局技巧全解析：打造响应式设计，轻松实现高效排版

轻松上手Oracle Linux Kubernetes集群，五大实战技巧助力企业云原生转型