Cao Note

外观

Python常用Web框架 ​

框架对比 ​

框架类型特点适用场景
Django全栈框架功能丰富、内置ORM、Admin后台、认证系统大型Web应用、企业级项目
Flask微框架轻量灵活、扩展丰富、学习曲线平缓小型Web应用、API服务、原型开发
FastAPI现代Web框架高性能、自动文档、类型提示、异步支持RESTful API、微服务、高性能应用

Django ​

Django简介 ​

Django是一个高级Python Web框架，鼓励快速开发和清洁、实用的设计。它由经验丰富的开发者构建，解决了Web开发中的许多常见问题，因此您可以专注于编写应用程序，而无需重新发明轮子。

安装Django ​

bash
pip install django
pip install django

创建Django项目 ​

bash
# 创建项目
django-admin startproject myproject

# 进入项目目录
cd myproject

# 创建应用
python manage.py startapp myapp

# 运行开发服务器
python manage.py runserver
# 创建项目
django-admin startproject myproject

# 进入项目目录
cd myproject

# 创建应用
python manage.py startapp myapp

# 运行开发服务器
python manage.py runserver

Django基本结构 ​

myproject/
├── manage.py          # 项目管理脚本
├── myproject/         # 项目配置目录
│   ├── __init__.py
│   ├── asgi.py        # ASGI入口
│   ├── settings.py    # 项目设置
│   ├── urls.py        # 主URL配置
│   └── wsgi.py        # WSGI入口
└── myapp/             # 应用目录
    ├── __init__.py
    ├── admin.py       # 后台管理配置
    ├── apps.py        # 应用配置
    ├── migrations/    # 数据库迁移文件
    ├── models.py      # 数据模型
    ├── tests.py       # 测试文件
    └── views.py       # 视图函数
myproject/
├── manage.py          # 项目管理脚本
├── myproject/         # 项目配置目录
│   ├── __init__.py
│   ├── asgi.py        # ASGI入口
│   ├── settings.py    # 项目设置
│   ├── urls.py        # 主URL配置
│   └── wsgi.py        # WSGI入口
└── myapp/             # 应用目录
    ├── __init__.py
    ├── admin.py       # 后台管理配置
    ├── apps.py        # 应用配置
    ├── migrations/    # 数据库迁移文件
    ├── models.py      # 数据模型
    ├── tests.py       # 测试文件
    └── views.py       # 视图函数

Django视图 ​

python
# myapp/views.py
from django.http import HttpResponse
from django.shortcuts import render

# 函数视图
def index(request):
    return HttpResponse("Hello, Django!")

# 类视图
from django.views import View

class MyView(View):
    def get(self, request):
        return HttpResponse("This is a GET request")
    
    def post(self, request):
        return HttpResponse("This is a POST request")
# myapp/views.py
from django.http import HttpResponse
from django.shortcuts import render

# 函数视图
def index(request):
    return HttpResponse("Hello, Django!")

# 类视图
from django.views import View

class MyView(View):
    def get(self, request):
        return HttpResponse("This is a GET request")
    
    def post(self, request):
        return HttpResponse("This is a POST request")

Django路由 ​

python
# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('myapp/', include('myapp.urls')),  # 包含应用的URL配置
]

# myapp/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('', views.index, name='index'),  # 函数视图路由
    path('myview/', views.MyView.as_view(), name='myview'),  # 类视图路由
]
# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('myapp/', include('myapp.urls')),  # 包含应用的URL配置
]

# myapp/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('', views.index, name='index'),  # 函数视图路由
    path('myview/', views.MyView.as_view(), name='myview'),  # 类视图路由
]

Django模型 ​

python
# myapp/models.py
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)  # 字符串字段
    email = models.EmailField(unique=True)  # 邮箱字段，唯一约束
    age = models.IntegerField(null=True, blank=True)  # 整数字段，允许为空
    created_at = models.DateTimeField(auto_now_add=True)  # 创建时间，自动添加
    updated_at = models.DateTimeField(auto_now=True)  # 更新时间，自动更新
    
    def __str__(self):
        return self.name  # 模型实例的字符串表示
# myapp/models.py
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)  # 字符串字段
    email = models.EmailField(unique=True)  # 邮箱字段，唯一约束
    age = models.IntegerField(null=True, blank=True)  # 整数字段，允许为空
    created_at = models.DateTimeField(auto_now_add=True)  # 创建时间，自动添加
    updated_at = models.DateTimeField(auto_now=True)  # 更新时间，自动更新
    
    def __str__(self):
        return self.name  # 模型实例的字符串表示

Django迁移 ​

bash
# 创建迁移文件
python manage.py makemigrations

# 应用迁移
python manage.py migrate

# 查看迁移状态
python manage.py showmigrations
# 创建迁移文件
python manage.py makemigrations

# 应用迁移
python manage.py migrate

# 查看迁移状态
python manage.py showmigrations

Django Admin ​

Django内置了强大的后台管理系统，可以轻松管理数据库中的数据。

python
# myapp/admin.py
from django.contrib import admin
from .models import User

admin.site.register(User)  # 注册模型到Admin

# 自定义Admin
class UserAdmin(admin.ModelAdmin):
    list_display = ('name', 'email', 'age', 'created_at')  # 列表显示字段
    search_fields = ('name', 'email')  # 搜索字段
    list_filter = ('age', 'created_at')  # 过滤字段
    ordering = ('-created_at',)  # 排序字段

admin.site.register(User, UserAdmin)  # 注册自定义Admin
# myapp/admin.py
from django.contrib import admin
from .models import User

admin.site.register(User)  # 注册模型到Admin

# 自定义Admin
class UserAdmin(admin.ModelAdmin):
    list_display = ('name', 'email', 'age', 'created_at')  # 列表显示字段
    search_fields = ('name', 'email')  # 搜索字段
    list_filter = ('age', 'created_at')  # 过滤字段
    ordering = ('-created_at',)  # 排序字段

admin.site.register(User, UserAdmin)  # 注册自定义Admin

Django表单 ​

python
# myapp/forms.py
from django import forms
from .models import User

class UserForm(forms.ModelForm):
    class Meta:
        model = User
        fields = ('name', 'email', 'age')  # 表单包含的字段
        widgets = {
            'name': forms.TextInput(attrs={'class': 'form-control'}),
            'email': forms.EmailInput(attrs={'class': 'form-control'}),
            'age': forms.NumberInput(attrs={'class': 'form-control'}),
        }

# myapp/views.py
def create_user(request):
    if request.method == 'POST':
        form = UserForm(request.POST)
        if form.is_valid():
            form.save()  # 保存到数据库
            return HttpResponse("User created successfully!")
    else:
        form = UserForm()
    return render(request, 'create_user.html', {'form': form})
# myapp/forms.py
from django import forms
from .models import User

class UserForm(forms.ModelForm):
    class Meta:
        model = User
        fields = ('name', 'email', 'age')  # 表单包含的字段
        widgets = {
            'name': forms.TextInput(attrs={'class': 'form-control'}),
            'email': forms.EmailInput(attrs={'class': 'form-control'}),
            'age': forms.NumberInput(attrs={'class': 'form-control'}),
        }

# myapp/views.py
def create_user(request):
    if request.method == 'POST':
        form = UserForm(request.POST)
        if form.is_valid():
            form.save()  # 保存到数据库
            return HttpResponse("User created successfully!")
    else:
        form = UserForm()
    return render(request, 'create_user.html', {'form': form})

Django模板 ​

html
<!-- myapp/templates/create_user.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Create User</title>
</head>
<body>
    <h1>Create User</h1>
    <form method="post">
        {% csrf_token %}  <!-- CSRF保护 -->
        {{ form.as_p }}  <!-- 渲染表单 -->
        <button type="submit">Create</button>
    </form>
</body>
</html>
<!-- myapp/templates/create_user.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Create User</title>
</head>
<body>
    <h1>Create User</h1>
    <form method="post">
        {% csrf_token %}  <!-- CSRF保护 -->
        {{ form.as_p }}  <!-- 渲染表单 -->
        <button type="submit">Create</button>
    </form>
</body>
</html>

Django REST Framework (DRF) ​

Django REST Framework是Django的一个强大且灵活的工具包，用于构建Web API。

bash
pip install djangorestframework
pip install djangorestframework
python
# myproject/settings.py
INSTALLED_APPS = [
    # ...
    'rest_framework',
]

# myapp/serializers.py
from rest_framework import serializers
from .models import User

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = '__all__'  # 序列化所有字段

# myapp/views.py
from rest_framework import viewsets

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

# myapp/urls.py
from rest_framework.routers import DefaultRouter

router = DefaultRouter()
router.register(r'users', UserViewSet)

urlpatterns = [
    # ...
    path('api/', include(router.urls)),
]
# myproject/settings.py
INSTALLED_APPS = [
    # ...
    'rest_framework',
]

# myapp/serializers.py
from rest_framework import serializers
from .models import User

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = '__all__'  # 序列化所有字段

# myapp/views.py
from rest_framework import viewsets

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

# myapp/urls.py
from rest_framework.routers import DefaultRouter

router = DefaultRouter()
router.register(r'users', UserViewSet)

urlpatterns = [
    # ...
    path('api/', include(router.urls)),
]

Flask ​

Flask简介 ​

Flask是一个轻量级的WSGI Web应用框架。它旨在使入门快速且容易，并能够扩展到复杂的应用程序。Flask提供了建议，但不强制执行任何依赖项或项目布局。

安装Flask ​

bash
pip install flask
pip install flask

Flask基本示例 ​

python
# app.py
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/')
def hello():
    return "Hello, Flask!"

@app.route('/api', methods=['GET', 'POST'])
def api():
    if request.method == 'GET':
        return jsonify({"message": "This is a GET request"})
    elif request.method == 'POST':
        data = request.json
        return jsonify({"message": "This is a POST request", "data": data})

if __name__ == '__main__':
    app.run(debug=True)
# app.py
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/')
def hello():
    return "Hello, Flask!"

@app.route('/api', methods=['GET', 'POST'])
def api():
    if request.method == 'GET':
        return jsonify({"message": "This is a GET request"})
    elif request.method == 'POST':
        data = request.json
        return jsonify({"message": "This is a POST request", "data": data})

if __name__ == '__main__':
    app.run(debug=True)

运行Flask应用 ​

bash
python app.py
python app.py

Flask路由 ​

python
# 基本路由
@app.route('/')
def index():
    return 'Index Page'

# 带参数的路由
@app.route('/user/<username>')
def show_user_profile(username):
    return f'User {username}'

# 带类型转换的路由
@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post {post_id}'

# 多URL路由
@app.route('/')
@app.route('/index')
def index():
    return 'Index Page'

# 不同HTTP方法的路由
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理登录逻辑
        return 'Login POST'
    else:
        # 显示登录表单
        return 'Login GET'
# 基本路由
@app.route('/')
def index():
    return 'Index Page'

# 带参数的路由
@app.route('/user/<username>')
def show_user_profile(username):
    return f'User {username}'

# 带类型转换的路由
@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'Post {post_id}'

# 多URL路由
@app.route('/')
@app.route('/index')
def index():
    return 'Index Page'

# 不同HTTP方法的路由
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理登录逻辑
        return 'Login POST'
    else:
        # 显示登录表单
        return 'Login GET'

Flask模板 ​

bash
pip install jinja2
pip install jinja2
python
# app.py
from flask import render_template

@app.route('/hello/<name>')
def hello(name=None):
    return render_template('hello.html', name=name)
# app.py
from flask import render_template

@app.route('/hello/<name>')
def hello(name=None):
    return render_template('hello.html', name=name)
html
<!-- templates/hello.html -->
<!doctype html>
<html>
<head>
    <title>Hello from Flask</title>
</head>
<body>
    {% if name %}
        <h1>Hello {{ name }}!</h1>
    {% else %}
        <h1>Hello, World!</h1>
    {% endif %}
</body>
</html>
<!-- templates/hello.html -->
<!doctype html>
<html>
<head>
    <title>Hello from Flask</title>
</head>
<body>
    {% if name %}
        <h1>Hello {{ name }}!</h1>
    {% else %}
        <h1>Hello, World!</h1>
    {% endif %}
</body>
</html>

Flask请求处理 ​

python
from flask import request

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']  # 表单数据
    password = request.form['password']
    
    json_data = request.json  # JSON数据
    
    args = request.args  # URL参数
    
    files = request.files['file']  # 文件上传
    
    return f'Username: {username}'
from flask import request

@app.route('/login', methods=['POST'])
def login():
    username = request.form['username']  # 表单数据
    password = request.form['password']
    
    json_data = request.json  # JSON数据
    
    args = request.args  # URL参数
    
    files = request.files['file']  # 文件上传
    
    return f'Username: {username}'

Flask响应处理 ​

python
from flask import jsonify, make_response

@app.route('/json')
def json_response():
    return jsonify({"message": "Hello, JSON!"})

@app.route('/custom')
def custom_response():
    response = make_response('Custom Response')
    response.headers['Content-Type'] = 'text/plain'
    response.status_code = 201
    return response
from flask import jsonify, make_response

@app.route('/json')
def json_response():
    return jsonify({"message": "Hello, JSON!"})

@app.route('/custom')
def custom_response():
    response = make_response('Custom Response')
    response.headers['Content-Type'] = 'text/plain'
    response.status_code = 201
    return response

Flask扩展 ​

Flask有丰富的扩展生态系统，可以轻松添加各种功能：

扩展功能安装命令
Flask-SQLAlchemyORM数据库集成pip install flask-sqlalchemy
Flask-Migrate数据库迁移pip install flask-migrate
Flask-RESTfulREST API开发pip install flask-restful
Flask-JWT-ExtendedJWT认证pip install flask-jwt-extended
Flask-WTF表单处理pip install flask-wtf
Flask-Login用户认证pip install flask-login

Flask-SQLAlchemy示例 ​

python
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 定义模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    
    def __repr__(self):
        return f'<User {self.username}>'

# 创建数据库
with app.app_context():
    db.create_all()

# 使用模型
@app.route('/add_user/<username>/<email>')
def add_user(username, email):
    user = User(username=username, email=email)
    db.session.add(user)
    db.session.commit()
    return f'User {username} added'

@app.route('/users')
def get_users():
    users = User.query.all()
    return ', '.join([user.username for user in users])
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 定义模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    
    def __repr__(self):
        return f'<User {self.username}>'

# 创建数据库
with app.app_context():
    db.create_all()

# 使用模型
@app.route('/add_user/<username>/<email>')
def add_user(username, email):
    user = User(username=username, email=email)
    db.session.add(user)
    db.session.commit()
    return f'User {username} added'

@app.route('/users')
def get_users():
    users = User.query.all()
    return ', '.join([user.username for user in users])

FastAPI ​

FastAPI简介 ​

FastAPI是一个现代、快速（高性能）的Web框架，用于基于标准Python类型提示构建API。它具有以下特点：

  • 快速：非常高的性能，与NodeJS和Go相当
  • 快速编码：将开发速度提高约200%至300%
  • 更少的错误：减少约40%的人为（开发者）错误
  • 直观：强大的编辑器支持，自动补全无处不在
  • 简单：易于学习和使用
  • 短：减少代码重复
  • 健壮：生产就绪代码，具有自动交互式文档

安装FastAPI ​

bash
pip install fastapi uvicorn
pip install fastapi uvicorn

FastAPI基本示例 ​

python
# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}
# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

运行FastAPI应用 ​

bash
uvicorn main:app --reload
uvicorn main:app --reload

FastAPI类型提示 ​

FastAPI充分利用Python的类型提示来自动验证请求数据和生成文档。

python
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义数据模型
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 定义数据模型
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

FastAPI路径参数 ​

python
@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

# 路径参数验证
from fastapi import Path

@app.get("/items/{item_id}")
def read_item(item_id: int = Path(..., title="The ID of the item to get", ge=1)):
    return {"item_id": item_id}
@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

# 路径参数验证
from fastapi import Path

@app.get("/items/{item_id}")
def read_item(item_id: int = Path(..., title="The ID of the item to get", ge=1)):
    return {"item_id": item_id}

FastAPI查询参数 ​

python
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 查询参数验证
from fastapi import Query

@app.get("/items/")
def read_items(q: str = Query(None, min_length=3, max_length=50, regex="^fixedquery$")):
    return {"q": q}
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 查询参数验证
from fastapi import Query

@app.get("/items/")
def read_items(q: str = Query(None, min_length=3, max_length=50, regex="^fixedquery$")):
    return {"q": q}

FastAPI请求体 ​

python
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

FastAPI响应模型 ​

python
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

class ItemResponse(BaseModel):
    name: str
    price: float
    tax: float = None

@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
    return item
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

class ItemResponse(BaseModel):
    name: str
    price: float
    tax: float = None

@app.post("/items/", response_model=ItemResponse)
def create_item(item: Item):
    return item

FastAPI异步支持 ​

python
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/")
async def read_root():
    await asyncio.sleep(1)  # 模拟异步操作
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    await asyncio.sleep(0.5)  # 模拟异步操作
    return {"item_id": item_id, "q": q}
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/")
async def read_root():
    await asyncio.sleep(1)  # 模拟异步操作
    return {"Hello": "World"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    await asyncio.sleep(0.5)  # 模拟异步操作
    return {"item_id": item_id, "q": q}

FastAPI依赖注入 ​

python
from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖
def get_db():
    db = "Database Connection"
    try:
        yield db
    finally:
        # 关闭数据库连接
        pass

# 使用依赖
@app.get("/items/")
def read_items(db: str = Depends(get_db)):
    return {"db": db}
from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖
def get_db():
    db = "Database Connection"
    try:
        yield db
    finally:
        # 关闭数据库连接
        pass

# 使用依赖
@app.get("/items/")
def read_items(db: str = Depends(get_db)):
    return {"db": db}

FastAPI安全性 ​

bash
pip install python-multipart python-jose[cryptography] passlib[bcrypt]
pip install python-multipart python-jose[cryptography] passlib[bcrypt]
python
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

app = FastAPI()

# 配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码加密
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2密码Bearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模拟用户数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

# 验证密码
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# 获取用户
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return user_dict

# 认证用户
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

# 创建访问令牌
def create_access_token(data: dict, expires_delta: timedelta = 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=ALGORITHM)
    return encoded_jwt

# 获取当前用户
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 = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

# 登录端点
@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护的端点
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

app = FastAPI()

# 配置
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码加密
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2密码Bearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模拟用户数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    }
}

# 验证密码
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

# 获取用户
def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return user_dict

# 认证用户
def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

# 创建访问令牌
def create_access_token(data: dict, expires_delta: timedelta = 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=ALGORITHM)
    return encoded_jwt

# 获取当前用户
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 = get_user(fake_users_db, username=username)
    if user is None:
        raise credentials_exception
    return user

# 登录端点
@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护的端点
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

Pydantic v2（FastAPI 数据验证核心） ​

FastAPI 基于 Pydantic v2（Rust 实现，验证速度提升 5-50 倍）：

python
from pydantic import BaseModel, Field, EmailStr, ValidationError
from datetime import datetime

# 定义模型
class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    age: int = Field(ge=0, le=150)
    email: EmailStr | None = None
    tags: list[str] = []
    created_at: datetime | None = None

    model_config = {  # Pydantic v2 配置方式
        "extra": "forbid",     # 禁止额外字段
        "frozen": True,        # 不可变对象
    }

# 使用
try:
    user = UserCreate(name="Alice", age=30, email="alice@example.com")
    print(user.model_dump())   # v2 替代 .dict()
    print(user.model_dump_json())  # v2 替代 .json()
except ValidationError as e:
    print(e.errors())

# 自定义验证
from pydantic import field_validator, model_validator

class Order(BaseModel):
    items: list[str]
    total: float

    @field_validator("total")
    @classmethod
    def check_total(cls, v: float) -> float:
        if v < 0:
            raise ValueError("金额不能为负")
        return v

    @model_validator(mode="after")
    def check_items(self) -> "Order":
        if not self.items and self.total > 0:
            raise ValueError("有金额但无商品")
        return self

# Pydantic v1 vs v2 对照
# .dict() → .model_dump()
# .json() → .model_dump_json()
# .schema() → .model_json_schema()
# class Config → model_config
# @validator → @field_validator
# @root_validator → @model_validator
from pydantic import BaseModel, Field, EmailStr, ValidationError
from datetime import datetime

# 定义模型
class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    age: int = Field(ge=0, le=150)
    email: EmailStr | None = None
    tags: list[str] = []
    created_at: datetime | None = None

    model_config = {  # Pydantic v2 配置方式
        "extra": "forbid",     # 禁止额外字段
        "frozen": True,        # 不可变对象
    }

# 使用
try:
    user = UserCreate(name="Alice", age=30, email="alice@example.com")
    print(user.model_dump())   # v2 替代 .dict()
    print(user.model_dump_json())  # v2 替代 .json()
except ValidationError as e:
    print(e.errors())

# 自定义验证
from pydantic import field_validator, model_validator

class Order(BaseModel):
    items: list[str]
    total: float

    @field_validator("total")
    @classmethod
    def check_total(cls, v: float) -> float:
        if v < 0:
            raise ValueError("金额不能为负")
        return v

    @model_validator(mode="after")
    def check_items(self) -> "Order":
        if not self.items and self.total > 0:
            raise ValueError("有金额但无商品")
        return self

# Pydantic v1 vs v2 对照
# .dict() → .model_dump()
# .json() → .model_dump_json()
# .schema() → .model_json_schema()
# class Config → model_config
# @validator → @field_validator
# @root_validator → @model_validator

FastAPI 进阶用法 ​

python
from fastapi import FastAPI, Query, Path, Body, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Annotated

app = FastAPI(title="进阶 API", version="2.0.0")

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路径参数校验
@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(ge=1, le=1000)],
    q: Annotated[str | None, Query(max_length=50)] = None,
):
    return {"item_id": item_id, "q": q}

# 请求体
class Item(BaseModel):
    name: str
    price: Annotated[float, Field(gt=0)]

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
    return item

# 响应模型（自动过滤敏感字段）
class UserOut(BaseModel):
    id: int
    name: str

@app.post("/users", response_model=UserOut)
def create_user(user: UserCreate):
    # 即使返回了 password，也会被自动过滤
    return {"id": 1, "name": user.name, "password": "secret"}

# 生命周期事件
@app.on_event("startup")
async def startup():
    print("服务启动")

@app.on_event("shutdown")
async def shutdown():
    print("服务关闭")
from fastapi import FastAPI, Query, Path, Body, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Annotated

app = FastAPI(title="进阶 API", version="2.0.0")

# CORS 配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# 路径参数校验
@app.get("/items/{item_id}")
def read_item(
    item_id: Annotated[int, Path(ge=1, le=1000)],
    q: Annotated[str | None, Query(max_length=50)] = None,
):
    return {"item_id": item_id, "q": q}

# 请求体
class Item(BaseModel):
    name: str
    price: Annotated[float, Field(gt=0)]

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
    return item

# 响应模型（自动过滤敏感字段）
class UserOut(BaseModel):
    id: int
    name: str

@app.post("/users", response_model=UserOut)
def create_user(user: UserCreate):
    # 即使返回了 password，也会被自动过滤
    return {"id": 1, "name": user.name, "password": "secret"}

# 生命周期事件
@app.on_event("startup")
async def startup():
    print("服务启动")

@app.on_event("shutdown")
async def shutdown():
    print("服务关闭")

FastAPI自动文档 ​

FastAPI自动生成交互式API文档，您可以通过以下URL访问：

FastAPI 项目实战（重点） ​

推荐项目结构 ​

myapi/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口，注册路由/中间件
│   ├── config.py            # 配置管理（pydantic-settings）
│   ├── database.py          # 数据库引擎和会话
│   ├── dependencies.py      # 全局依赖（get_db, get_current_user）
│   ├── models/              # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic 请求/响应模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   └── users.py
│   ├── services/            # 业务逻辑层
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── utils/               # 工具函数
│       └── security.py
├── tests/
│   ├── __init__.py
│   └── test_users.py
├── alembic/                 # 数据库迁移
├── .env
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
└── requirements.txt
myapi/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口，注册路由/中间件
│   ├── config.py            # 配置管理（pydantic-settings）
│   ├── database.py          # 数据库引擎和会话
│   ├── dependencies.py      # 全局依赖（get_db, get_current_user）
│   ├── models/              # SQLAlchemy 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic 请求/响应模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # 路由模块
│   │   ├── __init__.py
│   │   └── users.py
│   ├── services/            # 业务逻辑层
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── utils/               # 工具函数
│       └── security.py
├── tests/
│   ├── __init__.py
│   └── test_users.py
├── alembic/                 # 数据库迁移
├── .env
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
└── requirements.txt

配置管理 ​

python
# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "MyAPI"
    debug: bool = False
    database_url: str = "sqlite+aiosqlite:///./test.db"
    redis_url: str = "redis://localhost:6379"
    secret_key: str = "change-me"
    access_token_expire_minutes: int = 30

    model_config = {"env_file": ".env"}

settings = Settings()
# app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    app_name: str = "MyAPI"
    debug: bool = False
    database_url: str = "sqlite+aiosqlite:///./test.db"
    redis_url: str = "redis://localhost:6379"
    secret_key: str = "change-me"
    access_token_expire_minutes: int = 30

    model_config = {"env_file": ".env"}

settings = Settings()

数据库会话管理 ​

python
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

from app.config import settings

engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# 依赖：获取数据库会话
async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

from app.config import settings

engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

# 依赖：获取数据库会话
async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

模型定义 ​

python
# app/models/user.py
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(120), unique=True)
    age: Mapped[int | None]
# app/models/user.py
from sqlalchemy import String, Integer
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(120), unique=True)
    age: Mapped[int | None]

Pydantic Schema ​

python
# app/schemas/user.py
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

# 创建用户请求
class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    email: EmailStr
    age: int | None = Field(None, ge=0, le=150)

# 用户响应（不返回敏感字段）
class UserOut(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime | None

    model_config = {"from_attributes": True}  # 支持 ORM 模式
# app/schemas/user.py
from pydantic import BaseModel, Field, EmailStr
from datetime import datetime

# 创建用户请求
class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    email: EmailStr
    age: int | None = Field(None, ge=0, le=150)

# 用户响应（不返回敏感字段）
class UserOut(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime | None

    model_config = {"from_attributes": True}  # 支持 ORM 模式

路由实现 ​

python
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserOut

router = APIRouter(prefix="/users", tags=["用户管理"])

# 创建用户
@router.post("/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(name=data.name, email=data.email, age=data.age)
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user

# 获取用户列表（分页）
@router.get("/", response_model=list[UserOut])
async def list_users(
    skip: int = 0,
    limit: int = 10,
    db: AsyncSession = Depends(get_db),
):
    stmt = select(User).offset(skip).limit(limit)
    result = await db.execute(stmt)
    return result.scalars().all()

# 获取单个用户
@router.get("/{user_id}", response_model=UserOut)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

# 删除用户
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    await db.delete(user)
# app/routers/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserOut

router = APIRouter(prefix="/users", tags=["用户管理"])

# 创建用户
@router.post("/", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(data: UserCreate, db: AsyncSession = Depends(get_db)):
    user = User(name=data.name, email=data.email, age=data.age)
    db.add(user)
    await db.flush()
    await db.refresh(user)
    return user

# 获取用户列表（分页）
@router.get("/", response_model=list[UserOut])
async def list_users(
    skip: int = 0,
    limit: int = 10,
    db: AsyncSession = Depends(get_db),
):
    stmt = select(User).offset(skip).limit(limit)
    result = await db.execute(stmt)
    return result.scalars().all()

# 获取单个用户
@router.get("/{user_id}", response_model=UserOut)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

# 删除用户
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    await db.delete(user)

服务层（业务逻辑） ​

python
# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.schemas.user import UserCreate

class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, data: UserCreate) -> User:
        # 检查邮箱是否已存在
        result = await self.db.execute(
            select(User).where(User.email == data.email)
        )
        if result.scalar_one_or_none():
            raise ValueError("邮箱已被注册")

        user = User(name=data.name, email=data.email, age=data.age)
        self.db.add(user)
        await self.db.flush()
        await self.db.refresh(user)
        return user

    async def list(self, skip: int = 0, limit: int = 10) -> list[User]:
        result = await self.db.execute(
            select(User).offset(skip).limit(limit)
        )
        return result.scalars().all()
# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.user import User
from app.schemas.user import UserCreate

class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def create(self, data: UserCreate) -> User:
        # 检查邮箱是否已存在
        result = await self.db.execute(
            select(User).where(User.email == data.email)
        )
        if result.scalar_one_or_none():
            raise ValueError("邮箱已被注册")

        user = User(name=data.name, email=data.email, age=data.age)
        self.db.add(user)
        await self.db.flush()
        await self.db.refresh(user)
        return user

    async def list(self, skip: int = 0, limit: int = 10) -> list[User]:
        result = await self.db.execute(
            select(User).offset(skip).limit(limit)
        )
        return result.scalars().all()

全局异常处理 ​

python
# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 全局异常处理器
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)},
    )

# 统一响应格式
class APIResponse:
    @staticmethod
    def success(data=None, message="成功"):
        return {"code": 0, "message": message, "data": data}

    @staticmethod
    def error(message="失败", code=-1):
        return {"code": code, "message": message}
# app/main.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 全局异常处理器
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)},
    )

# 统一响应格式
class APIResponse:
    @staticmethod
    def success(data=None, message="成功"):
        return {"code": 0, "message": message, "data": data}

    @staticmethod
    def error(message="失败", code=-1):
        return {"code": code, "message": message}

中间件（日志/耗时/限流） ​

python
import time
import logging
from fastapi import Request

logger = logging.getLogger("uvicorn")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start

    logger.info(
        f"{request.method} {request.url.path} "
        f"{response.status_code} {elapsed:.3f}s"
    )
    response.headers["X-Process-Time"] = str(elapsed)
    return response
import time
import logging
from fastapi import Request

logger = logging.getLogger("uvicorn")

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    elapsed = time.perf_counter() - start

    logger.info(
        f"{request.method} {request.url.path} "
        f"{response.status_code} {elapsed:.3f}s"
    )
    response.headers["X-Process-Time"] = str(elapsed)
    return response

文件上传 ​

python
from fastapi import UploadFile, File
import aiofiles

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # 保存文件
    content = await file.read()
    async with aiofiles.open(f"uploads/{file.filename}", "wb") as f:
        await f.write(content)

    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

# 多个文件上传
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    return [{"filename": f.filename} for f in files]
from fastapi import UploadFile, File
import aiofiles

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    # 保存文件
    content = await file.read()
    async with aiofiles.open(f"uploads/{file.filename}", "wb") as f:
        await f.write(content)

    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

# 多个文件上传
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    return [{"filename": f.filename} for f in files]

后台任务 ​

python
from fastapi import BackgroundTasks

def send_email(email: str, body: str):
    # 发送邮件的耗时操作
    pass

@app.post("/send-notification")
async def send_notification(email: str, tasks: BackgroundTasks):
    tasks.add_task(send_email, email, "欢迎注册！")
    return {"message": "邮件将在后台发送"}
from fastapi import BackgroundTasks

def send_email(email: str, body: str):
    # 发送邮件的耗时操作
    pass

@app.post("/send-notification")
async def send_notification(email: str, tasks: BackgroundTasks):
    tasks.add_task(send_email, email, "欢迎注册！")
    return {"message": "邮件将在后台发送"}

分页工具 ​

python
from math import ceil
from pydantic import BaseModel

class PaginationParams(BaseModel):
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1, le=100)

class PaginatedResult(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    total_pages: int

def paginate(query, total: int, params: PaginationParams) -> PaginatedResult:
    return PaginatedResult(
        items=list(query),
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=ceil(total / params.page_size),
    )

# 使用
@app.get("/users")
async def list_users(
    pagination: Annotated[PaginationParams, Query()],
    db: AsyncSession = Depends(get_db),
):
    stmt = select(User).offset(
        (pagination.page - 1) * pagination.page_size
    ).limit(pagination.page_size)
    result = await db.execute(stmt)
    users = result.scalars().all()

    total_stmt = select(func.count(User.id))
    total = (await db.execute(total_stmt)).scalar()

    return paginate(users, total, pagination)
from math import ceil
from pydantic import BaseModel

class PaginationParams(BaseModel):
    page: int = Field(default=1, ge=1)
    page_size: int = Field(default=10, ge=1, le=100)

class PaginatedResult(BaseModel):
    items: list
    total: int
    page: int
    page_size: int
    total_pages: int

def paginate(query, total: int, params: PaginationParams) -> PaginatedResult:
    return PaginatedResult(
        items=list(query),
        total=total,
        page=params.page,
        page_size=params.page_size,
        total_pages=ceil(total / params.page_size),
    )

# 使用
@app.get("/users")
async def list_users(
    pagination: Annotated[PaginationParams, Query()],
    db: AsyncSession = Depends(get_db),
):
    stmt = select(User).offset(
        (pagination.page - 1) * pagination.page_size
    ).limit(pagination.page_size)
    result = await db.execute(stmt)
    users = result.scalars().all()

    total_stmt = select(func.count(User.id))
    total = (await db.execute(total_stmt)).scalar()

    return paginate(users, total, pagination)

应用入口 ​

python
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.database import engine, Base
from app.routers import users

# 生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭时：释放资源
    await engine.dispose()

app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    lifespan=lifespan,
)

# 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(users.router, prefix="/api/v1")
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.database import engine, Base
from app.routers import users

# 生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时：创建表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # 关闭时：释放资源
    await engine.dispose()

app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    lifespan=lifespan,
)

# 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(users.router, prefix="/api/v1")

测试 ​

python
# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import get_db, Base, engine, async_session

# 使用测试数据库
@pytest.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_session() as session:
        yield session
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/api/v1/users/", json={
        "name": "Alice",
        "email": "alice@example.com",
        "age": 30,
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"

@pytest.mark.asyncio
async def test_list_users(client):
    response = await client.get("/api/v1/users/")
    assert response.status_code == 200
# tests/test_users.py
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.database import get_db, Base, engine, async_session

# 使用测试数据库
@pytest.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with async_session() as session:
        yield session
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture
async def client(db_session):
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client):
    response = await client.post("/api/v1/users/", json={
        "name": "Alice",
        "email": "alice@example.com",
        "age": 30,
    })
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "Alice"

@pytest.mark.asyncio
async def test_list_users(client):
    response = await client.get("/api/v1/users/")
    assert response.status_code == 200

Docker 部署 ​

dockerfile
# Dockerfile
FROM python:3.12-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", "8000"]
# Dockerfile
FROM python:3.12-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", "8000"]
yaml
# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/myapp
    depends_on:
      - db

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: myapp
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:
# docker-compose.yml
version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql+asyncpg://user:pass@db:5432/myapp
    depends_on:
      - db

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: myapp
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

常用依赖清单 ​

txt
# requirements.txt / pyproject.toml dependencies
fastapi>=0.109     # Web 框架
uvicorn[standard]  # ASGI 服务器
sqlalchemy>=2.0    # ORM
asyncpg            # PostgreSQL 异步驱动
alembic            # 数据库迁移
pydantic>=2.0      # 数据验证
pydantic-settings  # 环境变量管理
python-jose        # JWT 编解码
passlib[bcrypt]    # 密码加密
python-multipart   # 表单/文件上传
aiofiles           # 异步文件操作
httpx              # HTTP 客户端（测试/调用外部API）
celery[redis]      # 任务队列（可选）
redis              # 缓存（可选）
# requirements.txt / pyproject.toml dependencies
fastapi>=0.109     # Web 框架
uvicorn[standard]  # ASGI 服务器
sqlalchemy>=2.0    # ORM
asyncpg            # PostgreSQL 异步驱动
alembic            # 数据库迁移
pydantic>=2.0      # 数据验证
pydantic-settings  # 环境变量管理
python-jose        # JWT 编解码
passlib[bcrypt]    # 密码加密
python-multipart   # 表单/文件上传
aiofiles           # 异步文件操作
httpx              # HTTP 客户端（测试/调用外部API）
celery[redis]      # 任务队列（可选）
redis              # 缓存（可选）

框架选择建议 ​

  1. Django：

    • 适合需要快速开发大型Web应用的团队
    • 内置功能丰富，减少第三方依赖
    • 适合需要Admin后台的项目

  2. Flask：

    • 适合小型项目或API服务
    • 适合学习Web开发的初学者
    • 适合需要高度定制化的项目

  3. FastAPI：

    • 适合构建高性能的RESTful API
    • 适合需要异步支持的项目
    • 适合微服务架构
    • 适合需要自动文档的项目

部署建议 ​

  1. Django：

    • WSGI服务器：Gunicorn + Nginx
    • 容器化：Docker + Kubernetes
    • 云平台：AWS Elastic Beanstalk, Google App Engine

  2. Flask：

    • WSGI服务器：Gunicorn + Nginx
    • 容器化：Docker + Kubernetes
    • 云平台：Heroku, AWS Lambda

  3. FastAPI：

    • ASGI服务器：Uvicorn + Gunicorn + Nginx
    • 容器化：Docker + Kubernetes
    • 云平台：AWS Lambda, Google Cloud Functions

总结 ​

Python提供了多种优秀的Web框架，选择合适的框架取决于项目的需求、团队的经验和项目的规模。无论选择哪种框架，都应该遵循最佳实践，编写清晰、可维护的代码，并确保应用的安全性和性能。

  • Django适合需要全功能框架的大型项目
  • Flask适合需要灵活性的小型项目和API服务
  • FastAPI适合需要高性能和自动文档的现代API开发

通过学习和掌握这些框架，您可以构建各种类型的Web应用，从简单的API到复杂的企业级应用。

Copyright © 2024-present xiaocao