资源栈 - www.zyz88.com

Python FastAPI高性能API开发:从0到1构建RESTful服务

admin
2026-09-09 5 阅读 0 评论 0 点赞

Python做后端开发,FastAPI几乎是当下最好的选择。自动生成API文档、原生异步支持、类型安全的请求校验——这些特性让开发效率大幅提升。如果你还在用Flask或Django做纯API服务,不妨试试FastAPI。本文从0到1带你构建一个完整的RESTful API服务。

FastAPI简介

FastAPI是一个现代、高性能的Python Web框架,基于Starlette和Pydantic。它的核心优势在于:自动生成交互式API文档、原生异步支持、类型安全的请求校验、极高的开发效率。在2026年,FastAPI已经成为Python后端开发的首选框架之一。

项目初始化

pip install fastapi uvicorn sqlalchemy alembic pydantic[email] python-jose[cryptography] passlib[bcrypt]

项目结构:

app/
├── main.py              # 应用入口
├── config.py            # 配置管理
├── database.py          # 数据库连接
├── models/              # SQLAlchemy模型
│   ├── user.py
│   └── post.py
├── schemas/             # Pydantic模式
│   ├── user.py
│   └── post.py
├── api/                 # 路由
│   ├── auth.py
│   ├── users.py
│   └── posts.py
├── core/                # 核心功能
│   ├── security.py      # JWT和密码哈希
│   └── deps.py          # 依赖注入
└── tests/               # 测试

第一个API

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, EmailStr

app = FastAPI(title="Blog API", version="1.0")

class UserCreate(BaseModel):
    name: str
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    class Config:
        from_attributes = True

@app.post("/users", response_model=UserResponse, status_code=201)
def create_user(user: UserCreate):
    # 创建用户逻辑
    if user.email == "exists@example.com":
        raise HTTPException(status_code=400, detail="邮箱已注册")
    return {"id": 1, "name": user.name, "email": user.email}

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    return {"id": user_id, "name": "测试用户", "email": "test@example.com"}

启动后访问/docs即可看到自动生成的Swagger UI,可以直接在线测试API。

数据库集成

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://user:pass@localhost/dbname"
engine = create_engine(SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# models/user.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, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

依赖注入与认证

# core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from app.database import SessionLocal

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")

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

async def get_current_user(token: str = Depends(oauth2_scheme), db = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise credentials_exception
    return user

# 使用
@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

异步性能优化

FastAPI原生支持async/await,但要注意数据库操作:

# 使用异步数据库驱动
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = sessionmaker(engine, class_=AsyncSession)

@app.get("/posts")
async def list_posts(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Post))
    return result.scalars().all()
场景 推荐方式
CPU密集型 同步函数,FastAPI自动用线程池处理
IO密集型(DB/HTTP) async函数+异步驱动
混合场景 async函数中用run_in_threadpool调用同步代码

部署方案

  • 开发环境——uvicorn main:app –reload,自动重载
  • 生产环境——gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
  • Docker部署——用python:3.12-slim基础镜像,多阶段构建减小体积
  • 进程管理——用systemd或supervisor管理进程,异常自动重启
  • 反向代理——Nginx前置,处理静态资源、SSL、负载均衡

测试与文档

FastAPI配合pytest和TestClient可以轻松编写API测试:

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

client = TestClient(app)

def test_create_user():
    response = client.post("/users", json={
        "name": "测试", "email": "test@example.com", "password": "pass123"
    })
    assert response.status_code == 201
    assert response.json()["email"] == "test@example.com"

FastAPI自动生成的OpenAPI文档可以直接导入Postman、Apifox等工具,也可以用来生成前端SDK。

FastAPI的设计哲学是”高效开发+高性能运行”,非常适合快速构建RESTful API和微服务。配合Python丰富的生态(数据分析、AI/ML、爬虫),FastAPI在AI应用后端领域尤其有优势。

文章标题 Python FastAPI高性能API开发:从0到1构建RESTful服务
本文由 资源栈 原创发布,转载请注明出处并保留原文链接。