Flask框架入门与实战

Li JiaXin大约 12 分钟

Flask框架入门与实战

Flask就像是Python Web开发界的"瑞士军刀"——小巧、灵活、功能齐全。作为一名测试开发工程师，我发现Flask特别适合快速搭建测试工具和小型服务。它的"微框架"设计理念让你可以从最简单的Hello World开始，逐步构建复杂的测试平台。

一、Flask简介：为什么选择Flask？

Flask的特点

Flask被称为"微框架"，但这个"微"不是指功能少，而是指核心简单、扩展性强：

轻量级：核心代码简洁，学习曲线平缓
灵活性：不强制使用特定的数据库或模板引擎
扩展丰富：有大量第三方扩展可选
文档完善：官方文档清晰易懂

适用场景

在测试开发工作中，Flask特别适合：

快速原型开发
测试工具和小型服务
API接口开发
学习Web开发基础概念

二、Flask快速入门

1. 安装Flask

# 创建虚拟环境（推荐）
python -m venv flask_env
source flask_env/bin/activate  # Linux/Mac
# flask_env\Scripts\activate  # Windows

# 安装Flask
pip install Flask

2. 第一个Flask应用

# app.py
from flask import Flask

# 创建Flask应用实例
app = Flask(__name__)

@app.route('/')
def hello_world():
    """首页路由"""
    return 'Hello World! 这是我的第一个Flask应用！'

@app.route('/test')
def test_page():
    """测试页面"""
    return '这是测试页面，专门用来验证Flask是否正常工作'

if __name__ == '__main__':
    # 启动开发服务器
    app.run(debug=True, host='0.0.0.0', port=5000)

启动应用：

python app.py

访问 http://localhost:5000 就能看到你的第一个Flask应用了！

💡 测试工程师小贴士：debug=True 在开发时非常有用，代码修改后会自动重启，就像热部署一样方便！

3. 路由与参数处理

from flask import Flask, request

app = Flask(__name__)

# 基础路由
@app.route('/')
def index():
    return '欢迎来到测试平台首页'

# 带参数的路由
@app.route('/user/<int:user_id>')
def show_user(user_id):
    """显示用户信息"""
    return f'用户ID: {user_id}'

@app.route('/testcase/<string:case_name>')
def show_testcase(case_name):
    """显示测试用例"""
    return f'测试用例: {case_name}'

# 支持多种HTTP方法
@app.route('/api/data', methods=['GET', 'POST'])
def handle_data():
    if request.method == 'GET':
        return '这是GET请求'
    elif request.method == 'POST':
        return '这是POST请求'

4. 请求参数处理

from flask import Flask, request, jsonify

app = Flask(__name__)

def parse_request_data(req):
    """统一解析请求数据"""
    if req.method == 'POST':
        if req.is_json:
            return req.get_json()
        else:
            return req.form.to_dict()
    elif req.method == 'GET':
        return req.args.to_dict()
    return {}

@app.route('/search', methods=['GET', 'POST'])
def search_testcases():
    """搜索测试用例"""
    data = parse_request_data(request)
    
    # 获取查询参数
    keyword = data.get('keyword', '')
    page = int(data.get('page', 1))
    size = int(data.get('size', 10))
    
    # 模拟搜索结果
    results = [
        {'id': 1, 'name': f'包含"{keyword}"的测试用例1'},
        {'id': 2, 'name': f'包含"{keyword}"的测试用例2'}
    ]
    
    return jsonify({
        'keyword': keyword,
        'page': page,
        'size': size,
        'total': len(results),
        'results': results
    })

# GET请求示例: /search?keyword=登录&page=1&size=10
# POST请求示例: {"keyword": "登录", "page": 1, "size": 10}

5. 请求对象详解

Flask的request对象包含了请求的所有信息：

属性	说明	示例
`request.method`	HTTP方法	GET, POST, PUT, DELETE
`request.args`	URL查询参数	?name=john&age=25
`request.form`	表单数据	HTML表单提交的数据
`request.json`	JSON数据	POST请求的JSON体
`request.files`	上传文件	文件上传
`request.headers`	请求头	Content-Type, Authorization等
`request.cookies`	Cookie信息	用户会话数据

@app.route('/debug', methods=['GET', 'POST'])
def debug_request():
    """调试请求信息"""
    info = {
        'method': request.method,
        'url': request.url,
        'headers': dict(request.headers),
        'args': request.args.to_dict(),
        'form': request.form.to_dict() if request.form else None,
        'json': request.get_json() if request.is_json else None,
        'cookies': request.cookies.to_dict()
    }
    return jsonify(info)

三、模板与静态文件

1. 使用Jinja2模板

from flask import Flask, render_template

app = Flask(__name__)

@app.route('/dashboard')
def dashboard():
    """测试仪表板"""
    test_data = {
        'total_tests': 100,
        'passed': 85,
        'failed': 10,
        'skipped': 5,
        'recent_tests': [
            {'name': '登录功能测试', 'status': 'passed', 'duration': '2.3s'},
            {'name': '支付功能测试', 'status': 'failed', 'duration': '5.1s'},
            {'name': '搜索功能测试', 'status': 'passed', 'duration': '1.8s'}
        ]
    }
    return render_template('dashboard.html', data=test_data)

创建模板文件 templates/dashboard.html：

<!DOCTYPE html>
<html>
<head>
    <title>测试仪表板</title>
    <style>
        .stats { display: flex; gap: 20px; margin: 20px 0; }
        .stat-card { padding: 20px; border: 2px solid #ddd; border-radius: 5px; }
        .passed { background-color: #d4edda; }
        .failed { background-color: #f8d7da; }
        .skipped { background-color: #fff3cd; }
    </style>
</head>
<body>
    <h1>测试仪表板</h1>
    
    <div class="stats">
        <div class="stat-card">
            <h3>总测试数</h3>
            <p>{{ data.total_tests }}</p>
        </div>
        <div class="stat-card passed">
            <h3>通过</h3>
            <p>{{ data.passed }}</p>
        </div>
        <div class="stat-card failed">
            <h3>失败</h3>
            <p>{{ data.failed }}</p>
        </div>
        <div class="stat-card skipped">
            <h3>跳过</h3>
            <p>{{ data.skipped }}</p>
        </div>
    </div>
    
    <h2>最近测试</h2>
    <table border="1">
        <tr>
            <th>测试名称</th>
            <th>状态</th>
            <th>耗时</th>
        </tr>
        {% for test in data.recent_tests %}
        <tr>
            <td>{{ test.name }}</td>
            <td>{{ test.status }}</td>
            <td>{{ test.duration }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>

2. 静态文件处理

from flask import Flask, url_for

app = Flask(__name__)

@app.route('/static-demo')
def static_demo():
    """静态文件演示"""
    return f'''
    <h1>静态文件演示</h1>
    <img src="{url_for('static', filename='images/logo.png')}" alt="Logo">
    <link rel="stylesheet" href="{url_for('static', filename='css/style.css')}">
    <script src="{url_for('static', filename='js/app.js')}"></script>
    '''

项目结构：

your_project/
├── app.py
├── templates/
│   └── dashboard.html
└── static/
    ├── css/
    │   └── style.css
    ├── js/
    │   └── app.js
    └── images/
        └── logo.png

四、实战案例：测试用例管理API

让我们构建一个简单的测试用例管理系统：

from flask import Flask, request, jsonify
from datetime import datetime

app = Flask(__name__)

# 模拟数据库
testcases = [
    {
        'id': 1,
        'name': '用户登录测试',
        'description': '验证用户登录功能',
        'status': 'active',
        'created_at': '2024-01-01 10:00:00'
    },
    {
        'id': 2,
        'name': '支付功能测试',
        'description': '验证支付流程',
        'status': 'active',
        'created_at': '2024-01-02 11:00:00'
    }
]

@app.route('/api/testcases', methods=['GET'])
def get_testcases():
    """获取测试用例列表"""
    page = int(request.args.get('page', 1))
    size = int(request.args.get('size', 10))
    keyword = request.args.get('keyword', '')
    
    # 过滤
    filtered_cases = testcases
    if keyword:
        filtered_cases = [
            case for case in testcases 
            if keyword.lower() in case['name'].lower()
        ]
    
    # 分页
    start = (page - 1) * size
    end = start + size
    paginated_cases = filtered_cases[start:end]
    
    return jsonify({
        'code': 200,
        'message': 'success',
        'data': {
            'total': len(filtered_cases),
            'page': page,
            'size': size,
            'items': paginated_cases
        }
    })

@app.route('/api/testcases', methods=['POST'])
def create_testcase():
    """创建测试用例"""
    data = request.get_json()
    
    # 验证必填字段
    if not data or not data.get('name'):
        return jsonify({
            'code': 400,
            'message': '测试用例名称不能为空'
        }), 400
    
    # 创建新测试用例
    new_case = {
        'id': len(testcases) + 1,
        'name': data['name'],
        'description': data.get('description', ''),
        'status': 'active',
        'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    }
    
    testcases.append(new_case)
    
    return jsonify({
        'code': 201,
        'message': '测试用例创建成功',
        'data': new_case
    }), 201

@app.route('/api/testcases/<int:case_id>', methods=['GET'])
def get_testcase(case_id):
    """获取单个测试用例"""
    case = next((c for c in testcases if c['id'] == case_id), None)
    
    if not case:
        return jsonify({
            'code': 404,
            'message': '测试用例不存在'
        }), 404
    
    return jsonify({
        'code': 200,
        'message': 'success',
        'data': case
    })

@app.route('/api/testcases/<int:case_id>', methods=['PUT'])
def update_testcase(case_id):
    """更新测试用例"""
    case = next((c for c in testcases if c['id'] == case_id), None)
    
    if not case:
        return jsonify({
            'code': 404,
            'message': '测试用例不存在'
        }), 404
    
    data = request.get_json()
    
    # 更新字段
    if 'name' in data:
        case['name'] = data['name']
    if 'description' in data:
        case['description'] = data['description']
    if 'status' in data:
        case['status'] = data['status']
    
    return jsonify({
        'code': 200,
        'message': '测试用例更新成功',
        'data': case
    })

@app.route('/api/testcases/<int:case_id>', methods=['DELETE'])
def delete_testcase(case_id):
    """删除测试用例"""
    global testcases
    
    case = next((c for c in testcases if c['id'] == case_id), None)
    
    if not case:
        return jsonify({
            'code': 404,
            'message': '测试用例不存在'
        }), 404
    
    testcases = [c for c in testcases if c['id'] != case_id]
    
    return jsonify({
        'code': 200,
        'message': '测试用例删除成功'
    })

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

五、Flask扩展与中间件

1. 错误处理

from flask import Flask, jsonify

app = Flask(__name__)

@app.errorhandler(404)
def not_found(error):
    """处理404错误"""
    return jsonify({
        'code': 404,
        'message': '请求的资源不存在',
        'error': 'Not Found'
    }), 404

@app.errorhandler(500)
def internal_error(error):
    """处理500错误"""
    return jsonify({
        'code': 500,
        'message': '服务器内部错误',
        'error': 'Internal Server Error'
    }), 500

@app.errorhandler(Exception)
def handle_exception(e):
    """处理所有未捕获的异常"""
    return jsonify({
        'code': 500,
        'message': f'发生错误: {str(e)}',
        'error': 'Exception'
    }), 500

2. 请求钩子（中间件）

from flask import Flask, request, g
import time

app = Flask(__name__)

@app.before_request
def before_request():
    """请求前处理"""
    g.start_time = time.time()
    print(f"📥 {request.method} {request.url}")

@app.after_request
def after_request(response):
    """请求后处理"""
    if hasattr(g, 'start_time'):
        duration = time.time() - g.start_time
        response.headers['X-Response-Time'] = f"{duration:.3f}s"
        print(f"📤 {response.status_code} - {duration:.3f}s")
    return response

@app.teardown_request
def teardown_request(exception):
    """请求结束时清理"""
    if exception:
        print(f"❌ 请求处理出错: {exception}")

3. 蓝图（Blueprint）模块化

# blueprints/auth.py
from flask import Blueprint, request, jsonify

auth_bp = Blueprint('auth', __name__, url_prefix='/auth')

@auth_bp.route('/login', methods=['POST'])
def login():
    """用户登录"""
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')

    # 简单验证
    if username == 'admin' and password == '123456':
        return jsonify({
            'code': 200,
            'message': '登录成功',
            'token': 'fake-jwt-token'
        })

    return jsonify({
        'code': 401,
        'message': '用户名或密码错误'
    }), 401

@auth_bp.route('/logout', methods=['POST'])
def logout():
    """用户登出"""
    return jsonify({
        'code': 200,
        'message': '登出成功'
    })

# blueprints/api.py
from flask import Blueprint, jsonify

api_bp = Blueprint('api', __name__, url_prefix='/api')

@api_bp.route('/health')
def health_check():
    """健康检查"""
    return jsonify({
        'status': 'healthy',
        'service': 'test-platform',
        'version': '1.0.0'
    })

@api_bp.route('/stats')
def get_stats():
    """获取统计信息"""
    return jsonify({
        'total_tests': 100,
        'passed': 85,
        'failed': 10,
        'skipped': 5
    })

# main.py
from flask import Flask
from blueprints.auth import auth_bp
from blueprints.api import api_bp

app = Flask(__name__)

# 注册蓝图
app.register_blueprint(auth_bp)
app.register_blueprint(api_bp)

@app.route('/')
def index():
    return '测试平台主页'

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

六、配置管理

1. 配置类

import os

class Config:
    """基础配置"""
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
    DEBUG = False
    TESTING = False

class DevelopmentConfig(Config):
    """开发环境配置"""
    DEBUG = True
    DATABASE_URL = 'sqlite:///dev.db'

class TestingConfig(Config):
    """测试环境配置"""
    TESTING = True
    DATABASE_URL = 'sqlite:///test.db'

class ProductionConfig(Config):
    """生产环境配置"""
    DATABASE_URL = os.environ.get('DATABASE_URL') or 'sqlite:///prod.db'

# 配置字典
config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

# app.py
from flask import Flask
from config import config

def create_app(config_name='default'):
    """应用工厂函数"""
    app = Flask(__name__)
    app.config.from_object(config[config_name])

    # 注册蓝图
    from blueprints.auth import auth_bp
    from blueprints.api import api_bp
    app.register_blueprint(auth_bp)
    app.register_blueprint(api_bp)

    return app

if __name__ == '__main__':
    import os
    config_name = os.environ.get('FLASK_ENV', 'development')
    app = create_app(config_name)
    app.run()

七、文件上传处理

from flask import Flask, request, jsonify
import os
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB

# 确保上传目录存在
os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif', 'xml', 'json'}

def allowed_file(filename):
    """检查文件类型是否允许"""
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['POST'])
def upload_file():
    """文件上传"""
    if 'file' not in request.files:
        return jsonify({
            'code': 400,
            'message': '没有选择文件'
        }), 400

    file = request.files['file']

    if file.filename == '':
        return jsonify({
            'code': 400,
            'message': '没有选择文件'
        }), 400

    if file and allowed_file(file.filename):
        filename = secure_filename(file.filename)
        file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
        file.save(file_path)

        return jsonify({
            'code': 200,
            'message': '文件上传成功',
            'filename': filename,
            'file_size': os.path.getsize(file_path)
        })

    return jsonify({
        'code': 400,
        'message': '不支持的文件类型'
    }), 400

@app.route('/upload/batch', methods=['POST'])
def batch_upload():
    """批量文件上传"""
    files = request.files.getlist('files')

    if not files:
        return jsonify({
            'code': 400,
            'message': '没有选择文件'
        }), 400

    results = []
    for file in files:
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            file_path = os.path.join(app.config['UPLOAD_FOLDER'], filename)
            file.save(file_path)

            results.append({
                'filename': filename,
                'status': 'success',
                'file_size': os.path.getsize(file_path)
            })
        else:
            results.append({
                'filename': file.filename if file else 'unknown',
                'status': 'failed',
                'reason': '不支持的文件类型'
            })

    return jsonify({
        'code': 200,
        'message': '批量上传完成',
        'results': results
    })

八、部署与生产环境

1. 使用Gunicorn部署

# 安装Gunicorn
pip install gunicorn

# 启动应用
gunicorn -w 4 -b 0.0.0.0:8000 app:app

# 或者使用配置文件
gunicorn -c gunicorn.conf.py app:app

创建 gunicorn.conf.py：

# Gunicorn配置文件
bind = "0.0.0.0:8000"
workers = 4
worker_class = "sync"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 100
timeout = 30
keepalive = 2
preload_app = True

2. Docker部署

创建 Dockerfile：

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "-c", "gunicorn.conf.py", "app:app"]

创建 docker-compose.yml：

version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - FLASK_ENV=production
    volumes:
      - ./uploads:/app/uploads

九、总结

Flask作为Python Web开发的经典框架，具有以下优势：

✅ 简单易学 - 最小化的核心，容易上手 ✅ 灵活扩展 - 丰富的第三方扩展生态 ✅ 文档完善 - 官方文档清晰详细 ✅ 社区活跃 - 大量的教程和示例 ✅ 适合小项目 - 快速原型和小型服务的首选

学习建议

从基础开始：掌握路由、模板、请求处理
实战练习：构建完整的CRUD应用
学习扩展：Flask-SQLAlchemy、Flask-Login等
部署实践：了解生产环境部署方案

🎯 下一步：建议学习Flask-SQLAlchemy进行数据库操作，或者对比学习FastAPI来了解现代Web框架的特性。

Flask虽然是"微框架"，但通过合理的扩展和组织，完全可以构建出功能完整的测试平台。它的简洁性让你能专注于业务逻辑，而不是框架本身的复杂性。

八、Flask + Peewee完整CRUD实战

项目结构设计

test_platform/
├── app.py              # 主应用文件
├── models.py           # 数据模型
├── config.py           # 配置文件
├── requirements.txt    # 依赖包
└── tests/             # 测试文件
    └── test_api.py

数据模型设计

# models.py
from peewee import *
from datetime import datetime
import os

# 数据库配置
DATABASE_URL = os.getenv('DATABASE_URL', 'test_platform.db')
db = SqliteDatabase(DATABASE_URL)

class BaseModel(Model):
    """基础模型类"""
    created_at = DateTimeField(default=datetime.now)
    updated_at = DateTimeField(default=datetime.now)

    class Meta:
        database = db

    def save(self, *args, **kwargs):
        self.updated_at = datetime.now()
        return super().save(*args, **kwargs)

class Project(BaseModel):
    """项目模型"""
    name = CharField(max_length=100, unique=True)
    description = TextField(null=True)
    status = CharField(max_length=20, default='active')  # active, inactive

    def __str__(self):
        return self.name

class TestCase(BaseModel):
    """测试用例模型"""
    PRIORITY_CHOICES = [
        ('low', '低'),
        ('medium', '中'),
        ('high', '高'),
        ('critical', '紧急')
    ]

    STATUS_CHOICES = [
        ('pending', '待执行'),
        ('running', '执行中'),
        ('passed', '通过'),
        ('failed', '失败'),
        ('skipped', '跳过')
    ]

    name = CharField(max_length=200)
    project = ForeignKeyField(Project, backref='test_cases')
    priority = CharField(max_length=10, choices=PRIORITY_CHOICES, default='medium')
    status = CharField(max_length=10, choices=STATUS_CHOICES, default='pending')

    # 测试内容
    url = CharField(max_length=500)
    method = CharField(max_length=10, default='GET')
    headers = TextField(null=True)  # JSON格式存储
    body = TextField(null=True)     # JSON格式存储
    expected_status = IntegerField(default=200)

    # 执行结果
    actual_status = IntegerField(null=True)
    response_time = FloatField(null=True)
    response_body = TextField(null=True)
    error_message = TextField(null=True)

    # 统计信息
    run_count = IntegerField(default=0)
    success_count = IntegerField(default=0)

# 创建表
def create_tables():
    """创建数据库表"""
    with db:
        db.create_tables([Project, TestCase])

if __name__ == '__main__':
    create_tables()

完整API实现

# app.py
from flask import Flask, request, jsonify
from models import *
import json
import requests
import time
from datetime import datetime

app = Flask(__name__)

# 错误处理装饰器
def handle_errors(f):
    def wrapper(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except DoesNotExist:
            return jsonify({'error': '资源不存在'}), 404
        except Exception as e:
            return jsonify({'error': str(e)}), 500
    wrapper.__name__ = f.__name__
    return wrapper

# ============ 项目管理 API ============

@app.route('/api/projects', methods=['GET'])
@handle_errors
def get_projects():
    """获取项目列表"""
    page = int(request.args.get('page', 1))
    per_page = int(request.args.get('per_page', 10))

    query = Project.select()

    # 搜索过滤
    search = request.args.get('search')
    if search:
        query = query.where(Project.name.contains(search))

    # 分页
    total = query.count()
    projects = query.paginate(page, per_page)

    return jsonify({
        'projects': [
            {
                'id': p.id,
                'name': p.name,
                'description': p.description,
                'status': p.status,
                'test_cases_count': p.test_cases.count(),
                'created_at': p.created_at.isoformat()
            } for p in projects
        ],
        'pagination': {
            'page': page,
            'per_page': per_page,
            'total': total,
            'pages': (total + per_page - 1) // per_page
        }
    })

@app.route('/api/projects', methods=['POST'])
@handle_errors
def create_project():
    """创建项目"""
    data = request.get_json()

    # 验证必填字段
    if not data.get('name'):
        return jsonify({'error': '项目名称不能为空'}), 400

    # 检查名称是否已存在
    if Project.select().where(Project.name == data['name']).exists():
        return jsonify({'error': '项目名称已存在'}), 400

    project = Project.create(
        name=data['name'],
        description=data.get('description', ''),
        status=data.get('status', 'active')
    )

    return jsonify({
        'id': project.id,
        'name': project.name,
        'description': project.description,
        'status': project.status,
        'created_at': project.created_at.isoformat()
    }), 201

@app.route('/api/projects/<int:project_id>', methods=['PUT'])
@handle_errors
def update_project(project_id):
    """更新项目"""
    project = Project.get_by_id(project_id)
    data = request.get_json()

    # 检查名称唯一性（排除自己）
    if data.get('name') and data['name'] != project.name:
        if Project.select().where(
            (Project.name == data['name']) & (Project.id != project_id)
        ).exists():
            return jsonify({'error': '项目名称已存在'}), 400

    # 更新字段
    if 'name' in data:
        project.name = data['name']
    if 'description' in data:
        project.description = data['description']
    if 'status' in data:
        project.status = data['status']

    project.save()

    return jsonify({
        'id': project.id,
        'name': project.name,
        'description': project.description,
        'status': project.status,
        'updated_at': project.updated_at.isoformat()
    })

@app.route('/api/projects/<int:project_id>', methods=['DELETE'])
@handle_errors
def delete_project(project_id):
    """删除项目"""
    project = Project.get_by_id(project_id)

    # 检查是否有关联的测试用例
    if project.test_cases.count() > 0:
        return jsonify({'error': '项目下还有测试用例，无法删除'}), 400

    project.delete_instance()
    return jsonify({'message': '项目删除成功'})