当前位置： 首页 > article >正文

后端开发入门

article 2024/11/26 9:51:16

后端开发最佳实践详解

1. 引言

后端开发不仅仅是编写功能代码，还涉及到如何构建稳定、可靠且高效的系统。掌握后端开发的最佳实践，可以帮助您避免常见的错误，提高代码质量，确保应用的可维护性和扩展性。以下内容将详细讲解这些关键点，并通过示例帮助您理解和应用。

2. 错误处理

错误处理是后端开发中不可或缺的一部分。合理的错误处理不仅可以提升用户体验，还能帮助开发者快速定位和修复问题。

2.1 使用 try-except 进行异常捕获

在 Python 中，try-except 语句用于捕获和处理可能发生的异常，防止程序因未处理的错误而崩溃。

基本语法：

try:
    # 可能发生异常的代码
    result = 10 / 0
except ZeroDivisionError:
    # 处理特定异常
    print("不能除以零")
except Exception as e:
    # 处理其他异常
    print(f"发生错误: {e}")
finally:
    # 无论是否发生异常，都会执行的代码
    print("执行结束")

输出：

不能除以零
执行结束

解释：

  • try 块：包含可能发生异常的代码。
  • except 块：捕获并处理特定的异常。
  • Exception：捕获所有其他未被捕获的异常。
  • finally 块：无论是否发生异常，都会执行的代码，常用于资源释放（如关闭文件、数据库连接）。

示例：安全除法函数

def safe_divide(a, b):
    try:
        return a / b
    except ZeroDivisionError:
        return "错误：除数不能为零"
    except TypeError:
        return "错误：输入必须是数字"

使用：

print(safe_divide(10, 2))  # 输出：5.0
print(safe_divide(10, 0))  # 输出：错误：除数不能为零
print(safe_divide(10, 'a'))  # 输出：错误：输入必须是数字

2.2 在 FastAPI 中处理异常

FastAPI 提供了内置的异常处理机制，使得在 API 中捕获和处理异常更加方便。

示例：处理数据库连接错误

from fastapi import FastAPI, HTTPException
import pymysql

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        connection = pymysql.connect(
            host='localhost',
            user='root',
            password='your_password',
            database='mydatabase'
        )
        with connection.cursor() as cursor:
            sql = "SELECT * FROM users WHERE id = %s"
            cursor.execute(sql, (user_id,))
            result = cursor.fetchone()
            if not result:
                raise HTTPException(status_code=404, detail="User not found")
            return result
    except pymysql.MySQLError as e:
        # 捕获数据库错误
        raise HTTPException(status_code=500, detail="数据库连接错误")
    finally:
        connection.close()

解释：

  • 使用 try-except 捕获可能的数据库连接错误。
  • 如果用户不存在，抛出 HTTPException，返回 404 错误。
  • 如果数据库连接失败，抛出 HTTPException，返回 500 错误。

2.3 自定义异常

有时，内置的异常类不足以满足需求，可以定义自定义异常并创建相应的处理器。

步骤：

  1. 定义自定义异常类
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name
  1. 注册异常处理器
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oh no! {exc.name} did something wrong."},
    )
  1. 在路径操作中抛出自定义异常
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

使用：

访问 http://127.0.0.1:8000/unicorns/yolo 时，会返回：

{
  "message": "Oh no! yolo did something wrong."
}

3. 日志记录

日志记录是监控应用运行状况、调试和审计的重要工具。通过日志，开发者可以了解应用的行为、发现问题并追踪错误。

3.1 为什么需要日志记录

  • 调试：在开发和测试过程中，日志可以帮助定位和解决问题。
  • 监控：在生产环境中，日志用于监控应用的健康状态和性能。
  • 审计：记录用户行为和系统事件，满足合规性要求。
  • 错误追踪：记录异常和错误信息，帮助快速响应和修复问题。

3.2 配置 Python 的 logging 模块

Python 内置的 logging 模块功能强大，可以满足大多数日志记录需求。

基本配置：

import logging

# 配置日志记录器
logging.basicConfig(
    level=logging.INFO,  # 设置日志级别
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',  # 日志格式
    handlers=[
        logging.FileHandler("app.log"),  # 写入日志文件
        logging.StreamHandler()  # 输出到控制台
    ]
)

# 获取日志记录器
logger = logging.getLogger(__name__)

解释：

  • level：设置最低日志级别。常见级别依次为 DEBUG < INFO < WARNING < ERROR < CRITICAL。

  • format

    ：定义日志输出格式，常用的占位符：

    • %(asctime)s：时间戳
    • %(name)s：记录器名称
    • %(levelname)s：日志级别
    • %(message)s：日志消息

  • handlers

    ：定义日志处理器，决定日志的输出位置。常见的处理器有：

    • FileHandler：将日志写入文件
    • StreamHandler：将日志输出到控制台
    • RotatingFileHandler：支持日志文件轮转，防止日志文件过大

示例：配置日志记录器

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[
        logging.FileHandler("debug.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("my_app")

3.3 在 FastAPI 中集成日志

将日志记录集成到 FastAPI 应用中，可以记录请求、响应、错误等信息。

步骤：

  1. 配置日志记录器
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("my_fastapi_app")
  1. 记录请求信息

使用中间件记录每个请求的信息。

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    response = await call_next(request)
    logger.info(f"Response status: {response.status_code}")
    return response
  1. 记录异常信息

在异常处理器中记录详细的错误信息。

from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.error(f"HTTPException: {exc.detail}")
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

3.4 日志级别与格式

日志级别：

  • DEBUG：详细的信息，通常只在开发时使用。
  • INFO：确认程序按预期运行的信息。
  • WARNING：表明某些意外情况或潜在问题，但程序仍然可以继续运行。
  • ERROR：由于更严重的问题，程序无法执行某些功能。
  • CRITICAL：非常严重的错误，导致程序终止。

调整日志级别：

根据不同的环境（开发、测试、生产），调整日志级别以控制日志输出的详细程度。

import os

log_level = os.getenv("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=log_level,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

日志格式示例：

format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'

输出示例：

2024-04-27 10:00:00 [INFO] my_fastapi_app: Request: GET http://127.0.0.1:8000/items/1
2024-04-27 10:00:00 [INFO] my_fastapi_app: Response status: 200
2024-04-27 10:00:01 [ERROR] my_fastapi_app: HTTPException: Item not found

3.5 示例：记录请求和错误

完整示例：

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
import logging

# 配置日志记录器
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
    handlers=[
        logging.FileHandler("app.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("my_fastapi_app")

app = FastAPI()

# 中间件：记录请求和响应
@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url}")
    try:
        response = await call_next(request)
        logger.info(f"Response status: {response.status_code}")
        return response
    except Exception as e:
        logger.error(f"Unhandled exception: {e}")
        raise e

# 自定义异常处理器
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    logger.error(f"HTTPException: {exc.detail}")
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

# 示例路径操作
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=400, detail="Invalid item ID")
    return {"item_id": item_id}

运行并测试：

  1. 启动应用

    uvicorn main:app --reload

  2. 访问有效路径

    访问 http://127.0.0.1:8000/items/1，日志文件 app.log 中将记录：

    2024-04-27 10:00:00 [INFO] my_fastapi_app: Request: GET http://127.0.0.1:8000/items/1
2024-04-27 10:00:00 [INFO] my_fastapi_app: Response status: 200

  3. 访问无效路径

    访问 http://127.0.0.1:8000/items/0，日志文件 app.log 中将记录：

    2024-04-27 10:00:01 [INFO] my_fastapi_app: Request: GET http://127.0.0.1:8000/items/0
2024-04-27 10:00:01 [ERROR] my_fastapi_app: HTTPException: Invalid item ID

4. 数据库事务管理

事务（Transaction）是数据库操作的基本单位，确保一系列操作要么全部成功，要么全部失败，保证数据的一致性和完整性。

4.1 什么是事务

  • 原子性（Atomicity）：事务中的所有操作要么全部完成，要么全部不执行。
  • 一致性（Consistency）：事务执行前后，数据库都处于一致的状态。
  • 隔离性（Isolation）：事务的执行不受其他事务的干扰。
  • 持久性（Durability）：事务一旦提交，结果是永久性的，即使系统崩溃也不会丢失。

这四个特性通常被称为 ACID 属性。

4.2 在 SQLAlchemy 中管理事务

SQLAlchemy 提供了强大的事务管理功能，确保数据库操作的原子性和一致性。

使用上下文管理器管理事务：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Base, User

engine = create_engine('mysql+pymysql://user:password@localhost/mydatabase')
SessionLocal = sessionmaker(bind=engine)

def create_user(name: str, email: str, age: int):
    session = SessionLocal()
    try:
        new_user = User(name=name, email=email, age=age)
        session.add(new_user)
        session.commit()  # 提交事务
        session.refresh(new_user)  # 刷新实例，获取生成的主键等信息
        return new_user
    except Exception as e:
        session.rollback()  # 回滚事务
        print(f"Error occurred: {e}")
        return None
    finally:
        session.close()  # 关闭会话

解释：

  • 创建会话：使用 SessionLocal() 创建一个会话实例。
  • 添加对象：通过 session.add() 添加新对象到会话。
  • 提交事务：使用 session.commit() 提交事务，将更改保存到数据库。
  • 刷新对象：使用 session.refresh() 更新对象实例，以获取数据库生成的数据（如主键）。
  • 回滚事务：在发生异常时，使用 session.rollback() 回滚事务，撤销未完成的操作。
  • 关闭会话：无论是否发生异常，都会执行 session.close() 关闭会话，释放资源。

4.3 示例：安全地处理数据库操作

示例：创建用户并处理事务

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Base, User
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# 配置数据库引擎
engine = create_engine('mysql+pymysql://user:password@localhost/mydatabase')
SessionLocal = sessionmaker(bind=engine)

# 创建 FastAPI 应用
app = FastAPI()

# Pydantic 模型
class UserCreate(BaseModel):
    name: str
    email: str
    age: int

@app.post("/users/", response_model=UserCreate)
async def create_user(user: UserCreate):
    session = SessionLocal()
    try:
        new_user = User(name=user.name, email=user.email, age=user.age)
        session.add(new_user)
        session.commit()
        session.refresh(new_user)
        return new_user
    except Exception as e:
        session.rollback()
        raise HTTPException(status_code=500, detail="数据库操作失败")
    finally:
        session.close()

解释：

  • 路径操作函数：create_user 接收用户数据，尝试将其添加到数据库。
  • 事务管理：通过 try-except-finally 结构，确保在数据库操作中发生异常时回滚事务，并在最后关闭会话。
  • 错误处理：在发生异常时，抛出 HTTPException，返回 500 错误给客户端。

5. 输入验证与数据清洗

确保用户输入的数据有效且安全，是后端开发的重要环节。FastAPI 利用 Pydantic 提供了强大的数据验证和解析功能。

5.1 使用 Pydantic 进行输入验证

Pydantic 是一个数据解析和验证库，FastAPI 利用其强大的功能来验证请求数据。

定义 Pydantic 模型：

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=120)

解释：

  • BaseModel：所有 Pydantic 模型都继承自 BaseModel。
  • 字段定义：
    • name: str：必填字段，类型为字符串。
    • email: EmailStr：必填字段，类型为有效的电子邮件地址。
    • age: Optional[int]：可选字段，类型为整数，且在 0 到 120 之间。
  • Field：用于添加更多的字段约束，如最小长度、最大值等。

使用 Pydantic 模型进行输入验证：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr, Field
from typing import Optional

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=120)

@app.post("/users/", response_model=UserCreate)
async def create_user(user: UserCreate):
    # 此处可以添加数据库操作
    return user

示例：发送无效数据

发送以下请求体：

{
    "name": "A",
    "email": "invalid-email",
    "age": -5
}

响应：

FastAPI 会自动返回 422 错误，指出验证失败的字段和原因。

{
  "detail": [
    {
      "loc": ["body", "name"],
      "msg": "ensure this value has at least 2 characters",
      "type": "value_error.any_str.min_length",
      "ctx": {"limit_value": 2}
    },
    {
      "loc": ["body", "email"],
      "msg": "value is not a valid email address",
      "type": "value_error.email"
    },
    {
      "loc": ["body", "age"],
      "msg": "ensure this value is greater than or equal to 0",
      "type": "value_error.number.not_ge",
      "ctx": {"limit_value": 0}
    }
  ]
}

5.2 避免 SQL 注入

SQL 注入是一种常见的安全漏洞，攻击者通过构造恶意输入来执行未授权的 SQL 语句。使用 ORM 或参数化查询可以有效防止 SQL 注入。

使用 ORM 防止 SQL 注入

ORM 自动处理参数化查询，避免直接拼接 SQL 语句。

# 安全的 ORM 查询
user = session.query(User).filter(User.name == 'Alice').first()

使用参数化查询防止 SQL 注入

即使使用原生 SQL，参数化查询也能有效防止 SQL 注入。

with connection.cursor() as cursor:
    sql = "SELECT * FROM users WHERE name = %s"
    cursor.execute(sql, ('Alice',))  # 使用参数化查询
    result = cursor.fetchone()

避免拼接字符串构建 SQL 语句

不安全的示例：

def get_user(name):
    sql = f"SELECT * FROM users WHERE name = '{name}'"
    cursor.execute(sql)
    return cursor.fetchone()

攻击者可以通过传入 name = "Alice'; DROP TABLE users; --" 来执行恶意 SQL 语句。

安全的示例：

def get_user(name):
    sql = "SELECT * FROM users WHERE name = %s"
    cursor.execute(sql, (name,))  # 使用参数化查询
    return cursor.fetchone()

5.3 示例：验证用户输入

定义 Pydantic 模型

from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=120)

路径操作函数

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from models import User
from database import SessionLocal, engine
from pydantic import BaseModel
from typing import List

app = FastAPI()

# 初始化数据库
Base.metadata.create_all(bind=engine)

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=UserCreate)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 检查是否已存在用户
    existing_user = db.query(User).filter(User.email == user.email).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # 创建新用户
    new_user = User(name=user.name, email=user.email, age=user.age)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

解释：

  • Pydantic 模型：定义了用户创建所需的字段及其验证规则。
  • 依赖注入：通过 Depends(get_db) 获取数据库会话。
  • 验证逻辑：
    • 检查电子邮件是否已被注册。
    • 如果已注册，抛出 400 错误。
    • 否则，创建新用户并保存到数据库。

测试：

发送以下请求体：

{
    "name": "Alice",
    "email": "alice@example.com",
    "age": 30
}

  • 成功响应：

    {
  "name": "Alice",
  "email": "alice@example.com",
  "age": 30
}

  • 重复电子邮件响应：

    {
  "detail": "Email already registered"
}

6. 安全性最佳实践

后端应用的安全性直接关系到数据的保密性和完整性。以下是一些关键的安全性最佳实践。

6.1 密码存储与哈希

永远不要以明文形式存储密码。应使用密码哈希函数将密码转换为不可逆的哈希值，并存储哈希值。

使用 passlib 进行密码哈希

  1. 安装 passlib

    pip install passlib[bcrypt]

  2. 配置 passlib

    from passlib.context import CryptContext

# 配置密码上下文，使用 bcrypt 算法
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

  3. 在用户注册中哈希密码

    @app.post("/register/")
async def register(user: UserCreate, db: Session = Depends(get_db)):
    # 检查是否已存在用户
    existing_user = db.query(User).filter(User.email == user.email).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    # 哈希密码
    hashed_password = hash_password(user.password)
    
    # 创建新用户
    new_user = User(name=user.name, email=user.email, hashed_password=hashed_password, age=user.age)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

  4. 在登录时验证密码

    @app.post("/login/")
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    
    # 生成 JWT 令牌（后续章节详细介绍）
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

解释：

  • 哈希密码：使用 passlib 将密码转换为哈希值后存储在数据库中。
  • 验证密码：在用户登录时，使用 passlib 验证输入的密码与存储的哈希值是否匹配。

6.2 认证与授权

认证（Authentication）：验证用户的身份，例如通过用户名和密码登录。

授权（Authorization）：确定用户是否有权限访问特定资源或执行特定操作。

使用 JWT 进行认证与授权

  1. 安装依赖

    pip install python-jose[cryptography] passlib[bcrypt]

  2. 配置安全工具

    from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

# 密码哈希配置
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 配置
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# JWT 配置
SECRET_KEY = "your_secret_key"  # 应该使用环境变量存储
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def authenticate_user(db, email: str, password: str):
    user = db.query(User).filter(User.email == email).first()
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user

  3. 创建令牌生成路径操作

    @app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    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.email})
    return {"access_token": access_token, "token_type": "bearer"}

  4. 创建获取当前用户的依赖项

    from pydantic import BaseModel

class TokenData(BaseModel):
    email: Optional[str] = None

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
        token_data = TokenData(email=email)
    except JWTError:
        raise credentials_exception
    
    user = db.query(User).filter(User.email == token_data.email).first()
    if user is None:
        raise credentials_exception
    return user

  5. 保护路径操作，要求用户登录

    @app.get("/users/me/", response_model=UserCreate)
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

解释：

  • 创建令牌：在用户登录时，生成 JWT 令牌，包含用户的电子邮件（sub）。
  • 验证令牌：在需要保护的路径操作中，验证令牌的有效性，提取用户信息。
  • 保护路径：使用 Depends(get_current_user)，确保只有经过认证的用户才能访问。

6.3 防止跨站脚本攻击 (XSS) 和跨站请求伪造 (CSRF)

跨站脚本攻击 (XSS) 和 跨站请求伪造 (CSRF) 是常见的 Web 安全漏洞。

防止 XSS：

  • 输入验证与清洗：验证和清洗用户输入，防止恶意脚本注入。
  • 内容安全策略 (CSP)：在响应头中设置 CSP，限制浏览器加载的资源来源。
  • 转义输出：在渲染用户输入时，确保正确转义，防止脚本执行。

防止 CSRF：

  • 使用 CSRF 令牌：为敏感操作添加 CSRF 令牌，验证请求来源。
  • SameSite Cookie：设置 Cookie 的 SameSite 属性，限制跨站请求携带 Cookie。
  • 验证 Referer 和 Origin 头：确保请求来源的合法性。

示例：设置 CSP 头

@app.middleware("http")
async def add_csp_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

6.4 HTTPS 与安全传输

使用 HTTPS 确保客户端与服务器之间的通信是加密的，保护数据的机密性和完整性。

在部署时配置 HTTPS：

  • 使用反向代理服务器：如 Nginx、Apache，配置 SSL/TLS 证书。

  • 获取 SSL 证书：可以使用 Let’s Encrypt 获取免费的 SSL 证书。

  • 配置反向代理服务器：

    Nginx 配置示例：

    server {
    listen 80;
    server_name your_domain.com;
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    server_name your_domain.com;

    ssl_certificate /path/to/fullchain.pem;
    ssl_certificate_key /path/to/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

7. 代码结构与组织

良好的代码结构和组织有助于提高代码的可读性、可维护性和可扩展性。

7.1 模块化设计

将代码拆分为独立的模块，每个模块负责特定的功能，避免代码冗长和耦合。

示例：组织 FastAPI 项目

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── database.py
│   ├── crud.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   └── core/
│       ├── __init__.py
│       ├── config.py
│       └── security.py
├── tests/
│   ├── __init__.py
│   └── test_users.py
├── requirements.txt
└── README.md

解释：

  • app/

    ：主应用目录。

    • main.py：应用入口，定义 FastAPI 实例和包含的路由。
    • models.py：定义 ORM 模型。
    • schemas.py：定义 Pydantic 模型，用于请求和响应。
    • database.py：配置数据库连接和会话。
    • crud.py：定义 CRUD（创建、读取、更新、删除）操作。
    • api/：包含不同的 API 路由模块，如 users.py、items.py。
    • core/：包含核心配置和安全相关代码。

  • tests/：包含测试代码。

  • requirements.txt：列出项目依赖。

  • README.md：项目说明文件。

7.2 分层架构

采用分层架构将应用逻辑分为不同的层，每一层专注于特定的职责。

常见的分层：

  1. 表示层（Presentation Layer）：处理 HTTP 请求和响应，定义 API 路由。
  2. 业务逻辑层（Business Logic Layer）：处理具体的业务规则和逻辑。
  3. 数据访问层（Data Access Layer）：与数据库进行交互，执行 CRUD 操作。
  4. 模型层（Model Layer）：定义数据结构和关系。

示例：分层架构

# app/api/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.database import get_db

router = APIRouter()

@router.post("/users/", response_model=schemas.UserCreate)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

# app/crud.py
from sqlalchemy.orm import Session
from app import models, schemas

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

def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(name=user.name, email=user.email, age=user.age)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

解释：

  • api/users.py：定义用户相关的 API 路由，调用 crud 层的函数进行数据库操作。
  • crud.py：定义具体的数据库操作函数，提供与数据访问层的接口。

7.3 示例：组织 FastAPI 项目结构

项目结构：

my_fastapi_app/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── schemas.py
│   ├── database.py
│   ├── crud.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   └── core/
│       ├── __init__.py
│       ├── config.py
│       └── security.py
├── tests/
│   ├── __init__.py
│   └── test_users.py
├── requirements.txt
└── README.md

关键文件示例：

  1. app/main.py

    from fastapi import FastAPI
from app.api import users, items

app = FastAPI()

app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

  2. app/database.py

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

DATABASE_URL = "mysql+pymysql://user:password@localhost/mydatabase"

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

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

  3. app/models.py

    from sqlalchemy import Column, Integer, String
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100))
    email = Column(String(100), unique=True, index=True)
    age = Column(Integer)

  4. app/schemas.py

    from pydantic import BaseModel, EmailStr, Field
from typing import Optional

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=100)
    email: EmailStr
    age: Optional[int] = Field(None, ge=0, le=120)

class UserRead(BaseModel):
    id: int
    name: str
    email: EmailStr
    age: Optional[int]

    class Config:
        orm_mode = True

  5. app/crud.py

    from sqlalchemy.orm import Session
from app import models, schemas

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

def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(name=user.name, email=user.email, age=user.age)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

  6. app/api/users.py

    from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.database import get_db

router = APIRouter()

@router.post("/", response_model=schemas.UserRead)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)

@router.get("/{user_id}", response_model=schemas.UserRead)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(models.User).filter(models.User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

解释：

  • 模块化设计：将用户相关的 API 路由放在 api/users.py，数据模型在 models.py，数据验证模型在 schemas.py，数据库操作在 crud.py，数据库配置在 database.py。
  • 依赖注入：通过 Depends(get_db) 获取数据库会话，确保每个请求都有独立的数据库连接。
  • 分层架构：将 API 路由、业务逻辑和数据访问层分离，提高代码的可维护性和可扩展性。

8. 测试与质量保证

测试是确保代码质量和功能正确性的重要环节。通过编写测试，可以发现并修复潜在的问题，提升代码的可靠性。

8.1 编写单元测试

单元测试：针对代码的最小可测试单元（如函数、方法）进行测试，确保其按预期工作。

使用 pytest 进行单元测试

  1. 安装 pytest

    pip install pytest

  2. 编写测试函数

    文件结构：

    my_fastapi_app/
├── app/
│   ├── ...
├── tests/
│   ├── __init__.py
│   └── test_users.py
└── ...

    tests/test_users.py

    from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users/",
        json={"name": "Test User", "email": "test@example.com", "age": 25}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test User"
    assert data["email"] == "test@example.com"
    assert data["age"] == 25

  3. 运行测试

    在项目根目录下运行：

    pytest

    输出示例：

    ============================= test session starts ==============================
platform darwin -- Python 3.9.7, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: /path/to/my_fastapi_app
collected 1 item

tests/test_users.py .                                                   [100%]

============================== 1 passed in 0.50s ===============================

解释：

  • TestClient：FastAPI 提供的测试客户端，模拟 HTTP 请求和响应。
  • 测试函数：以 test_ 开头的函数，pytest 会自动识别并执行。
  • 断言：使用 assert 语句验证响应的状态码和数据内容。

8.2 集成测试

集成测试：测试多个模块或服务之间的交互，确保它们协同工作。

示例：测试用户创建和读取

def test_create_and_read_user():
    # 创建用户
    response = client.post(
        "/users/",
        json={"name": "Integration Test", "email": "integration@example.com", "age": 30}
    )
    assert response.status_code == 200
    user = response.json()
    user_id = user["id"]

    # 读取用户
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    fetched_user = response.json()
    assert fetched_user["name"] == "Integration Test"
    assert fetched_user["email"] == "integration@example.com"
    assert fetched_user["age"] == 30

8.3 使用测试框架

pytest 是一个功能强大且易于使用的 Python 测试框架，支持参数化测试、fixture、插件扩展等。

示例：使用 fixture 提供测试数据

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

@pytest.fixture
def new_user():
    return {"name": "Fixture User", "email": "fixture@example.com", "age": 28}

def test_create_user_fixture(new_user):
    response = client.post("/users/", json=new_user)
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == new_user["name"]
    assert data["email"] == new_user["email"]
    assert data["age"] == new_user["age"]

解释：

  • @pytest.fixture：定义一个 fixture，提供测试函数所需的资源或数据。
  • 测试函数：接收 fixture 作为参数，pytest 会自动注入。

8.4 示例：使用 pytest 进行测试

完整测试文件：tests/test_users.py

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

@pytest.fixture
def new_user():
    return {"name": "Fixture User", "email": "fixture@example.com", "age": 28}

def test_create_user():
    response = client.post(
        "/users/",
        json={"name": "Test User", "email": "test@example.com", "age": 25}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Test User"
    assert data["email"] == "test@example.com"
    assert data["age"] == 25

def test_create_user_fixture(new_user):
    response = client.post("/users/", json=new_user)
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == new_user["name"]
    assert data["email"] == new_user["email"]
    assert data["age"] == new_user["age"]

def test_read_user():
    # 首先创建一个用户
    response = client.post(
        "/users/",
        json={"name": "Read Test", "email": "read@example.com", "age": 30}
    )
    assert response.status_code == 200
    user = response.json()
    user_id = user["id"]

    # 读取用户
    response = client.get(f"/users/{user_id}")
    assert response.status_code == 200
    fetched_user = response.json()
    assert fetched_user["name"] == "Read Test"
    assert fetched_user["email"] == "read@example.com"
    assert fetched_user["age"] == 30

def test_read_nonexistent_user():
    response = client.get("/users/9999")  # 假设 ID 9999 不存在
    assert response.status_code == 404
    data = response.json()
    assert data["detail"] == "User not found"

运行测试：

pytest

输出示例：

============================= test session starts ==============================
platform linux -- Python 3.9.7, pytest-6.2.4, py-1.10.0, pluggy-0.13.1
rootdir: /path/to/my_fastapi_app
collected 4 items

tests/test_users.py ....                                               [100%]

============================== 4 passed in 0.50s ===============================

解释：

  • 多个测试函数：测试用户创建、读取和处理不存在用户的情况。
  • 使用 fixture：new_user fixture 提供标准化的测试数据。

9. 部署与监控

部署是将开发好的应用发布到生产环境中，让用户可以访问和使用。监控确保应用在生产环境中运行稳定，及时发现和响应问题。

9.1 部署环境配置

  • 配置文件管理：使用环境变量或配置文件管理不同环境的配置（开发、测试、生产）。
  • 依赖管理：通过 requirements.txt 管理项目依赖，确保部署环境的一致性。
  • 安全配置：确保敏感信息（如数据库密码、API 密钥）通过环境变量或安全存储进行管理，不在代码中硬编码。

示例：使用 python-dotenv 管理环境变量

  1. 安装 python-dotenv

    pip install python-dotenv

  2. 创建 .env 文件

    DATABASE_URL=mysql+pymysql://user:password@localhost/mydatabase
SECRET_KEY=your_secret_key

  3. 加载环境变量

    app/core/config.py

    from pydantic import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str

    class Config:
        env_file = ".env"

settings = Settings()

  4. 在应用中使用配置

    from app.core.config import settings
from sqlalchemy import create_engine

engine = create_engine(settings.database_url)

9.2 持续集成与持续部署 (CI/CD)

CI/CD 是软件工程中的一套实践，通过自动化构建、测试和部署流程，提高开发效率和代码质量。

常见工具：

  • 持续集成 (CI)：Jenkins、GitHub Actions、GitLab CI、Travis CI
  • 持续部署 (CD)：Jenkins、GitHub Actions、GitLab CI、CircleCI

示例：使用 GitHub Actions 进行 CI

.github/workflows/ci.yml

name: CI

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  build:

    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v2

    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.9'

    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt

    - name: Run tests
      run: |
        pytest

解释：

  • 触发条件：在 main 分支的 push 和 pull request 事件触发。

  • 作业步骤

    ：

    • 检出代码：使用 actions/checkout 检出代码。
    • 设置 Python：使用 actions/setup-python 设置 Python 环境。
    • 安装依赖：安装项目依赖。
    • 运行测试：使用 pytest 运行测试。

9.3 监控与报警

监控可以实时了解应用的运行状态，及时发现和响应问题。

常见的监控工具：

  • 应用性能监控（APM）：New Relic、Datadog、Prometheus + Grafana
  • 日志监控：ELK Stack（Elasticsearch、Logstash、Kibana）、Graylog
  • 服务器监控：Nagios、Zabbix

示例：使用 Prometheus 和 Grafana 进行监控

  1. 安装 Prometheus

    • 访问 Prometheus 下载页面 下载并安装。

  2. 配置 Prometheus

    prometheus.yml

    global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'fastapi'
    static_configs:
      - targets: ['localhost:8000']

  3. 集成 FastAPI 与 Prometheus

    使用 prometheus-fastapi-instrumentator 进行集成。

    • 安装依赖

      pip install prometheus-fastapi-instrumentator

    • 配置 FastAPI

      from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator

app = FastAPI()

# 定义路由
@app.get("/")
async def read_root():
    return {"message": "Hello, World!"}

# 集成 Prometheus
Instrumentator().instrument(app).expose(app)

  4. 运行 Prometheus 和 Grafana

    • 启动 Prometheus

      prometheus --config.file=prometheus.yml

    • 安装并启动 Grafana

      • 访问 Grafana 官网 下载并安装。
      • 添加 Prometheus 作为数据源。
      • 创建仪表板，展示 FastAPI 应用的指标。

9.4 示例：使用 Docker 部署 FastAPI 应用

Dockerfile

FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 复制依赖文件
COPY requirements.txt .

# 安装依赖
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用代码
COPY . .

# 运行应用
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

构建 Docker 镜像

docker build -t my_fastapi_app .

运行 Docker 容器

docker run -d -p 80:80 my_fastapi_app

使用 Docker Compose 进行部署

docker-compose.yml

version: '3.8'

services:
  web:
    build: .
    ports:
      - "80:80"
    environment:
      - DATABASE_URL=mysql+pymysql://user:password@db/mydatabase
    depends_on:
      - db

  db:
    image: mysql:8.0
    restart: always
    environment:
      MYSQL_DATABASE: mydatabase
      MYSQL_USER: user
      MYSQL_PASSWORD: password
      MYSQL_ROOT_PASSWORD: rootpassword
    ports:
      - "3306:3306"
    volumes:
      - db_data:/var/lib/mysql

volumes:
  db_data:

启动服务

docker-compose up -d

解释：

  • web 服务：构建并运行 FastAPI 应用，暴露 80 端口，依赖于 db 服务。
  • db 服务：运行 MySQL 8.0，配置数据库、用户和密码，暴露 3306 端口，并使用 Docker 卷持久化数据。

10. 附加资源

  • 官方文档
    • FastAPI 官方文档
    • SQLAlchemy 官方文档
    • Pydantic 官方文档
    • PyMySQL 官方文档
    • Passlib 官方文档
    • pytest 官方文档
  • 教程与课程
    • FastAPI 教程 by Sebastián Ramírez
    • SQLAlchemy 教程 by Miguel Grinberg
    • Udemy Python 后端开发课程
  • 社区与论坛
    • Stack Overflow FastAPI 标签
    • Reddit FastAPI 社区
    • SQLAlchemy GitHub 讨论区

11. 总结

后端开发涉及多个方面，包括错误处理、日志记录、数据库管理、安全性、代码组织、测试和部署等。通过掌握以下关键点，您可以构建出高质量、稳定且可维护的后端应用：

  1. 错误处理：
    • 使用 try-except 捕获异常，防止应用崩溃。
    • 在 FastAPI 中利用内置异常处理器和自定义异常，提高错误响应的质量和一致性。
  2. 日志记录：
    • 配置 logging 模块，记录关键信息和错误。
    • 在应用中集成日志记录，实时监控请求和异常。
  3. 数据库事务管理：
    • 理解事务的 ACID 属性，确保数据的一致性和完整性。
    • 使用 SQLAlchemy 提供的事务管理功能，安全地处理数据库操作。
  4. 输入验证与数据清洗：
    • 利用 Pydantic 模型验证用户输入，防止恶意数据和 SQL 注入。
    • 使用 ORM 或参数化查询，确保数据库操作的安全性。
  5. 安全性最佳实践：
    • 哈希和验证用户密码，保护用户信息。
    • 实现认证与授权，控制用户访问权限。
    • 防止常见的 Web 安全漏洞，如 XSS 和 CSRF。
    • 使用 HTTPS 确保数据传输的安全性。
  6. 代码结构与组织：
    • 模块化设计，分层架构，提高代码的可读性和可维护性。
    • 组织项目结构，遵循最佳实践，促进团队协作。
  7. 测试与质量保证：
    • 编写单元测试和集成测试，确保代码功能正确。
    • 使用测试框架（如 pytest）自动化测试流程，提高开发效率。
  8. 部署与监控：
    • 配置部署环境，使用工具（如 Docker）简化部署过程。
    • 实现持续集成与持续部署（CI/CD），提高发布效率。
    • 监控应用运行状况，及时发现和解决问题。

接下来的步骤：

  1. 动手实践：
    • 根据本指南，构建一个简单的 FastAPI 应用，集成数据库操作、认证和日志记录。
    • 编写测试，确保应用功能的正确性。
  2. 深入学习：
    • 探索更高级的功能，如 WebSocket、后台任务、文件上传等。
    • 学习更多关于数据库优化、分布式系统和微服务架构的知识。
  3. 参与社区：
    • 加入 FastAPI 和 SQLAlchemy 的社区，与其他开发者交流经验。
    • 参与开源项目，提升实战能力。
  4. 持续提升：
    • 学习和应用新的工具和技术，保持技术的先进性。
    • 关注安全性更新，确保应用的安全。

http://www.kler.cn/a/409778.html

相关文章：

  • Leetcode207. 课程表(HOT100)
  • 如何选择服务器
  • 【大数据技术与开发实训】携程景点在线评论分析
  • 轻松解析 PDF 文档：深入了解 Python 的 pdfplumber 库
  • 七天掌握SQL---＞第五天：数据库安全与权限管理
  • 深入解析 EasyExcel 组件原理与应用
  • 游卡，科锐国际，蓝禾，汤臣倍健，三七互娱，顺丰，快手，途游游戏25秋招内推
  • Oracle-索引的创建和优化
  • 学习prompt
  • GitLab｜GitLab报错：Restoring PostgreSQL database gitlabhq_production...
  • HTML密码小眼睛
  • 区块链学习笔记(1)--区块、链和共识 区块链技术入门
  • 【分治】--- 快速选择算法
  • 【优选算法】前缀和
  • C++入门学习基础
  • C++ 编程指南06 - 不要泄漏任何资源
  • 蓝桥杯每日真题 - 第23天
  • 【C++】C++11新特性详解：可变参数模板与emplace系列的应用
  • World of Warcraft /script SetRaidTarget(“target“, n, ““) n=8,7,6,5,4,3,2,1,0
  • 深入探讨异步 API 的设计与实现
  • [C++]了解内置类型升级
  • Qt 开发笔记
  • 提供html2canvas+jsPDF将HTML页面以A4纸方式导出为PDF后，内容分页时存在截断的解决思路
  • 人工智能学习框架：理论与实践的结合
  • JavaScript网页设计案例：动态交互与用户体验提升
  • 音频档案批量拷贝：专业SD拷贝机解决方案