当前位置：首页 > ds >正文

【Python 后端框架】总结

ds 2025/9/1 20:22:11

Python后端框架全面总结

概述

Python作为一门简洁、易学、功能强大的编程语言，在后端开发领域有着广泛的应用。随着Web技术的不断发展，Python后端框架也在不断演进，从传统的同步框架发展到现代的异步框架，为开发者提供了丰富的选择。

主流框架介绍

Django

发展历程

2005年：由Adrian Holovaty和Simon Willison创建
2006年：发布1.0版本
2008年：被Google采用，获得广泛关注
2012年：发布1.4版本，引入数据库迁移
2017年：发布2.0版本，支持Python 3.7+
2020年：发布3.0版本，支持异步视图
2023年：发布4.2版本，持续优化性能

核心特性

全栈框架：包含ORM、模板引擎、认证系统、管理后台等
MVT架构：Model-View-Template，类似MVC
内置安全：CSRF保护、XSS防护、SQL注入防护
自动管理后台：基于模型自动生成管理界面
丰富的中间件：缓存、会话、认证等
国际化支持：多语言、时区处理

工作原理

请求 → URL路由 → 中间件 → 视图 → 模型 → 数据库
响应 ← 模板渲染 ← 视图 ← 模型 ← 数据库

代码案例

1. 项目结构

myproject/
├── manage.py
├── myproject/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── blog/
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── models.py
│   ├── views.py
│   ├── urls.py
│   └── templates/
│       └── blog/
│           ├── post_list.html
│           └── post_detail.html
└── requirements.txt

2. 模型定义 (models.py)

from django.db import models
from django.contrib.auth.models import User
from django.utils import timezoneclass Category(models.Model):name = models.CharField(max_length=100)slug = models.SlugField(unique=True)description = models.TextField(blank=True)class Meta:verbose_name_plural = "categories"def __str__(self):return self.nameclass Post(models.Model):STATUS_CHOICES = (('draft', '草稿'),('published', '已发布'),)title = models.CharField(max_length=200)slug = models.SlugField(max_length=200, unique=True)author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='blog_posts')content = models.TextField()category = models.ForeignKey(Category, on_delete=models.CASCADE, related_name='posts')status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='draft')created = models.DateTimeField(auto_now_add=True)updated = models.DateTimeField(auto_now=True)published = models.DateTimeField(default=timezone.now)class Meta:ordering = ['-published']def __str__(self):return self.titledef get_absolute_url(self):return reverse('blog:post_detail', args=[self.slug])

3. 视图定义 (views.py)

from django.shortcuts import render, get_object_or_404
from django.views.generic import ListView, DetailView
from django.contrib.auth.mixins import LoginRequiredMixin
from .models import Post, Categoryclass PostListView(ListView):model = Posttemplate_name = 'blog/post_list.html'context_object_name = 'posts'paginate_by = 10def get_queryset(self):return Post.objects.filter(status='published')def get_context_data(self, **kwargs):context = super().get_context_data(**kwargs)context['categories'] = Category.objects.all()return contextclass PostDetailView(DetailView):model = Posttemplate_name = 'blog/post_detail.html'context_object_name = 'post'def get_queryset(self):return Post.objects.filter(status='published')# 函数式视图示例
def category_posts(request, category_slug):category = get_object_or_404(Category, slug=category_slug)posts = Post.objects.filter(category=category, status='published')return render(request, 'blog/category_posts.html', {'category': category,'posts': posts})# 异步视图示例 (Django 3.0+)
import asyncio
from django.http import JsonResponseasync def async_api_view(request):# 模拟异步操作await asyncio.sleep(0.1)data = {'message': '异步API响应','timestamp': timezone.now().isoformat()}return JsonResponse(data)

4. URL配置 (urls.py)

from django.urls import path, include
from . import viewsapp_name = 'blog'urlpatterns = [path('', views.PostListView.as_view(), name='post_list'),path('post/<slug:slug>/', views.PostDetailView.as_view(), name='post_detail'),path('category/<slug:category_slug>/', views.category_posts, name='category_posts'),path('api/async/', views.async_api_view, name='async_api'),
]

5. 模板示例 (post_list.html)

{% extends 'base.html' %}
{% load static %}{% block title %}博客文章列表{% endblock %}{% block content %}
<div class="container"><div class="row"><div class="col-md-8">{% for post in posts %}<article class="post-preview"><h2><a href="{{ post.get_absolute_url }}">{{ post.title }}</a></h2><p class="post-meta">作者: {{ post.author.username }} | 发布时间: {{ post.published|date:"Y-m-d H:i" }} |分类: {{ post.category.name }}</p><p class="post-excerpt">{{ post.content|truncatewords:50 }}</p></article>{% empty %}<p>暂无文章</p>{% endfor %}{% if is_paginated %}<nav aria-label="分页导航"><ul class="pagination">{% if page_obj.has_previous %}<li class="page-item"><a class="page-link" href="?page={{ page_obj.previous_page_number }}">上一页</a></li>{% endif %}{% for num in page_obj.paginator.page_range %}<li class="page-item {% if page_obj.number == num %}active{% endif %}"><a class="page-link" href="?page={{ num }}">{{ num }}</a></li>{% endfor %}{% if page_obj.has_next %}<li class="page-item"><a class="page-link" href="?page={{ page_obj.next_page_number }}">下一页</a></li>{% endif %}</ul></nav>{% endif %}</div><div class="col-md-4"><div class="sidebar"><h3>分类</h3><ul class="list-unstyled">{% for category in categories %}<li><a href="{% url 'blog:category_posts' category.slug %}">{{ category.name }}</a></li>{% endfor %}</ul></div></div></div>
</div>
{% endblock %}

6. 管理后台配置 (admin.py)

from django.contrib import admin
from .models import Category, Post@admin.register(Category)
class CategoryAdmin(admin.ModelAdmin):list_display = ['name', 'slug', 'description']prepopulated_fields = {'slug': ('name',)}search_fields = ['name']@admin.register(Post)
class PostAdmin(admin.ModelAdmin):list_display = ['title', 'slug', 'author', 'category', 'status', 'created', 'published']list_filter = ['status', 'created', 'published', 'category', 'author']search_fields = ['title', 'content']prepopulated_fields = {'slug': ('title',)}raw_id_fields = ['author']date_hierarchy = 'published'ordering = ['status', 'published']fieldsets = (('基本信息', {'fields': ('title', 'slug', 'author', 'category')}),('内容', {'fields': ('content',)}),('发布设置', {'fields': ('status', 'published')}),)

7. 表单处理

from django import forms
from .models import Post, Commentclass PostForm(forms.ModelForm):class Meta:model = Postfields = ['title', 'content', 'category', 'status']widgets = {'title': forms.TextInput(attrs={'class': 'form-control'}),'content': forms.Textarea(attrs={'class': 'form-control', 'rows': 10}),'category': forms.Select(attrs={'class': 'form-control'}),'status': forms.Select(attrs={'class': 'form-control'}),}class CommentForm(forms.ModelForm):class Meta:model = Commentfields = ['content']widgets = {'content': forms.Textarea(attrs={'class': 'form-control', 'rows': 3, 'placeholder': '写下你的评论...'})}# 在视图中使用表单
from django.contrib.auth.decorators import login_required
from django.shortcuts import redirect@login_required
def create_post(request):if request.method == 'POST':form = PostForm(request.POST)if form.is_valid():post = form.save(commit=False)post.author = request.userpost.save()return redirect('blog:post_detail', slug=post.slug)else:form = PostForm()return render(request, 'blog/post_form.html', {'form': form})

8. 中间件示例

# custom_middleware.py
import time
from django.utils.deprecation import MiddlewareMixinclass RequestTimingMiddleware(MiddlewareMixin):def process_request(self, request):request.start_time = time.time()def process_response(self, request, response):if hasattr(request, 'start_time'):duration = time.time() - request.start_timeresponse['X-Request-Time'] = str(duration)return responseclass UserActivityMiddleware(MiddlewareMixin):def process_request(self, request):if request.user.is_authenticated:# 记录用户活动request.user.last_activity = timezone.now()request.user.save(update_fields=['last_activity'])

9. 信号处理

# signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from .models import Post@receiver(post_save, sender=Post)
def post_saved(sender, instance, created, **kwargs):if created:# 新文章发布，清除缓存cache.delete('recent_posts')cache.delete('post_count')# 更新搜索索引# update_search_index(instance)@receiver(post_delete, sender=Post)
def post_deleted(sender, instance, **kwargs):# 文章删除，清除缓存cache.delete('recent_posts')cache.delete('post_count')

10. 测试代码

# tests.py
from django.test import TestCase, Client
from django.contrib.auth.models import User
from django.urls import reverse
from .models import Category, Postclass PostModelTest(TestCase):def setUp(self):self.user = User.objects.create_user(username='testuser', password='testpass')self.category = Category.objects.create(name='测试分类', slug='test-category')self.post = Post.objects.create(title='测试文章',slug='test-post',author=self.user,content='这是测试内容',category=self.category,status='published')def test_post_creation(self):self.assertEqual(self.post.title, '测试文章')self.assertEqual(self.post.author.username, 'testuser')self.assertEqual(self.post.status, 'published')def test_post_str_method(self):self.assertEqual(str(self.post), '测试文章')class PostViewTest(TestCase):def setUp(self):self.client = Client()self.user = User.objects.create_user(username='testuser', password='testpass')self.category = Category.objects.create(name='测试分类', slug='test-category')self.post = Post.objects.create(title='测试文章',slug='test-post',author=self.user,content='这是测试内容',category=self.category,status='published')def test_post_list_view(self):response = self.client.get(reverse('blog:post_list'))self.assertEqual(response.status_code, 200)self.assertContains(response, '测试文章')def test_post_detail_view(self):response = self.client.get(reverse('blog:post_detail', args=['test-post']))self.assertEqual(response.status_code, 200)self.assertContains(response, '测试文章')

优势

开发效率高，开箱即用
文档完善，社区活跃
安全性强，内置多种防护机制
适合快速原型开发
企业级应用支持好

劣势

学习曲线陡峭
灵活性相对较低
性能在大型应用中可能成为瓶颈
过度设计，可能包含不需要的功能

Flask

发展历程

2010年：由Armin Ronacher创建
2011年：发布0.1版本
2012年：发布0.8版本，引入蓝图
2015年：发布0.10版本，改进错误处理
2018年：发布1.0版本，API稳定
2022年：发布2.3版本，持续优化

核心特性

微框架：核心简单，扩展性强
Werkzeug：基于WSGI工具库
Jinja2模板：灵活强大的模板引擎
蓝图系统：模块化应用组织
扩展生态：丰富的第三方扩展
轻量级：核心代码简洁

工作原理

请求 → WSGI → Flask应用 → 路由 → 视图函数 → 响应

代码案例

1. 应用结构

myapp/
├── app.py
├── config.py
├── requirements.txt
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── views.py
│   ├── forms.py
│   ├── templates/
│   │   ├── base.html
│   │   ├── index.html
│   │   └── posts/
│   │       ├── list.html
│   │       └── detail.html
│   ├── static/
│   │   ├── css/
│   │   ├── js/
│   │   └── images/
│   └── __init__.py
└── migrations/

2. 应用工厂模式 (app/init.py)

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from config import Configdb = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
mail = Mail()def create_app(config_class=Config):app = Flask(__name__)app.config.from_object(config_class)# 初始化扩展db.init_app(app)migrate.init_app(app, db)login_manager.init_app(app)mail.init_app(app)# 配置登录管理器login_manager.login_view = 'auth.login'login_manager.login_message = '请先登录'# 注册蓝图from app.main import bp as main_bpapp.register_blueprint(main_bp)from app.auth import bp as auth_bpapp.register_blueprint(auth_bp, url_prefix='/auth')from app.posts import bp as posts_bpapp.register_blueprint(posts_bp, url_prefix='/posts')return appfrom app import models

3. 配置管理 (config.py)

import os
from datetime import timedeltaclass Config:SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard-to-guess-string'SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \'sqlite:///' + os.path.join(os.path.abspath(os.path.dirname(__file__)), 'app.db')SQLALCHEMY_TRACK_MODIFICATIONS = False# 邮件配置MAIL_SERVER = os.environ.get('MAIL_SERVER')MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587)MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']MAIL_USERNAME = os.environ.get('MAIL_USERNAME')MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')# 分页配置POSTS_PER_PAGE = 10# 缓存配置CACHE_TYPE = 'simple'CACHE_DEFAULT_TIMEOUT = 300class DevelopmentConfig(Config):DEBUG = TrueSQLALCHEMY_ECHO = Trueclass ProductionConfig(Config):DEBUG = Falseclass TestingConfig(Config):TESTING = TrueSQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'config = {'development': DevelopmentConfig,'production': ProductionConfig,'testing': TestingConfig,'default': DevelopmentConfig
}

4. 数据模型 (app/models.py)

from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin
from app import db, login_manager@login_manager.user_loader
def load_user(id):return User.query.get(int(id))class User(UserMixin, db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(64), unique=True, nullable=False)email = db.Column(db.String(120), unique=True, nullable=False)password_hash = db.Column(db.String(128))about_me = db.Column(db.String(140))last_seen = db.Column(db.DateTime, default=datetime.utcnow)posts = db.relationship('Post', backref='author', lazy='dynamic')def set_password(self, password):self.password_hash = generate_password_hash(password)def check_password(self, password):return check_password_hash(self.password_hash, password)def __repr__(self):return f'<User {self.username}>'class Post(db.Model):id = db.Column(db.Integer, primary_key=True)title = db.Column(db.String(200), nullable=False)content = db.Column(db.Text, nullable=False)timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)def __repr__(self):return f'<Post {self.title}>'class Category(db.Model):id = db.Column(db.Integer, primary_key=True)name = db.Column(db.String(50), unique=True, nullable=False)posts = db.relationship('Post', backref='category', lazy='dynamic')def __repr__(self):return f'<Category {self.name}>'

5. 视图函数 (app/views.py)

from flask import render_template, flash, redirect, url_for, request, jsonify, current_app
from flask_login import login_required, current_user, login_user, logout_user
from app import db
from app.models import User, Post, Category
from app.forms import LoginForm, RegistrationForm, PostForm
from app.main import bp
from flask import abort@bp.route('/')
@bp.route('/index')
def index():page = request.args.get('page', 1, type=int)posts = Post.query.order_by(Post.timestamp.desc()).paginate(page, current_app.config['POSTS_PER_PAGE'], False)return render_template('index.html', posts=posts.items, pagination=posts)@bp.route('/user/<username>')
@login_required
def user(username):user = User.query.filter_by(username=username).first_or_404()page = request.args.get('page', 1, type=int)posts = user.posts.order_by(Post.timestamp.desc()).paginate(page, current_app.config['POSTS_PER_PAGE'], False)return render_template('user.html', user=user, posts=posts.items)@bp.route('/post/new', methods=['GET', 'POST'])
@login_required
def new_post():form = PostForm()if form.validate_on_submit():post = Post(title=form.title.data, content=form.content.data, author=current_user)db.session.add(post)db.session.commit()flash('文章发布成功！')return redirect(url_for('main.index'))return render_template('post_form.html', title='发布文章', form=form)@bp.route('/post/<int:id>')
def post(id):post = Post.query.get_or_404(id)return render_template('post.html', post=post)@bp.route('/post/<int:id>/edit', methods=['GET', 'POST'])
@login_required
def edit_post(id):post = Post.query.get_or_404(id)if post.author != current_user:abort(403)form = PostForm()if form.validate_on_submit():post.title = form.title.datapost.content = form.content.datadb.session.commit()flash('文章更新成功！')return redirect(url_for('main.post', id=post.id))elif request.method == 'GET':form.title.data = post.titleform.content.data = post.contentreturn render_template('post_form.html', title='编辑文章', form=form)# API端点示例
@bp.route('/api/posts')
def api_posts():page = request.args.get('page', 1, type=int)per_page = request.args.get('per_page', 10, type=int)posts = Post.query.order_by(Post.timestamp.desc()).paginate(page, per_page, False)return jsonify({'posts': [{'id': post.id,'title': post.title,'content': post.content,'author': post.author.username,'timestamp': post.timestamp.isoformat()} for post in posts.items],'total': posts.total,'pages': posts.pages,'current_page': page})@bp.route('/api/posts/<int:id>')
def api_post(id):post = Post.query.get_or_404(id)return jsonify({'id': post.id,'title': post.title,'content': post.content,'author': post.author.username,'timestamp': post.timestamp.isoformat()})

6. 表单处理 (app/forms.py)

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, BooleanField, SubmitField, TextAreaField
from wtforms.validators import DataRequired, Email, EqualTo, Length, ValidationError
from app.models import Userclass LoginForm(FlaskForm):username = StringField('用户名', validators=[DataRequired()])password = PasswordField('密码', validators=[DataRequired()])remember_me = BooleanField('记住我')submit = SubmitField('登录')class RegistrationForm(FlaskForm):username = StringField('用户名', validators=[DataRequired(), Length(min=2, max=20)])email = StringField('邮箱', validators=[DataRequired(), Email()])password = PasswordField('密码', validators=[DataRequired()])confirm_password = PasswordField('确认密码', validators=[DataRequired(), EqualTo('password')])submit = SubmitField('注册')def validate_username(self, username):user = User.query.filter_by(username=username.data).first()if user:raise ValidationError('用户名已被使用')def validate_email(self, email):user = User.query.filter_by(email=email.data).first()if user:raise ValidationError('邮箱已被注册')class PostForm(FlaskForm):title = StringField('标题', validators=[DataRequired(), Length(min=1, max=200)])content = TextAreaField('内容', validators=[DataRequired()])submit = SubmitField('发布')class SearchForm(FlaskForm):q = StringField('搜索', validators=[DataRequired()])submit = SubmitField('搜索')

7. 蓝图组织 (app/main/init.py)

from flask import Blueprintbp = Blueprint('main', __name__)from app.main import routes

8. 模板示例 (app/templates/base.html)

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>{% block title %}{% endblock %} - Flask博客</title><link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css"><link rel="stylesheet" href="{{ url_for('static', filename='css/style.css') }}">
</head>
<body><nav class="navbar navbar-expand-lg navbar-dark bg-dark"><div class="container"><a class="navbar-brand" href="{{ url_for('main.index') }}">Flask博客</a><button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarNav"><span class="navbar-toggler-icon"></span></button><div class="collapse navbar-collapse" id="navbarNav"><ul class="navbar-nav me-auto"><li class="nav-item"><a class="nav-link" href="{{ url_for('main.index') }}">首页</a></li>{% if current_user.is_authenticated %}<li class="nav-item"><a class="nav-link" href="{{ url_for('main.new_post') }}">发布文章</a></li>{% endif %}</ul><ul class="navbar-nav">{% if current_user.is_anonymous %}<li class="nav-item"><a class="nav-link" href="{{ url_for('auth.login') }}">登录</a></li><li class="nav-item"><a class="nav-link" href="{{ url_for('auth.register') }}">注册</a></li>{% else %}<li class="nav-item dropdown"><a class="nav-link dropdown-toggle" href="#" id="navbarDropdown" role="button" data-bs-toggle="dropdown">{{ current_user.username }}</a><ul class="dropdown-menu"><li><a class="dropdown-item" href="{{ url_for('main.user', username=current_user.username) }}">个人资料</a></li><li><hr class="dropdown-divider"></li><li><a class="dropdown-item" href="{{ url_for('auth.logout') }}">退出</a></li></ul></li>{% endif %}</ul></div></div></nav><main class="container mt-4">{% with messages = get_flashed_messages() %}{% if messages %}{% for message in messages %}<div class="alert alert-info alert-dismissible fade show" role="alert">{{ message }}<button type="button" class="btn-close" data-bs-dismiss="alert"></button></div>{% endfor %}{% endif %}{% endwith %}{% block content %}{% endblock %}</main><script src="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/js/bootstrap.bundle.min.js"></script><script src="{{ url_for('static', filename='js/main.js') }}"></script>
</body>
</html>

9. 错误处理

from flask import render_template
from app import db
from app.main import bp@bp.app_errorhandler(404)
def not_found_error(error):return render_template('errors/404.html'), 404@bp.app_errorhandler(500)
def internal_error(error):db.session.rollback()return render_template('errors/500.html'), 500@bp.app_errorhandler(403)
def forbidden_error(error):return render_template('errors/403.html'), 403

10. 中间件和装饰器

from functools import wraps
from flask import request, jsonify, g
import timedef timing_middleware():g.start_time = time.time()def timing_after_request(response):if hasattr(g, 'start_time'):duration = time.time() - g.start_timeresponse.headers['X-Request-Time'] = str(duration)return responsedef require_api_key(f):@wraps(f)def decorated_function(*args, **kwargs):api_key = request.headers.get('X-API-Key')if not api_key or api_key != 'your-secret-key':return jsonify({'error': 'Invalid API key'}), 401return f(*args, **kwargs)return decorated_functiondef rate_limit(limit=100, window=3600):def decorator(f):@wraps(f)def decorated_function(*args, **kwargs):# 简单的速率限制实现client_ip = request.remote_addr# 这里应该使用Redis等缓存来存储请求计数return f(*args, **kwargs)return decorated_functionreturn decorator# 使用装饰器
@bp.route('/api/limited')
@require_api_key
@rate_limit(limit=10, window=60)
def limited_api():return jsonify({'message': 'Rate limited API'})

11. 测试代码

import unittest
from app import create_app, db
from app.models import User, Post
from config import TestingConfigclass TestConfig(unittest.TestCase):def setUp(self):self.app = create_app(TestingConfig)self.client = self.app.test_client()self.app_context = self.app.app_context()self.app_context.push()db.create_all()# 创建测试用户self.user = User(username='testuser', email='test@example.com')self.user.set_password('password')db.session.add(self.user)db.session.commit()def tearDown(self):db.session.remove()db.drop_all()self.app_context.pop()def test_home_page(self):response = self.client.get('/')self.assertEqual(response.status_code, 200)def test_user_registration(self):response = self.client.post('/auth/register', data={'username': 'newuser','email': 'new@example.com','password': 'password','confirm_password': 'password'}, follow_redirects=True)self.assertEqual(response.status_code, 200)# 验证用户是否创建user = User.query.filter_by(username='newuser').first()self.assertIsNotNone(user)def test_user_login(self):response = self.client.post('/auth/login', data={'username': 'testuser','password': 'password'}, follow_redirects=True)self.assertEqual(response.status_code, 200)def test_create_post(self):# 先登录self.client.post('/auth/login', data={'username': 'testuser','password': 'password'})# 创建文章response = self.client.post('/post/new', data={'title': 'Test Post','content': 'This is a test post'}, follow_redirects=True)self.assertEqual(response.status_code, 200)# 验证文章是否创建post = Post.query.filter_by(title='Test Post').first()self.assertIsNotNone(post)if __name__ == '__main__':unittest.main()

12. 应用入口 (app.py)

from app import create_app, db
from app.models import User, Post, Categoryapp = create_app()@app.shell_context_processor
def make_shell_context():return {'db': db,'User': User,'Post': Post,'Category': Category}if __name__ == '__main__':app.run(debug=True)

优势

学习曲线平缓
灵活性极高
代码简洁明了
扩展性强
适合小型到中型项目

劣势

需要手动配置很多功能
安全性需要额外实现
大型项目可能变得复杂
缺乏内置的数据库支持

FastAPI

发展历程

2018年：由Sebastián Ramírez创建
2019年：发布0.1版本
2020年：发布0.68版本，获得广泛关注
2021年：发布0.70版本，性能大幅提升
2022年：发布0.95版本，持续优化
2023年：发布0.104版本，稳定发展

核心特性

现代异步：基于Python 3.6+的async/await
高性能：接近Node.js和Go的性能
自动文档：基于OpenAPI的交互式API文档
类型提示：完整的Python类型提示支持
数据验证：基于Pydantic的自动数据验证
WebSocket支持：原生WebSocket支持

工作原理

请求 → ASGI → FastAPI应用 → 路由 → 异步视图 → 响应

代码案例

1. 项目结构

fastapi_app/
├── main.py
├── requirements.txt
├── app/
│   ├── __init__.py
│   ├── models.py
│   ├── schemas.py
│   ├── crud.py
│   ├── database.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── users.py
│   │   │   │   └── posts.py
│   │   │   └── api.py
│   │   └── deps.py
│   └── core/
│       ├── config.py
│       └── security.py
└── alembic/

2. 主应用文件 (main.py)

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.core.config import settings
from app.api.v1.api import api_routerapp = FastAPI(title=settings.PROJECT_NAME,version=settings.VERSION,description="FastAPI博客应用",openapi_url=f"{settings.API_V1_STR}/openapi.json"
)# CORS中间件
app.add_middleware(CORSMiddleware,allow_origins=settings.BACKEND_CORS_ORIGINS,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 注册路由
app.include_router(api_router, prefix=settings.API_V1_STR)@app.get("/")
async def root():return {"message": "欢迎使用FastAPI博客系统"}@app.get("/health")
async def health_check():return {"status": "healthy"}

3. 数据模型 (app/models.py)

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from datetime import datetimeBase = declarative_base()class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)email = Column(String, unique=True, index=True, nullable=False)username = Column(String, unique=True, index=True, nullable=False)hashed_password = Column(String, nullable=False)is_active = Column(Boolean, default=True)is_superuser = Column(Boolean, default=False)created_at = Column(DateTime(timezone=True), server_default=func.now())updated_at = Column(DateTime(timezone=True), onupdate=func.now())posts = relationship("Post", back_populates="author")comments = relationship("Comment", back_populates="author")class Post(Base):__tablename__ = "posts"id = Column(Integer, primary_key=True, index=True)title = Column(String, index=True, nullable=False)content = Column(Text, nullable=False)published = Column(Boolean, default=False)author_id = Column(Integer, ForeignKey("users.id"))created_at = Column(DateTime(timezone=True), server_default=func.now())updated_at = Column(DateTime(timezone=True), onupdate=func.now())author = relationship("User", back_populates="posts")comments = relationship("Comment", back_populates="post")class Comment(Base):__tablename__ = "comments"id = Column(Integer, primary_key=True, index=True)content = Column(Text, nullable=False)author_id = Column(Integer, ForeignKey("users.id"))post_id = Column(Integer, ForeignKey("posts.id"))created_at = Column(DateTime(timezone=True), server_default=func.now())author = relationship("User", back_populates="comments")post = relationship("Post", back_populates="comments")

4. Pydantic模式 (app/schemas.py)

from pydantic import BaseModel, EmailStr, Field
from typing import Optional, List
from datetime import datetime# User schemas
class UserBase(BaseModel):email: EmailStrusername: str = Field(..., min_length=3, max_length=50)class UserCreate(UserBase):password: str = Field(..., min_length=6)class UserUpdate(BaseModel):email: Optional[EmailStr] = Noneusername: Optional[str] = Field(None, min_length=3, max_length=50)is_active: Optional[bool] = Noneclass UserInDBBase(UserBase):id: intis_active: boolcreated_at: datetimeclass Config:from_attributes = Trueclass User(UserInDBBase):passclass UserInDB(UserInDBBase):hashed_password: str# Post schemas
class PostBase(BaseModel):title: str = Field(..., min_length=1, max_length=200)content: strpublished: bool = Falseclass PostCreate(PostBase):passclass PostUpdate(BaseModel):title: Optional[str] = Field(None, min_length=1, max_length=200)content: Optional[str] = Nonepublished: Optional[bool] = Noneclass PostInDBBase(PostBase):id: intauthor_id: intcreated_at: datetimeupdated_at: Optional[datetime] = Noneclass Config:from_attributes = Trueclass Post(PostInDBBase):author: Userclass PostInDB(PostInDBBase):pass# Comment schemas
class CommentBase(BaseModel):content: strclass CommentCreate(CommentBase):passclass Comment(CommentBase):id: intauthor_id: intpost_id: intcreated_at: datetimeauthor: Userclass Config:from_attributes = True# Token schemas
class Token(BaseModel):access_token: strtoken_type: strclass TokenPayload(BaseModel):sub: Optional[int] = None

5. 数据库配置 (app/database.py)

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settingsengine = create_engine(settings.SQLALCHEMY_DATABASE_URI,pool_pre_ping=True,echo=settings.SQLALCHEMY_ECHO
)SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)Base = declarative_base()def get_db():db = SessionLocal()try:yield dbfinally:db.close()

6. API端点 (app/api/v1/endpoints/users.py)

from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.api import depsrouter = APIRouter()@router.get("/", response_model=List[schemas.User])
def read_users(db: Session = Depends(deps.get_db),skip: int = 0,limit: int = 100,current_user: models.User = Depends(deps.get_current_active_superuser),
) -> Any:"""获取用户列表 (仅超级用户)"""users = crud.user.get_multi(db, skip=skip, limit=limit)return users@router.post("/", response_model=schemas.User)
def create_user(*,db: Session = Depends(deps.get_db),user_in: schemas.UserCreate,
) -> Any:"""创建新用户"""user = crud.user.get_by_email(db, email=user_in.email)if user:raise HTTPException(status_code=400,detail="该邮箱已被注册")user = crud.user.create(db, obj_in=user_in)return user@router.get("/me", response_model=schemas.User)
def read_user_me(current_user: models.User = Depends(deps.get_current_active_user),
) -> Any:"""获取当前用户信息"""return current_user@router.put("/me", response_model=schemas.User)
def update_user_me(*,db: Session = Depends(deps.get_db),password: str = None,full_name: str = None,email: str = None,current_user: models.User = Depends(deps.get_current_active_user),
) -> Any:"""更新当前用户信息"""current_user_data = schemas.UserUpdate(**current_user.__dict__)if password is not None:current_user_data.password = passwordif full_name is not None:current_user_data.full_name = full_nameif email is not None:current_user_data.email = emailuser = crud.user.update(db, db_obj=current_user, obj_in=current_user_data)return user@router.get("/{user_id}", response_model=schemas.User)
def read_user_by_id(user_id: int,current_user: models.User = Depends(deps.get_current_active_user),db: Session = Depends(deps.get_db),
) -> Any:"""根据ID获取用户信息"""user = crud.user.get(db, id=user_id)if user == current_user:return userif not crud.user.is_superuser(current_user):raise HTTPException(status_code=400, detail="权限不足")return user

7. 依赖注入 (app/api/deps.py)

from typing import Generator, Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app import crud, models, schemas
from app.core import security
from app.core.config import settings
from app.database import SessionLocaloauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/login/access-token")def get_db() -> Generator:try:db = SessionLocal()yield dbfinally:db.close()def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> models.User:try:payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[security.ALGORITHM])token_data = schemas.TokenPayload(**payload)except (JWTError, ValidationError):raise HTTPException(status_code=status.HTTP_403_FORBIDDEN,detail="无法验证凭据",)user = crud.user.get(db, id=token_data.sub)if not user:raise HTTPException(status_code=404, detail="用户不存在")return userdef get_current_active_user(current_user: models.User = Depends(get_current_user),
) -> models.User:if not crud.user.is_active(current_user):raise HTTPException(status_code=400, detail="用户未激活")return current_userdef get_current_active_superuser(current_user: models.User = Depends(get_current_user),
) -> models.User:if not crud.user.is_superuser(current_user):raise HTTPException(status_code=400, detail="用户权限不足")return current_user

8. 安全配置 (app/core/security.py)

from datetime import datetime, timedelta
from typing import Any, Union
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settingspwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")ALGORITHM = "HS256"def create_access_token(subject: Union[str, Any], expires_delta: timedelta = None
) -> str:if expires_delta:expire = datetime.utcnow() + expires_deltaelse:expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)to_encode = {"exp": expire, "sub": str(subject)}encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)return encoded_jwtdef verify_password(plain_password: str, hashed_password: str) -> bool:return pwd_context.verify(plain_password, hashed_password)def get_password_hash(password: str) -> str:return pwd_context.hash(password)

9. 异步操作示例

import asyncio
from fastapi import APIRouter, BackgroundTasks
from app.schemas import PostCreate, Post
from app.crud import post as crud_postrouter = APIRouter()@router.post("/async-create", response_model=Post)
async def create_post_async(post_in: PostCreate,background_tasks: BackgroundTasks,db: Session = Depends(deps.get_db),current_user: models.User = Depends(deps.get_current_active_user),
):"""异步创建文章"""# 模拟异步数据库操作await asyncio.sleep(0.1)# 添加后台任务background_tasks.add_task(send_notification, current_user.email, "文章创建成功")post = crud_post.create_with_owner(db=db, obj_in=post_in, owner_id=current_user.id)return post@router.get("/async-search")
async def search_posts_async(q: str,db: Session = Depends(deps.get_db),skip: int = 0,limit: int = 100,
):"""异步搜索文章"""# 模拟异步搜索操作await asyncio.sleep(0.05)posts = crud_post.search(db, query=q, skip=skip, limit=limit)return postsasync def send_notification(email: str, message: str):"""后台任务：发送通知"""await asyncio.sleep(1)  # 模拟发送邮件print(f"发送通知到 {email}: {message}")

10. WebSocket支持

from fastapi import WebSocket, WebSocketDisconnect
from typing import Listclass ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] = []async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)async def send_personal_message(self, message: str, websocket: WebSocket):await websocket.send_text(message)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()@router.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):await manager.connect(websocket)try:while True:data = await websocket.receive_text()await manager.send_personal_message(f"你发送了: {data}", websocket)await manager.broadcast(f"客户端 #{client_id} 说: {data}")except WebSocketDisconnect:manager.disconnect(websocket)await manager.broadcast(f"客户端 #{client_id} 离开了聊天室")

11. 测试代码

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base, get_db
from app.main import app
import pytestSQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)def override_get_db():try:db = TestingSessionLocal()yield dbfinally:db.close()app.dependency_overrides[get_db] = override_get_dbclient = TestClient(app)def test_create_user():response = client.post("/api/v1/users/",json={"email": "test@example.com", "username": "testuser", "password": "password123"})assert response.status_code == 200data = response.json()assert data["email"] == "test@example.com"assert data["username"] == "testuser"assert "id" in datadef test_read_users():response = client.get("/api/v1/users/")assert response.status_code == 200assert isinstance(response.json(), list)def test_root():response = client.get("/")assert response.status_code == 200assert response.json() == {"message": "欢迎使用FastAPI博客系统"}

优势

性能卓越，接近原生性能
开发体验优秀，类型提示完整
自动生成API文档
数据验证自动化
现代化设计，支持异步

劣势

相对较新，生态系统还在发展
学习成本相对较高
对Python版本要求较高
企业级应用案例相对较少

Tornado

发展历程

2009年：由FriendFeed创建
2011年：被Facebook收购
2012年：开源发布
2015年：发布4.0版本，支持Python 3
2017年：发布5.0版本，性能优化
2020年：发布6.0版本，持续改进

核心特性

异步网络库：非阻塞I/O
Web框架：内置HTTP服务器
WebSocket支持：实时通信
高性能：适合长连接应用
单线程：事件循环驱动

工作原理

事件循环 → 非阻塞I/O → 回调处理 → 响应

代码案例

1. 基本应用结构

import tornado.ioloop
import tornado.web
import tornado.websocket
import json
import asyncioclass MainHandler(tornado.web.RequestHandler):def get(self):self.write("Hello, Tornado!")class JSONHandler(tornado.web.RequestHandler):def get(self):data = {"message": "Hello from Tornado","status": "success","timestamp": tornado.ioloop.IOLoop.current().time()}self.set_header("Content-Type", "application/json")self.write(json.dumps(data))class PostHandler(tornado.web.RequestHandler):def post(self):try:data = json.loads(self.request.body)# 处理数据response = {"status": "success", "data": data}self.set_header("Content-Type", "application/json")self.write(json.dumps(response))except json.JSONDecodeError:self.set_status(400)self.write({"error": "Invalid JSON"})def make_app():return tornado.web.Application([(r"/", MainHandler),(r"/api/json", JSONHandler),(r"/api/post", PostHandler),])if __name__ == "__main__":app = make_app()app.listen(8888)print("Tornado server running on http://localhost:8888")tornado.ioloop.IOLoop.current().start()

2. 异步处理示例

import tornado.web
import tornado.gen
import asyncioclass AsyncHandler(tornado.web.RequestHandler):@tornado.gen.coroutinedef get(self):# 模拟异步数据库查询result = yield self.async_operation()self.write({"result": result})@tornado.gen.coroutinedef async_operation(self):# 模拟异步操作yield tornado.gen.sleep(1)return "异步操作完成"class ModernAsyncHandler(tornado.web.RequestHandler):async def get(self):# 使用现代async/await语法result = await self.async_operation()self.write({"result": result})async def async_operation(self):await asyncio.sleep(1)return "现代异步操作完成"

3. WebSocket实现

import tornado.websocket
import jsonclass ChatWebSocket(tornado.websocket.WebSocketHandler):clients = set()def open(self):ChatWebSocket.clients.add(self)self.write_message(json.dumps({"type": "system","message": "欢迎加入聊天室！"}))def on_message(self, message):try:data = json.loads(message)# 广播消息给所有客户端for client in ChatWebSocket.clients:client.write_message(json.dumps({"type": "chat","user": data.get("user", "匿名"),"message": data.get("message", ""),"timestamp": tornado.ioloop.IOLoop.current().time()}))except json.JSONDecodeError:self.write_message(json.dumps({"type": "error","message": "消息格式错误"}))def on_close(self):ChatWebSocket.clients.remove(self)# 通知其他用户离开for client in ChatWebSocket.clients:client.write_message(json.dumps({"type": "system","message": "有用户离开了聊天室"}))class WebSocketApp(tornado.web.Application):def __init__(self):handlers = [(r"/ws", ChatWebSocket),(r"/chat", tornado.web.StaticFileHandler, {"path": "static/chat.html"})]super().__init__(handlers)

4. 数据库集成

import tornado.web
import motor.motor_asyncio
import asyncioclass DatabaseHandler(tornado.web.RequestHandler):def initialize(self):# 初始化MongoDB连接self.client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017")self.db = self.client.blogasync def get(self):# 异步查询数据库posts = await self.db.posts.find().to_list(100)self.write({"posts": posts,"count": len(posts)})async def post(self):try:data = json.loads(self.request.body)result = await self.db.posts.insert_one(data)self.write({"status": "success","id": str(result.inserted_id)})except Exception as e:self.set_status(500)self.write({"error": str(e)})

5. 中间件和认证

import tornado.web
import jwt
import timeclass AuthMiddleware:def __init__(self, secret_key):self.secret_key = secret_keydef authenticate(self, handler):auth_header = handler.request.headers.get("Authorization")if not auth_header:return Nonetry:token = auth_header.split(" ")[1]payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])return payloadexcept:return Noneclass ProtectedHandler(tornado.web.RequestHandler):def initialize(self, auth_middleware):self.auth_middleware = auth_middlewaredef get(self):user = self.auth_middleware.authenticate(self)if not user:self.set_status(401)self.write({"error": "未授权访问"})returnself.write({"message": f"欢迎, {user['username']}!","user_id": user['user_id']})# 使用中间件
auth = AuthMiddleware("your-secret-key")
app = tornado.web.Application([(r"/protected", ProtectedHandler, {"auth_middleware": auth})
])

6. 错误处理

import tornado.web
import loggingclass ErrorHandler(tornado.web.RequestHandler):def write_error(self, status_code, **kwargs):if status_code == 404:self.render("404.html")elif status_code == 500:self.render("500.html")else:self.render("error.html", status_code=status_code)class CustomErrorHandler(tornado.web.ErrorHandler, ErrorHandler):passclass Application(tornado.web.Application):def __init__(self):handlers = [(r"/", MainHandler),]settings = {"debug": False,"default_handler_class": CustomErrorHandler,}super().__init__(handlers, **settings)

7. 配置管理

import tornado.options
import os# 定义命令行选项
tornado.options.define("port", default=8888, help="运行端口", type=int)
tornado.options.define("debug", default=False, help="调试模式", type=bool)
tornado.options.define("config", default="config.py", help="配置文件路径")class Config:def __init__(self):self.port = tornado.options.options.portself.debug = tornado.options.options.debug# 从环境变量读取配置self.database_url = os.environ.get("DATABASE_URL", "sqlite:///app.db")self.secret_key = os.environ.get("SECRET_KEY", "default-secret-key")# 从配置文件读取if os.path.exists(tornado.options.options.config):with open(tornado.options.options.config) as f:exec(f.read(), {}, self.__dict__)# 使用配置
tornado.options.parse_command_line()
config = Config()app = tornado.web.Application([(r"/", MainHandler),
], debug=config.debug)if __name__ == "__main__":app.listen(config.port)print(f"Server running on port {config.port}")tornado.ioloop.IOLoop.current().start()

8. 性能监控

import tornado.web
import time
import psutilclass PerformanceHandler(tornado.web.RequestHandler):def initialize(self):self.start_time = time.time()def on_finish(self):duration = time.time() - self.start_time# 记录请求时间print(f"请求处理时间: {duration:.4f}秒")# 记录系统资源使用cpu_percent = psutil.cpu_percent()memory = psutil.virtual_memory()print(f"CPU使用率: {cpu_percent}%")print(f"内存使用: {memory.percent}%")class StatsHandler(tornado.web.RequestHandler):def get(self):stats = {"cpu_percent": psutil.cpu_percent(),"memory_percent": psutil.virtual_memory().percent,"disk_usage": psutil.disk_usage('/').percent,"uptime": time.time() - psutil.boot_time()}self.write(stats)

优势

异步性能优秀
适合长连接应用
内置HTTP服务器
实时通信支持好

劣势

学习曲线陡峭
调试相对困难
生态系统相对较小
不适合传统CRUD应用

Sanic

发展历程

2016年：由ChannelCat创建
2017年：发布0.1版本
2018年：发布18.0版本，性能突破
2019年：发布19.0版本，功能完善
2020年：发布20.0版本，稳定发展
2023年：发布23.0版本，持续优化

核心特性

异步框架：基于uvloop和httptools
高性能：接近Go的性能
类Flask语法：熟悉的API设计
WebSocket支持：原生WebSocket
中间件系统：灵活的中间件支持

工作原理

请求 → uvloop → Sanic应用 → 路由 → 异步视图 → 响应

代码案例

1. 基本应用结构

from sanic import Sanic, Request, response
from sanic.response import json, html, text
from sanic_cors import CORS
import asyncioapp = Sanic("MySanicApp")
CORS(app)@app.route("/")
async def hello_world(request: Request):return text("Hello, Sanic!")@app.route("/api/json")
async def json_response(request: Request):return json({"message": "Hello from Sanic","status": "success","framework": "Sanic"})@app.route("/api/users/<user_id:int>")
async def get_user(request: Request, user_id: int):# 模拟数据库查询user = {"id": user_id, "name": f"User{user_id}", "email": f"user{user_id}@example.com"}return json(user)if __name__ == "__main__":app.run(host="0.0.0.0", port=8000, debug=True)

2. 异步视图和中间件

from sanic import Sanic, Request
from sanic.response import json
import timeapp = Sanic("AsyncApp")@app.middleware("request")
async def add_start_time(request: Request):request.ctx.start_time = time.time()@app.middleware("response")
async def add_process_time(request: Request, response):process_time = time.time() - request.ctx.start_timeresponse.headers["X-Process-Time"] = str(process_time)@app.route("/async-operation")
async def async_operation(request: Request):# 模拟异步数据库操作await asyncio.sleep(0.1)return json({"message": "异步操作完成","process_time": time.time() - request.ctx.start_time})@app.route("/parallel-operations")
async def parallel_operations(request: Request):# 并行执行多个异步操作tasks = [asyncio.sleep(0.1),asyncio.sleep(0.2),asyncio.sleep(0.3)]results = await asyncio.gather(*tasks)return json({"message": "并行操作完成","results": results})

3. 蓝图组织

from sanic import Blueprint, Request
from sanic.response import json# 用户蓝图
users_bp = Blueprint("users", url_prefix="/users")@users_bp.route("/")
async def get_users(request: Request):users = [{"id": 1, "name": "Alice"},{"id": 2, "name": "Bob"}]return json(users)@users_bp.route("/<user_id:int>")
async def get_user(request: Request, user_id: int):user = {"id": user_id, "name": f"User{user_id}"}return json(user)# 文章蓝图
posts_bp = Blueprint("posts", url_prefix="/posts")@posts_bp.route("/")
async def get_posts(request: Request):posts = [{"id": 1, "title": "First Post", "content": "Hello World"},{"id": 2, "title": "Second Post", "content": "Sanic is fast"}]return json(posts)@posts_bp.route("/<post_id:int>")
async def get_post(request: Request, post_id: int):post = {"id": post_id, "title": f"Post {post_id}", "content": "Content here"}return json(post)# 主应用
app = Sanic("BlueprintApp")
app.blueprint(users_bp)
app.blueprint(posts_bp)

4. 数据库集成

from sanic import Sanic, Request
from sanic.response import json
import asyncpg
import asyncioapp = Sanic("DatabaseApp")# 数据库连接池
async def get_db_pool():return await asyncpg.create_pool(user="postgres",password="password",database="blog",host="localhost",port=5432)@app.listener("before_server_start")
async def setup_db(app, loop):app.ctx.db_pool = await get_db_pool()@app.listener("after_server_stop")
async def close_db(app, loop):await app.ctx.db_pool.close()@app.route("/posts")
async def get_posts(request: Request):async with app.ctx.db_pool.acquire() as conn:rows = await conn.fetch("SELECT * FROM posts ORDER BY created_at DESC")posts = [dict(row) for row in rows]return json(posts)@app.route("/posts", methods=["POST"])
async def create_post(request: Request):data = request.jsonasync with app.ctx.db_pool.acquire() as conn:row = await conn.fetchrow("INSERT INTO posts (title, content, author_id) VALUES ($1, $2, $3) RETURNING *",data["title"], data["content"], data["author_id"])return json(dict(row))

5. WebSocket支持

from sanic import Sanic, Request
from sanic.response import json
from sanic.websocket import WebSocketProtocolapp = Sanic("WebSocketApp")@app.websocket("/ws")
async def websocket_handler(request: Request, ws):try:while True:data = await ws.recv()# 处理接收到的消息response = f"收到消息: {data}"await ws.send(response)except Exception as e:print(f"WebSocket错误: {e}")@app.route("/broadcast")
async def broadcast_message(request: Request):message = request.args.get("message", "Hello everyone!")# 这里应该实现向所有连接的WebSocket客户端广播消息return json({"status": "broadcast sent", "message": message})if __name__ == "__main__":app.run(host="0.0.0.0", port=8000, protocol=WebSocketProtocol)

6. 配置管理

from sanic import Sanic
import osclass Config:DEBUG = FalseHOST = "0.0.0.0"PORT = 8000DATABASE_URL = "postgresql://user:pass@localhost/db"SECRET_KEY = "your-secret-key"class DevelopmentConfig(Config):DEBUG = TrueDATABASE_URL = "postgresql://dev:dev@localhost/dev_db"class ProductionConfig(Config):DEBUG = FalseHOST = "0.0.0.0"PORT = int(os.environ.get("PORT", 8000))DATABASE_URL = os.environ.get("DATABASE_URL")# 根据环境选择配置
config_class = DevelopmentConfig if os.environ.get("ENV") == "development" else ProductionConfigapp = Sanic("ConfigApp")
app.config.update(config_class.__dict__)@app.route("/config")
async def show_config(request):return {"debug": app.config.DEBUG,"host": app.config.HOST,"port": app.config.PORT}

7. 错误处理

from sanic import Sanic, Request
from sanic.response import json
from sanic.exceptions import NotFound, ServerErrorapp = Sanic("ErrorHandlingApp")@app.exception(NotFound)
async def not_found(request: Request, exception):return json({"error": "页面未找到","path": request.path,"status_code": 404}, status=404)@app.exception(ServerError)
async def server_error(request: Request, exception):return json({"error": "服务器内部错误","path": request.path,"status_code": 500}, status=500)@app.exception(Exception)
async def general_exception(request: Request, exception):return json({"error": "未知错误","message": str(exception),"status_code": 500}, status=500)@app.route("/error-test")
async def error_test(request: Request):# 故意引发错误来测试错误处理raise ValueError("这是一个测试错误")

8. 性能优化

from sanic import Sanic, Request
from sanic.response import json
import asyncio
import timeapp = Sanic("PerformanceApp")# 缓存装饰器
def cache_response(ttl=300):def decorator(f):async def wrapper(request: Request, *args, **kwargs):cache_key = f"cache:{request.path}"# 这里应该实现Redis缓存逻辑# 简化示例，实际应该使用Redisresult = await f(request, *args, **kwargs)return resultreturn wrapperreturn decorator@app.route("/cached-data")
@cache_response(ttl=60)
async def cached_data(request: Request):# 模拟耗时操作await asyncio.sleep(0.1)return json({"data": "这是缓存的数据","timestamp": time.time()})@app.route("/batch-operations")
async def batch_operations(request: Request):# 批量处理多个操作start_time = time.time()# 并行执行多个异步操作tasks = []for i in range(10):task = asyncio.sleep(0.1)  # 模拟异步操作tasks.append(task)await asyncio.gather(*tasks)end_time = time.time()return json({"message": "批量操作完成","total_time": end_time - start_time,"operations": 10})

9. 测试代码

import pytest
from sanic import Sanic
from sanic.testing import SanicTestClient@pytest.fixture
def app():app = Sanic("TestApp")@app.route("/")async def hello_world(request):return {"message": "Hello, World!"}@app.route("/users/<user_id:int>")async def get_user(request, user_id):return {"id": user_id, "name": f"User{user_id}"}return app@pytest.fixture
def client(app):return SanicTestClient(app)def test_hello_world(client):request, response = client.get("/")assert response.status == 200assert response.json["message"] == "Hello, World!"def test_get_user(client):request, response = client.get("/users/1")assert response.status == 200assert response.json["id"] == 1assert response.json["name"] == "User1"def test_get_user_invalid_id(client):request, response = client.get("/users/invalid")assert response.status == 404if __name__ == "__main__":pytest.main([__file__])

优势

性能卓越
语法熟悉（类似Flask）
异步支持好
适合高性能应用

劣势

生态系统相对较小
文档相对简单
企业级支持有限

aiohttp

发展历程

2013年：由Andrew Svetlov创建
2014年：发布0.1版本
2016年：发布1.0版本，API稳定
2018年：发布3.0版本，性能提升
2020年：发布3.7版本，持续优化
2023年：发布3.8版本，稳定发展

核心特性

异步HTTP客户端/服务器：完整的异步HTTP支持
WebSocket支持：客户端和服务器端WebSocket
中间件系统：灵活的中间件支持
信号支持：应用生命周期管理
静态文件服务：内置静态文件支持

工作原理

异步事件循环 → HTTP请求/响应 → 异步处理 → 响应

代码案例

1. 基本HTTP服务器

from aiohttp import web, ClientSession
import asyncio
import jsonasync def hello(request):return web.Response(text="Hello, aiohttp!")async def json_response(request):data = {"message": "Hello from aiohttp", "status": "success"}return web.json_response(data)async def get_user(request):user_id = request.match_info['user_id']user = {"id": user_id, "name": f"User{user_id}"}return web.json_response(user)app = web.Application()
app.router.add_get('/', hello)
app.router.add_get('/api/json', json_response)
app.router.add_get('/api/users/{user_id}', get_user)if __name__ == '__main__':web.run_app(app, host='127.0.0.1', port=8080)

2. 异步数据库操作

import asyncpg
from aiohttp import webasync def get_posts(request):pool = request.app['db_pool']async with pool.acquire() as conn:rows = await conn.fetch("SELECT * FROM posts ORDER BY created_at DESC")posts = [dict(row) for row in rows]return web.json_response(posts)async def create_post(request):pool = request.app['db_pool']data = await request.json()async with pool.acquire() as conn:row = await conn.fetchrow("INSERT INTO posts (title, content) VALUES ($1, $2) RETURNING *",data['title'], data['content'])return web.json_response(dict(row))async def init_db(app):app['db_pool'] = await asyncpg.create_pool(user='postgres', password='password',database='blog', host='localhost')async def close_db(app):await app['db_pool'].close()app = web.Application()
app.on_startup.append(init_db)
app.on_cleanup.append(close_db)
app.router.add_get('/posts', get_posts)
app.router.add_post('/posts', create_post)

3. WebSocket实现

from aiohttp import web, WSMsgTypeasync def websocket_handler(request):ws = web.WebSocketResponse()await ws.prepare(request)try:async for msg in ws:if msg.type == WSMsgType.TEXT:await ws.send_str(f"收到消息: {msg.data}")elif msg.type == WSMsgType.ERROR:print(f'WebSocket错误: {ws.exception()}')finally:return wsapp = web.Application()
app.router.add_get('/ws', websocket_handler)

优势

异步性能优秀
HTTP客户端/服务器完整
WebSocket支持好
适合微服务架构

劣势

学习成本较高
调试相对困难
生态系统相对较小

Starlette

发展历程

2018年：由Tom Christie创建
2019年：发布0.1版本
2020年：发布0.14版本，功能完善
2021年：发布0.17版本，性能优化
2022年：发布0.20版本，稳定发展
2023年：发布0.27版本，持续改进

核心特性

ASGI工具库：ASGI规范实现
轻量级：核心功能简洁
高性能：优化的性能表现
WebSocket支持：原生WebSocket
后台任务：异步后台任务支持

工作原理

ASGI应用 → 中间件 → 路由 → 视图 → 响应

代码案例

1. 基本ASGI应用

from starlette.applications import Starlette
from starlette.responses import JSONResponse, HTMLResponse
from starlette.routing import Route
import uvicornasync def homepage(request):return HTMLResponse("""<html><head><title>Starlette App</title></head><body><h1>Hello from Starlette!</h1></body></html>""")async def users(request):users = [{"id": 1, "name": "Alice"},{"id": 2, "name": "Bob"}]return JSONResponse({"users": users})async def user_detail(request):user_id = request.path_params["user_id"]user = {"id": int(user_id), "name": f"User{user_id}"}return JSONResponse(user)routes = [Route("/", homepage),Route("/users", users),Route("/users/{user_id}", user_detail),
]app = Starlette(routes=routes)if __name__ == "__main__":uvicorn.run(app, host="0.0.0.0", port=8000)

2. 中间件和后台任务

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
from starlette.background import BackgroundTask
import timeclass TimingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return responseasync def send_notification(email: str):# 模拟发送通知await asyncio.sleep(1)print(f"通知已发送到 {email}")async def create_user(request):data = await request.json()# 添加后台任务background = BackgroundTask(send_notification, data["email"])return JSONResponse({"message": "用户创建成功"},background=background)app = Starlette(routes=[Route("/users", create_user, methods=["POST"])],middleware=[Middleware(TimingMiddleware)]
)

3. WebSocket支持

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket
import jsonasync def websocket_endpoint(websocket: WebSocket):await websocket.accept()try:while True:data = await websocket.receive_text()message = json.loads(data)# 处理消息response = {"type": "response","message": f"收到消息: {message.get('content', '')}","timestamp": time.time()}await websocket.send_text(json.dumps(response))except Exception as e:print(f"WebSocket错误: {e}")finally:await websocket.close()app = Starlette(routes=[WebSocketRoute("/ws", websocket_endpoint)]
)

优势

轻量级设计
性能优秀
ASGI标准实现
适合构建自定义框架

劣势

功能相对简单
需要更多手动配置
学习资源相对较少

Bottle

发展历程

2009年：由Marcel Hellkamp创建
2010年：发布0.1版本
2012年：发布0.11版本，功能完善
2015年：发布0.12版本，稳定发展
2018年：发布0.12.18版本，持续维护
2023年：发布0.12.25版本，安全更新

核心特性

单文件框架：整个框架在一个文件中
轻量级：核心功能简洁
内置服务器：开发服务器
模板引擎：内置简单模板
WSGI支持：标准WSGI应用

工作原理

请求 → WSGI → Bottle应用 → 路由 → 视图 → 响应

代码案例

1. 基本应用

from bottle import Bottle, run, request, response
import jsonapp = Bottle()@app.route('/')
def hello():return "Hello, Bottle!"@app.route('/api/json')
def json_response():response.content_type = 'application/json'return json.dumps({"message": "Hello from Bottle", "status": "success"})@app.route('/api/users/<user_id:int>')
def get_user(user_id):user = {"id": user_id, "name": f"User{user_id}"}response.content_type = 'application/json'return json.dumps(user)@app.route('/api/posts', method='POST')
def create_post():data = request.json# 处理数据response.content_type = 'application/json'return json.dumps({"status": "success", "data": data})if __name__ == '__main__':run(app, host='localhost', port=8080, debug=True)

2. 模板使用

from bottle import Bottle, run, request, response, templateapp = Bottle()@app.route('/')
def index():users = [{"id": 1, "name": "Alice"},{"id": 2, "name": "Bob"}]return template('index', users=users)@app.route('/user/<user_id:int>')
def user_profile(user_id):user = {"id": user_id, "name": f"User{user_id}"}return template('user', user=user)# 模板文件 (views/index.tpl)
"""
<!DOCTYPE html>
<html>
<head><title>用户列表</title></head>
<body><h1>用户列表</h1><ul>% for user in users:<li><a href="/user/{{user['id']}}">{{user['name']}}</a></li>% end</ul>
</body>
</html>
"""

优势

极简设计
单文件部署
学习成本低
适合小型项目

劣势

功能相对有限
不适合大型项目
生态系统小
性能相对较低

Pyramid

发展历程

2008年：由Pylons项目创建
2010年：发布1.0版本
2012年：发布1.4版本，功能完善
2015年：发布1.6版本，性能优化
2018年：发布1.9版本，稳定发展
2023年：发布2.0版本，现代化改进

核心特性

灵活架构：可大可小的应用架构
配置系统：强大的配置管理
认证授权：完整的权限系统
视图系统：灵活的视图组织
中间件支持：丰富的中间件

工作原理

请求 → WSGI → Pyramid应用 → 路由 → 视图 → 响应

代码案例

1. 基本应用结构

from pyramid.config import Configurator
from pyramid.response import Response
from pyramid.view import view_config
import json@view_config(route_name='home', renderer='json')
def home_view(request):return {"message": "Hello from Pyramid"}@view_config(route_name='users', renderer='json')
def users_view(request):users = [{"id": 1, "name": "Alice"},{"id": 2, "name": "Bob"}]return {"users": users}@view_config(route_name='user', renderer='json')
def user_view(request):user_id = request.matchdict['user_id']user = {"id": user_id, "name": f"User{user_id}"}return userdef main(global_config, **settings):config = Configurator(settings=settings)# 添加路由config.add_route('home', '/')config.add_route('users', '/api/users')config.add_route('user', '/api/users/{user_id}')# 扫描视图config.scan()return config.make_wsgi_app()if __name__ == '__main__':from wsgiref.simple_server import make_serverapp = main({})server = make_server('0.0.0.0', 6543, app)server.serve_forever()

2. 认证和授权

from pyramid.config import Configurator
from pyramid.authentication import AuthTktAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy
from pyramid.security import Allow, Deny, Everyone, Authenticatedclass RootFactory:__acl__ = [(Allow, Everyone, 'view'),(Allow, Authenticated, 'edit'),(Allow, 'admin', 'admin')]def __init__(self, request):self.request = request@view_config(route_name='protected', renderer='json', permission='edit')
def protected_view(request):return {"message": "这是受保护的资源", "user": request.authenticated_userid}def main(global_config, **settings):config = Configurator(settings=settings)# 认证策略authn_policy = AuthTktAuthenticationPolicy('secret')authz_policy = ACLAuthorizationPolicy()config.set_authentication_policy(authn_policy)config.set_authorization_policy(authz_policy)config.set_root_factory(RootFactory)config.add_route('protected', '/api/protected')config.scan()return config.make_wsgi_app()

优势

架构灵活
配置强大
认证授权完整
适合中型到大型项目

劣势

学习曲线陡峭
配置相对复杂
性能相对较低
社区相对较小

CherryPy

发展历程

2002年：由Remi Delon创建
2004年：发布1.0版本
2008年：发布3.0版本，重大改进
2012年：发布3.2版本，功能完善
2016年：发布18.0版本，现代化
2023年：发布18.8版本，持续维护

核心特性

面向对象：基于对象的Web框架
内置服务器：高性能HTTP服务器
插件系统：灵活的插件架构
配置管理：强大的配置系统
工具支持：丰富的开发工具

工作原理

请求 → HTTP服务器 → CherryPy应用 → 对象方法 → 响应

代码案例

1. 基本应用结构

import cherrypyclass Root:@cherrypy.exposedef index(self):return "Hello, CherryPy!"@cherrypy.exposedef about(self):return "About CherryPy"class Users:@cherrypy.exposedef index(self):return "用户列表"@cherrypy.exposedef profile(self, user_id):return f"用户 {user_id} 的个人资料"class Posts:@cherrypy.exposedef index(self):return "文章列表"@cherrypy.exposedef view(self, post_id):return f"查看文章 {post_id}"# 挂载子应用
root = Root()
root.users = Users()
root.posts = Posts()if __name__ == '__main__':cherrypy.quickstart(root, '/')

2. JSON API实现

import cherrypy
import jsonclass API:@cherrypy.expose@cherrypy.tools.json_out()def users(self):users = [{"id": 1, "name": "Alice", "email": "alice@example.com"},{"id": 2, "name": "Bob", "email": "bob@example.com"}]return {"users": users, "count": len(users)}@cherrypy.expose@cherrypy.tools.json_out()def user(self, user_id):user = {"id": int(user_id), "name": f"User{user_id}", "email": f"user{user_id}@example.com"}return user@cherrypy.expose@cherrypy.tools.json_in()@cherrypy.tools.json_out()def create_user(self):data = cherrypy.request.json# 处理用户创建逻辑return {"status": "success", "message": "用户创建成功", "data": data}# 配置JSON工具
cherrypy.tools.json_in.on = True
cherrypy.tools.json_out.on = Trueif __name__ == '__main__':cherrypy.quickstart(API(), '/api')

3. 配置管理

import cherrypy
import osclass Config:def __init__(self):self.host = "0.0.0.0"self.port = 8080self.debug = True# 从环境变量读取配置if os.environ.get("CHERRYPY_HOST"):self.host = os.environ.get("CHERRYPY_HOST")if os.environ.get("CHERRYPY_PORT"):self.port = int(os.environ.get("CHERRYPY_PORT"))if os.environ.get("CHERRYPY_DEBUG"):self.debug = os.environ.get("CHERRYPY_DEBUG").lower() == "true"class App:@cherrypy.exposedef index(self):return "配置化CherryPy应用"@cherrypy.exposedef config(self):return f"Host: {cherrypy.server.socket_host}, Port: {cherrypy.server.socket_port}"def main():config = Config()# 服务器配置cherrypy.config.update({'server.socket_host': config.host,'server.socket_port': config.port,'engine.autoreload.on': config.debug,'log.screen': config.debug})# 启动应用cherrypy.quickstart(App())if __name__ == '__main__':main()

优势

面向对象设计
内置服务器性能好
插件系统灵活
配置管理强大

劣势

学习曲线陡峭
文档相对简单
生态系统相对较小
现代化程度相对较低

框架对比分析

性能对比

框架	同步性能	异步性能	内存占用	CPU使用率
Django	中等	低	高	中等
Flask	中等	低	中等	中等
FastAPI	高	很高	低	低
Tornado	高	很高	中等	低
Sanic	很高	很高	低	低
aiohttp	高	很高	低	低
Starlette	高	很高	低	低
Bottle	中等	低	低	中等
Pyramid	中等	低	中等	中等
CherryPy	中等	低	中等	中等

学习曲线对比

框架	初学者友好度	文档质量	社区活跃度	学习资源
Django	中等	优秀	很高	丰富
Flask	高	优秀	很高	丰富
FastAPI	中等	优秀	高	丰富
Tornado	低	中等	中等	中等
Sanic	中等	中等	中等	中等
aiohttp	低	中等	中等	中等
Starlette	中等	中等	中等	中等
Bottle	很高	简单	低	少
Pyramid	低	中等	中等	中等
CherryPy	低	中等	低	少

生态系统对比

框架	扩展数量	第三方库	企业支持	部署工具
Django	很多	丰富	很好	丰富
Flask	很多	丰富	好	丰富
FastAPI	中等	中等	中等	中等
Tornado	少	中等	中等	中等
Sanic	少	中等	少	中等
aiohttp	少	中等	中等	中等
Starlette	少	少	少	少
Bottle	很少	少	很少	少
Pyramid	中等	中等	中等	中等
CherryPy	很少	少	很少	少

代码对比示例

1. 简单路由实现对比

Django

from django.urls import path
from . import viewsurlpatterns = [path('hello/', views.hello, name='hello'),
]def hello(request):return JsonResponse({"message": "Hello, World!"})

Flask

from flask import Flask, jsonifyapp = Flask(__name__)@app.route('/hello/')
def hello():return jsonify({"message": "Hello, World!"})

FastAPI

from fastapi import FastAPIapp = FastAPI()@app.get("/hello/")
async def hello():return {"message": "Hello, World!"}

Tornado

import tornado.web
import jsonclass HelloHandler(tornado.web.RequestHandler):def get(self):self.write(json.dumps({"message": "Hello, World!"}))

Sanic

from sanic import Sanicapp = Sanic("MyApp")@app.route("/hello/")
async def hello(request):return {"message": "Hello, World!"}

2. 数据库查询对比

Django ORM

from django.shortcuts import render
from .models import Userdef get_users(request):users = User.objects.all()return render(request, 'users.html', {'users': users})

Flask + SQLAlchemy

from flask import render_template
from .models import User@app.route('/users')
def get_users():users = User.query.all()return render_template('users.html', users=users)

FastAPI + SQLAlchemy

from fastapi import Depends
from sqlalchemy.orm import Session
from .database import get_db
from .models import User@app.get("/users")
async def get_users(db: Session = Depends(get_db)):users = db.query(User).all()return users

Tornado + Motor

import motor.motor_asyncioclass UsersHandler(tornado.web.RequestHandler):async def get(self):client = motor.motor_asyncio.AsyncIOMotorClient()db = client.test_databaseusers = await db.users.find().to_list(100)self.write({"users": users})

3. 中间件实现对比

Django 中间件

class TimingMiddleware:def __init__(self, get_response):self.get_response = get_responsedef __call__(self, request):start_time = time.time()response = self.get_response(request)duration = time.time() - start_timeresponse['X-Process-Time'] = str(duration)return response

Flask 中间件

@app.before_request
def before_request():g.start_time = time.time()@app.after_request
def after_request(response):duration = time.time() - g.start_timeresponse.headers['X-Process-Time'] = str(duration)return response

FastAPI 中间件

from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddlewareclass TimingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request, call_next):start_time = time.time()response = await call_next(request)duration = time.time() - start_timeresponse.headers["X-Process-Time"] = str(duration)return responseapp.add_middleware(TimingMiddleware)

Sanic 中间件

@app.middleware("request")
async def add_start_time(request):request.ctx.start_time = time.time()@app.middleware("response")
async def add_process_time(request, response):process_time = time.time() - request.ctx.start_timeresponse.headers["X-Process-Time"] = str(process_time)

4. 错误处理对比

Django 错误处理

from django.shortcuts import renderdef handler404(request, exception):return render(request, '404.html', status=404)def handler500(request):return render(request, '500.html', status=500)

Flask 错误处理

@app.errorhandler(404)
def not_found_error(error):return render_template('404.html'), 404@app.errorhandler(500)
def internal_error(error):return render_template('500.html'), 500

FastAPI 错误处理

from fastapi import HTTPException
from fastapi.responses import JSONResponse@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):return JSONResponse(status_code=exc.status_code,content={"detail": exc.detail})

Tornado 错误处理

class CustomErrorHandler(tornado.web.ErrorHandler):def write_error(self, status_code, **kwargs):if status_code == 404:self.render("404.html")else:self.render("error.html", status_code=status_code)

5. 配置管理对比

Django 配置

# settings.py
import osDEBUG = os.environ.get('DEBUG', 'False').lower() == 'true'
DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///db.sqlite3')
SECRET_KEY = os.environ.get('SECRET_KEY', 'default-key')

Flask 配置

# config.py
import osclass Config:DEBUG = os.environ.get('FLASK_DEBUG', False)DATABASE_URL = os.environ.get('DATABASE_URL', 'sqlite:///app.db')SECRET_KEY = os.environ.get('SECRET_KEY', 'dev')app.config.from_object(Config)

FastAPI 配置

# config.py
from pydantic_settings import BaseSettingsclass Settings(BaseSettings):debug: bool = Falsedatabase_url: str = "sqlite:///./test.db"secret_key: str = "secret"class Config:env_file = ".env"settings = Settings()

Sanic 配置

# config.py
import osclass Config:DEBUG = os.environ.get('DEBUG', False)HOST = os.environ.get('HOST', '0.0.0.0')PORT = int(os.environ.get('PORT', 8000))app.config.update(Config.__dict__)