Python Web 开发高级应用指南

1. Web 开发基础

Python 的 Web 开发主要通过 Flask、Django 和 FastAPI 等框架实现。

# Flask 基础
from flask import Flask

app = Flask(__name__)

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

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

2. Flask 高级应用

2.1 蓝图

from flask import Flask, Blueprint

app = Flask(__name__)

# 创建蓝图
api_bp = Blueprint('api', __name__, url_prefix='/api')

# 注册路由
@api_bp.route('/users')
def get_users():
    return {'users': ['Alice', 'Bob', 'Charlie']}

@api_bp.route('/users/<int:user_id>')
def get_user(user_id):
    return {'user_id': user_id, 'name': f'User {user_id}'}

# 注册蓝图
app.register_blueprint(api_bp)

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

2.2 中间件

from flask import Flask, request

app = Flask(__name__)

# 中间件
@app.before_request
def before_request():
    print(f"Request: {request.method} {request.url}")

@app.after_request
def after_request(response):
    print(f"Response status: {response.status_code}")
    return response

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

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

2.3 数据库集成

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

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

# 定义模型
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(80), nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'email': self.email
        }

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

# 路由
@app.route('/users', methods=['GET', 'POST'])
def users():
    if request.method == 'GET':
        users = User.query.all()
        return jsonify([user.to_dict() for user in users])
    elif request.method == 'POST':
        data = request.json
        user = User(name=data['name'], email=data['email'])
        db.session.add(user)
        db.session.commit()
        return jsonify(user.to_dict()), 201

@app.route('/users/<int:user_id>', methods=['GET', 'PUT', 'DELETE'])
def user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({'error': 'User not found'}), 404
    
    if request.method == 'GET':
        return jsonify(user.to_dict())
    elif request.method == 'PUT':
        data = request.json
        user.name = data['name']
        user.email = data['email']
        db.session.commit()
        return jsonify(user.to_dict())
    elif request.method == 'DELETE':
        db.session.delete(user)
        db.session.commit()
        return jsonify({'message': 'User deleted'})

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

3. Django 高级应用

3.1 视图类

# views.py
from django.views import View
from django.http import JsonResponse
from .models import User

class UserListView(View):
    def get(self, request):
        users = User.objects.all()
        return JsonResponse([user.to_dict() for user in users], safe=False)
    
    def post(self, request):
        data = request.POST
        user = User(name=data['name'], email=data['email'])
        user.save()
        return JsonResponse(user.to_dict(), status=201)

class UserDetailView(View):
    def get(self, request, user_id):
        try:
            user = User.objects.get(id=user_id)
            return JsonResponse(user.to_dict())
        except User.DoesNotExist:
            return JsonResponse({'error': 'User not found'}, status=404)
    
    def put(self, request, user_id):
        try:
            user = User.objects.get(id=user_id)
            data = request.POST
            user.name = data['name']
            user.email = data['email']
            user.save()
            return JsonResponse(user.to_dict())
        except User.DoesNotExist:
            return JsonResponse({'error': 'User not found'}, status=404)
    
    def delete(self, request, user_id):
        try:
            user = User.objects.get(id=user_id)
            user.delete()
            return JsonResponse({'message': 'User deleted'})
        except User.DoesNotExist:
            return JsonResponse({'error': 'User not found'}, status=404)

# urls.py
from django.urls import path
from .views import UserListView, UserDetailView

urlpatterns = [
    path('users/', UserListView.as_view()),
    path('users/<int:user_id>/', UserDetailView.as_view()),
]

3.2 中间件

# middleware.py
class LoggingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        print(f"Request: {request.method} {request.path}")
        response = self.get_response(request)
        print(f"Response status: {response.status_code}")
        return response

# settings.py
MIDDLEWARE = [
    'myapp.middleware.LoggingMiddleware',
    # 其他中间件
]

3.3 信号

# signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import User

@receiver(post_save, sender=User)
def user_saved(sender, instance, created, **kwargs):
    if created:
        print(f"New user created: {instance.name}")
    else:
        print(f"User updated: {instance.name}")

# apps.py
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'myapp'
    
    def ready(self):
        import myapp.signals

4. FastAPI 高级应用

4.1 路径操作

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

app = FastAPI()

# 数据模型
class UserBase(BaseModel):
    name: str
    email: str

class UserCreate(UserBase):
    pass

class User(UserBase):
    id: int
    
    class Config:
        from_attributes = True

# 模拟数据库
users = []

# 路径操作
@app.post('/users/', response_model=User)
def create_user(user: UserCreate):
    user_id = len(users) + 1
    new_user = User(id=user_id, **user.dict())
    users.append(new_user)
    return new_user

@app.get('/users/', response_model=List[User])
def get_users():
    return users

@app.get('/users/{user_id}', response_model=User)
def get_user(user_id: int):
    for user in users:
        if user.id == user_id:
            return user
    raise HTTPException(status_code=404, detail='User not found')

@app.put('/users/{user_id}', response_model=User)
def update_user(user_id: int, user: UserCreate):
    for i, existing_user in enumerate(users):
        if existing_user.id == user_id:
            updated_user = User(id=user_id, **user.dict())
            users[i] = updated_user
            return updated_user
    raise HTTPException(status_code=404, detail='User not found')

@app.delete('/users/{user_id}')
def delete_user(user_id: int):
    for i, user in enumerate(users):
        if user.id == user_id:
            users.pop(i)
            return {'message': 'User deleted'}
    raise HTTPException(status_code=404, detail='User not found')

4.2 依赖注入

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

app = FastAPI()

# 数据模型
class UserBase(BaseModel):
    name: str
    email: str

class UserCreate(UserBase):
    pass

class User(UserBase):
    id: int
    
    class Config:
        from_attributes = True

# 模拟数据库
users = []

# 依赖项
def get_user_by_id(user_id: int):
    for user in users:
        if user.id == user_id:
            return user
    raise HTTPException(status_code=404, detail='User not found')

# 路径操作
@app.post('/users/', response_model=User)
def create_user(user: UserCreate):
    user_id = len(users) + 1
    new_user = User(id=user_id, **user.dict())
    users.append(new_user)
    return new_user

@app.get('/users/', response_model=List[User])
def get_users():
    return users

@app.get('/users/{user_id}', response_model=User)
def get_user(user: User = Depends(get_user_by_id)):
    return user

@app.put('/users/{user_id}', response_model=User)
def update_user(user_id: int, user: UserCreate, existing_user: User = Depends(get_user_by_id)):
    for i, u in enumerate(users):
        if u.id == user_id:
            updated_user = User(id=user_id, **user.dict())
            users[i] = updated_user
            return updated_user
    raise HTTPException(status_code=404, detail='User not found')

@app.delete('/users/{user_id}')
def delete_user(user_id: int, existing_user: User = Depends(get_user_by_id)):
    for i, user in enumerate(users):
        if user.id == user_id:
            users.pop(i)
            return {'message': 'User deleted'}
    raise HTTPException(status_code=404, detail='User not found')

4.3 安全

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional
import jwt
from datetime import datetime, timedelta

app = FastAPI()

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

# 安全
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 数据模型
class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

class Token(BaseModel):
    access_token: str
    token_type: str

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

# 模拟用户数据库
users_db = {
    "alice": {
        "username": "alice",
        "email": "alice@example.com",
        "full_name": "Alice Smith",
        "disabled": False,
        "hashed_password": "fakehashedsecret"
    },
    "bob": {
        "username": "bob",
        "email": "bob@example.com",
        "full_name": "Bob Brown",
        "disabled": True,
        "hashed_password": "fakehashedsecret2"
    }
}

# 工具函数
def verify_password(plain_password, hashed_password):
    return plain_password + "secret" == hashed_password

def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**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: Optional[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
        token_data = TokenData(username=username)
    except jwt.PyJWTError:
        raise credentials_exception
    user = get_user(users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

# 路径操作
@app.post("/token", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(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", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

5. 实际应用场景

5.1 RESTful API

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

app = FastAPI()

# 数据模型
class ItemBase(BaseModel):
    name: str
    description: str
    price: float
    tax: float = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    
    class Config:
        from_attributes = True

# 模拟数据库
items = []

# 路径操作
@app.post('/items/', response_model=Item)
def create_item(item: ItemCreate):
    item_id = len(items) + 1
    new_item = Item(id=item_id, **item.dict())
    items.append(new_item)
    return new_item

@app.get('/items/', response_model=List[Item])
def get_items():
    return items

@app.get('/items/{item_id}', response_model=Item)
def get_item(item_id: int):
    for item in items:
        if item.id == item_id:
            return item
    raise HTTPException(status_code=404, detail='Item not found')

@app.put('/items/{item_id}', response_model=Item)
def update_item(item_id: int, item: ItemCreate):
    for i, existing_item in enumerate(items):
        if existing_item.id == item_id:
            updated_item = Item(id=item_id, **item.dict())
            items[i] = updated_item
            return updated_item
    raise HTTPException(status_code=404, detail='Item not found')

@app.delete('/items/{item_id}')
def delete_item(item_id: int):
    for i, item in enumerate(items):
        if item.id == item_id:
            items.pop(i)
            return {'message': 'Item deleted'}
    raise HTTPException(status_code=404, detail='Item not found')

5.2 身份验证

from flask import Flask, request, jsonify, make_response
from flask_jwt_extended import JWTManager, create_access_token, jwt_required, get_jwt_identity
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'your-secret-key'
jwt = JWTManager(app)

# 模拟用户数据库
users = {
    'alice': {
        'username': 'alice',
        'password': generate_password_hash('secret'),
        'role': 'admin'
    },
    'bob': {
        'username': 'bob',
        'password': generate_password_hash('secret2'),
        'role': 'user'
    }
}

# 路由
@app.route('/login', methods=['POST'])
def login():
    data = request.json
    username = data.get('username')
    password = data.get('password')
    
    if not username or not password:
        return jsonify({'error': 'Username and password required'}), 400
    
    user = users.get(username)
    if not user or not check_password_hash(user['password'], password):
        return jsonify({'error': 'Invalid username or password'}), 401
    
    access_token = create_access_token(identity=username)
    return jsonify({'access_token': access_token}), 200

@app.route('/protected', methods=['GET'])
@jwt_required()
def protected():
    current_user = get_jwt_identity()
    return jsonify({'message': f'Hello, {current_user}!'}), 200

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

5.3 数据验证

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

app = FastAPI()

# 数据模型
class UserBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    email: str = Field(..., regex=r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')
    age: int = Field(..., ge=1, le=120)

class UserCreate(UserBase):
    password: str = Field(..., min_length=6)

class User(UserBase):
    id: int
    
    class Config:
        from_attributes = True

# 模拟数据库
users = []

# 路径操作
@app.post('/users/', response_model=User)
def create_user(user: UserCreate):
    # 检查邮箱是否已存在
    for existing_user in users:
        if existing_user.email == user.email:
            raise HTTPException(status_code=400, detail='Email already registered')
    
    user_id = len(users) + 1
    new_user = User(id=user_id, name=user.name, email=user.email, age=user.age)
    users.append(new_user)
    return new_user

@app.get('/users/', response_model=List[User])
def get_users():
    return users

6. 最佳实践

  1. 选择合适的框架：根据项目需求选择合适的 Web 框架。
  2. RESTful API 设计：遵循 RESTful API 设计原则，使用适当的 HTTP 方法和状态码。
  3. 数据验证：使用 Pydantic 或 Django 表单进行数据验证。
  4. 安全：实现适当的身份验证和授权机制。
  5. 数据库设计：设计合理的数据库 schema，使用 ORM 进行数据库操作。
  6. 错误处理：统一处理错误，返回友好的错误信息。
  7. 测试：编写单元测试和集成测试，确保代码质量。
  8. 部署：使用适当的部署策略，如容器化、负载均衡等。

7. 总结

Python 的 Web 开发框架提供了丰富的功能，从基本的路由和视图到高级的中间件、依赖注入和安全机制。通过掌握这些高级应用，我们可以构建更加健壮、安全、可维护的 Web 应用。

在实际应用中，Web 开发可以用于构建 RESTful API、身份验证系统、数据验证等多种场景，满足不同的业务需求。

希望本文对你理解和应用 Python Web 开发有所帮助！

# rust # github # python
Logo
CSDN-OPC开发者社区

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

更多推荐

cover

从零构建 Java 智能体 RAG 系统：Milvus 向量数据库实战指南

avatar CSDN-OPC开发者社区
cover

DeepSeek-Harness-入门指南

avatar CSDN-OPC开发者社区
cover

提升办公效率：OpenClaw 本地自动化 AI 工具搭建实战教程

avatar CSDN-OPC开发者社区