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

Flask 全栈学习指南

news 来源：原创 2025/5/6 15:22:51

一、Flask 基础核心

1. 核心概念与启动流程

WSGI 与 Werkzeug

Flask 基于 Werkzeug 实现 WSGI 协议，处理 HTTP 请求到响应的全流程。

手动实现 WSGI 应用示例：

def simple_app(environ, start_response):
    status = '200 OK'
    headers = [('Content-type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello, WSGI World!']

最小化 Flask 应用

from flask import Flask
app = Flask(__name__)

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

if __name__ == '__main__':
    app.run(debug=True)  # 开发模式启动

二、路由系统详解

1. 路由规则与动态参数

参数类型限定

@app.route('/user/<int:user_id>')      # 只匹配整数
@app.route('/path/<path:subpath>')     # 匹配含斜杠的子路径
@app.route('/uuid/<uuid:uuid_val>')    # 必须符合 UUID 格式

HTTP 方法控制

@app.route('/api', methods=['GET', 'POST'])
def handle_api():
    if request.method == 'POST':
        return 'Received data'
    return 'Send data here'

2. URL 构建与反向解析

from flask import url_for

@app.route('/user/<username>')
def profile(username):
    return f'User {username}'

# 在代码或模板中生成 URL
url_for('profile', username='alice')  # 输出 /user/alice

三、请求与响应对象

1. 请求数据获取

表单与文件上传

from flask import request

username = request.form.get('username')
uploaded_file = request.files['file']
uploaded_file.save('/path/to/save')

JSON 数据与请求头

data = request.get_json()
user_agent = request.headers.get('User-Agent')

2. 自定义响应

设置 Cookie 与 Header

from flask import make_response

@app.route('/set-cookie')
def set_cookie():
    resp = make_response('Cookie set')
    resp.set_cookie('theme', 'dark')
    resp.headers['X-Custom-Header'] = '123'
    return resp

四、Jinja2 模板引擎

1. 模板继承与复用

基础模板 (base.html)

<!DOCTYPE html>
<html>
<head>
    <title>{% block title %}{% endblock %}</title>
</head>
<body>
    {% block content %}{% endblock %}
</body>
</html>

子模板扩展 (child.html)

{% extends "base.html" %}
{% block title %}Home Page{% endblock %}
{% block content %}
  <h1>Welcome, {{ username }}</h1>
{% endblock %}

2. 高级模板功能

宏定义（类似函数）

{% macro render_comment(comment) %}
  <div class="comment">
    <p>{{ comment.text }}</p>
    <small>By {{ comment.author }}</small>
  </div>
{% endmacro %}

{{ render_comment(comment) }}

过滤器与全局变量

# Python 中注册自定义过滤器
@app.template_filter('reverse')
def reverse_filter(s):
    return s[::-1]

# 模板中使用
{{ "hello" | reverse }}  <!-- 输出 "olleh" -->

五、数据库与 ORM（Flask-SQLAlchemy）

1. 模型定义与CRUD

定义数据模型

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)

# 创建表（首次运行）
with app.app_context():
    db.create_all()

增删改查操作

# 添加用户
new_user = User(username='alice', email='alice@example.com')
db.session.add(new_user)
db.session.commit()

# 查询用户
user = User.query.filter_by(username='alice').first()

2. 高级功能

多数据库绑定

app.config['SQLALCHEMY_BINDS'] = {
    'users': 'sqlite:///users.db',
    'posts': 'sqlite:///posts.db'
}

class Post(db.Model):
    __bind_key__ = 'posts'
    id = db.Column(db.Integer, primary_key=True)

数据库事件监听

from sqlalchemy import event

def before_insert_listener(mapper, connection, target):
    target.created_at = datetime.utcnow()

event.listen(User, 'before_insert', before_insert_listener)

六、用户认证（Flask-Login）

1. 基础配置

初始化与用户加载

from flask_login import LoginManager, UserMixin

login_manager = LoginManager(app)
login_manager.login_view = 'login'

class User(UserMixin, db.Model):
    # ... 模型字段 ...

@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

2. 权限控制

路由保护装饰器

from flask_login import login_required, current_user

@app.route('/dashboard')
@login_required
def dashboard():
    return f'Welcome, {current_user.username}'

角色权限管理

from functools import wraps

def admin_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not current_user.is_admin:
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

七、大型项目架构

1. 工厂模式与蓝图

工厂函数 (create_app)

# app/__init__.py
from flask import Flask
from .config import Config

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

    # 初始化扩展
    db.init_app(app)
    login.init_app(app)

    # 注册蓝图
    from .auth import bp as auth_bp
    app.register_blueprint(auth_bp, url_prefix='/auth')

    return app

蓝图模块化示例

# app/auth/routes.py
from flask import Blueprint

bp = Blueprint('auth', __name__)

@bp.route('/login')
def login():
    return 'Login Page'

2. 数据库迁移（Flask-Migrate）

迁移命令流程

# 初始化迁移仓库
flask db init

# 生成迁移脚本
flask db migrate -m "Initial migration"

# 应用迁移
flask db upgrade

# 回滚迁移
flask db downgrade

八、RESTful API 开发（Flask-RESTful）

1. 基础 API 设计

资源类与路由绑定

from flask_restful import Resource, Api

api = Api(app)

class UserAPI(Resource):
    def get(self, user_id):
        user = User.query.get_or_404(user_id)
        return {'username': user.username}

api.add_resource(UserAPI, '/api/users/<int:user_id>')

2. 高级功能

请求解析与数据验证

from flask_restful import reqparse

parser = reqparse.RequestParser()
parser.add_argument('username', type=str, required=True)
parser.add_argument('email', type=str, required=True)

class UserListAPI(Resource):
    def post(self):
        args = parser.parse_args()
        new_user = User(username=args['username'], email=args['email'])
        db.session.add(new_user)
        db.session.commit()
        return {'message': 'User created'}, 201

九、生产部署与优化

1. WSGI 服务器配置

Gunicorn 启动命令

gunicorn --workers 4 --bind 0.0.0.0:5000 "app:create_app()"

2. Nginx 反向代理

server {
    listen 80;
    server_name yourdomain.com;

    location / {
        proxy_pass http://localhost:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }

    location /static {
        alias /path/to/static;
        expires 30d;
    }
}

3. Docker 容器化

FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["gunicorn", "--workers=4", "--bind=0.0.0.0:5000", "app:create_app()"]

十、扩展学习路线

1. 企业级实践

异步任务（Celery）

from celery import Celery

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])

@celery.task
def send_email_async(recipient, message):
    # 异步发送邮件逻辑
    pass

微服务通信（gRPC/HTTP）

import requests

@app.route('/aggregate')
def aggregate_data():
    user_service = requests.get('http://user-service:5000/users')
    return user_service.json()

2. 性能监控（Prometheus）

from prometheus_flask_exporter import PrometheusMetrics

metrics = PrometheusMetrics(app)
metrics.info('app_info', 'Application Info', version='1.0.0')