Python API 开发指南

目录

• RESTful API
• GraphQL
• WebSocket
• gRPC
• 中间件
• 认证授权
• API 文档
• 测试
• 部署

RESTful API

FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List

app = FastAPI()

class User(BaseModel):
    name: str
    email: str

@app.get("/users", response_model=List[User])
async def get_users():
    return user_service.get_all()

@app.post("/users", response_model=User)
async def create_user(user: User):
    return user_service.create(user)

@app.put("/users/{user_id}", response_model=User)
async def update_user(user_id: int, user: User):
    return user_service.update(user_id, user)

@app.delete("/users/{user_id}")
async def delete_user(user_id: int):
    user_service.delete(user_id)
    return {"message": "User deleted"}

Flask

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/users', methods=['GET'])
def get_users():
    users = user_service.get_all()
    return jsonify(users)

@app.route('/users', methods=['POST'])
def create_user():
    user_data = request.json
    user = user_service.create(user_data)
    return jsonify(user), 201

GraphQL

Strawberry

import strawberry
from typing import List

@strawberry.type
class User:
    id: int
    name: str
    email: str

@strawberry.type
class Query:
    @strawberry.field
    def users(self) -> List[User]:
        return user_service.get_all()

@strawberry.type
class Mutation:
    @strawberry.mutation
    def create_user(self, name: str, email: str) -> User:
        return user_service.create({"name": name, "email": email})

schema = strawberry.Schema(query=Query, mutation=Mutation)

WebSocket

FastAPI WebSocket

from fastapi import WebSocket
from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    async def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Message: {data}")
    except WebSocketDisconnect:
        await manager.disconnect(websocket)

gRPC

服务定义

syntax = "proto3";

package user;

service UserService {
    rpc GetUser (UserRequest) returns (UserResponse) {}
    rpc CreateUser (CreateUserRequest) returns (UserResponse) {}
}

message UserRequest {
    int64 id = 1;
}

message CreateUserRequest {
    string name = 1;
    string email = 2;
}

message UserResponse {
    int64 id = 1;
    string name = 2;
    string email = 3;
}

服务实现

import grpc
from concurrent import futures
import user_pb2
import user_pb2_grpc

class UserServicer(user_pb2_grpc.UserServiceServicer):
    def GetUser(self, request, context):
        user = user_service.get_by_id(request.id)
        return user_pb2.UserResponse(
            id=user.id,
            name=user.name,
            email=user.email
        )

    def CreateUser(self, request, context):
        user = user_service.create({
            "name": request.name,
            "email": request.email
        })
        return user_pb2.UserResponse(
            id=user.id,
            name=user.name,
            email=user.email
        )

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
user_pb2_grpc.add_UserServiceServicer_to_server(UserServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()

中间件

FastAPI 中间件

from fastapi import Request
import time

@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

认证授权

JWT 认证

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async 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"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = user_service.get_by_username(username)
    if user is None:
        raise credentials_exception
    return user

API 文档

FastAPI 自动文档

from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI(
    title="User API",
    description="API for user management",
    version="1.0.0"
)

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    openapi_schema = get_openapi(
        title="User API",
        version="1.0.0",
        description="API for user management",
        routes=app.routes,
    )
    app.openapi_schema = openapi_schema
    return app.openapi_schema

app.openapi = custom_openapi

测试

单元测试

from fastapi.testclient import TestClient
from app import app

client = TestClient(app)

def test_get_users():
    response = client.get("/users")
    assert response.status_code == 200
    assert len(response.json()) > 0

def test_create_user():
    response = client.post(
        "/users",
        json={"name": "John", "email": "[email protected]"}
    )
    assert response.status_code == 201
    assert response.json()["name"] == "John"

部署

Docker 部署

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes 部署

apiVersion: apps/v1
kind: Deployment
metadata:
  name: python-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: python-api
  template:
    metadata:
      labels:
        app: python-api
    spec:
      containers:
      - name: python-api
        image: python-api:latest
        ports:
        - containerPort: 8000