FastAPI快速构建高性能Web应用

📅 发布于 2024-01-05 ⏱️ 约18分钟阅读 👁️ 1567次阅读 💬 12条评论

FastAPI Web开发 Python

📑 文章目录

一、为什么选择FastAPI？
二、快速开始
三、路由与参数
四、请求体与数据验证
五、模板渲染
六、静态文件与媒体
七、数据库连接
八、部署上线

一、为什么选择FastAPI？

FastAPI是Python最现代化的高性能Web框架，基于Starlette和Pydantic构建，具有以下特点：

🚀

高性能

性能可媲美Node.js和Go，是Python最快的Web框架之一

📝

自动文档

自动生成API文档，支持Swagger和ReDoc

🔒

类型安全

基于Pydantic，自动验证请求数据

⚡

异步支持

原生支持async/await，高并发场景表现优异

二、快速开始

# 安装FastAPI
pip install fastapi uvicorn

# 创建main.py
from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/html")
async def html():
    return HTMLResponse("<h1>Hello HTML!</h1>")

💡 启动服务器 使用 uvicorn main:app --reload 启动开发服务器，修改代码自动重载。

2.1 项目结构

# 推荐的项目结构 ├── app/ # 应用主目录 │ ├── __init__.py │ ├── main.py # FastAPI入口 │ ├── routers/ # 路由模块 │ ├── models/ # 数据模型 │ ├── schemas/ # Pydantic模型 │ └── database.py # 数据库配置 ├── templates/ # HTML模板 ├── static/ # 静态文件 ├── requirements.txt └── .env

三、路由与参数

from fastapi import FastAPI, Query, Path
from typing import Optional

app = FastAPI()

# 路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

# 查询参数
@app.get("/items/")
async def read_items(
    q: Optional[str] = Query(None, min_length=2, max_length=50),
    skip: int = 0,
    limit: int = 10
):
    return {"q": q, "skip": skip, "limit": limit}

# 路径参数带验证
@app.get("/users/{user_id}")
async def read_user(
    user_id: int = Path(..., ge=1, le=1000),
    q: Optional[str] = None
):
    return {"user_id": user_id, "q": q}

四、请求体与数据验证

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

app = FastAPI()

# 定义数据模型
class UserIn(BaseModel):
    username: str
    email: EmailStr
    password: str
    age: Optional[int] = None

class UserOut(BaseModel):
    username: str
    email: str
    age: Optional[int] = None

# POST请求
@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):
    # 数据验证自动完成
    return {
        "username": user.username,
        "email": user.email,
        "age": user.age
    }

五、模板渲染

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# 挂载静态文件
app.mount("/static", StaticFiles(directory="static"), name="static")

# 初始化模板
templates = Jinja2Templates(directory="templates")

@app.get("/hello/{name}", response_class=HTMLResponse)
async def hello(request: Request, name: str):
    return templates.TemplateResponse("hello.html", {
        "request": request,
        "name": name
    })

📄 templates/hello.html 示例

<html>
<body>
    <h1>Hello, {{ name }}!</h1>
</body>
</html>

六、静态文件与媒体

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# CSS、JS、图片等静态资源
app.mount("/static", StaticFiles(directory="static"), name="static")
app.mount("/media", StaticFiles(directory="media"), name="media")

# 文件下载
from fastapi.responses import FileResponse

@app.get("/download/{filename}")
async def download_file(filename: str):
    return FileResponse(f"downloads/{filename}")

七、数据库连接

from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# 数据库配置
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 定义模型
class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String)

Base.metadata.create_all(bind=engine)

# 依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# CRUD操作
@app.post("/items/")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

八、部署上线

8.1 使用Gunicorn

# 安装gunicorn
pip install gunicorn

# 启动（4个工作进程）
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

8.2 使用Docker

# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

🚀 生产环境建议

使用进程管理器（如Supervisor）
配置Nginx反向代理
开启HTTPS
设置环境变量

🤖

博主

热爱技术的开发者，专注于Python、数据分析与Web开发

← 返回首页