Flask 框架入门，从基础到实践 - Python Web

抽时间将一个简单的 Flask 项目的结构理了一遍，并配了一套示例代码。水平不高，欢迎提出问题和意见与我交流。

Web框架概述与选择

Python Web框架为开发者提供了构建Web应用程序的基础设施和工具集。目前主流的框架包括Django、Flask、FastAPI等，每个框架都有其独特的优势和适用场景。

主流框架对比

Django：全功能型框架，提供完整的MVC架构、ORM、管理后台等开箱即用的组件，适合中大型项目。

Flask：微框架，核心简洁但扩展性强，通过插件机制可以按需添加功能，适合中小型项目和API服务。

FastAPI：现代高性能框架，基于类型提示和异步支持，特别适合构建RESTful API和实时应用。

Flask框架深度入门

环境配置与项目初始化

首先创建虚拟环境并安装依赖：

python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

pip install flask python-dotenv

创建基础项目结构：

my_flask_app/
├── app/
│   ├── __init__.py
│   ├── routes.py
│   └── templates/
├── config.py
├── .env
└── run.py

基础应用搭建

在 app/__init__.py中初始化应用：

from flask import Flask
from config import Config

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)
  
    # 注册蓝图
    from app.routes import main_bp
    app.register_blueprint(main_bp)
  
    return app

配置文件 config.py：

import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False

路由与视图开发

在 app/routes.py中定义路由：

from flask import Blueprint, render_template, request, jsonify, redirect, url_for
from werkzeug.exceptions import BadRequest

main_bp = Blueprint('main', __name__)

@main_bp.route('/')
def index():
    return render_template('index.html', title='首页')

@main_bp.route('/user/<username>')
def user_profile(username):
    return render_template('profile.html', username=username)

@main_bp.route('/api/data', methods=['GET', 'POST'])
def api_data():
    if request.method == 'GET':
        return jsonify({'status': 'success', 'data': ['item1', 'item2']})
  
    elif request.method == 'POST':
        data = request.get_json()
        if not data or 'name' not in data:
            raise BadRequest('Missing required field: name')
  
        # 处理数据逻辑
        return jsonify({'status': 'created', 'id': 1}), 201

模板系统使用

创建基础模板 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 %}My Flask App{% endblock %}</title>
    <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
</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 App</a>
        </div>
    </nav>
  
    <main class="container mt-4">
        {% block content %}{% endblock %}
    </main>
  
    <script src="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/js/bootstrap.bundle.min.js"></script>
</body>
</html>

首页模板 templates/index.html：

{% extends "base.html" %}

{% block title %}{{ title }}{% endblock %}

{% block content %}
<div class="row">
    <div class="col-md-8">
        <h1>欢迎使用Flask</h1>
        <p class="lead">这是一个基于Flask构建的Web应用示例。</p>
  
        <div class="card mt-4">
            <div class="card-body">
                <h5 class="card-title">功能特性</h5>
                <ul class="list-group list-group-flush">
                    <li class="list-group-item">RESTful API支持</li>
                    <li class="list-group-item">Jinja2模板引擎</li>
                    <li class="list-group-item">数据库集成</li>
                    <li class="list-group-item">用户认证系统</li>
                </ul>
            </div>
        </div>
    </div>
</div>
{% endblock %}

数据库集成与ORM

SQLAlchemy配置

安装数据库依赖：

pip install flask-sqlalchemy flask-migrate

在应用中集成数据库：

from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

db = SQLAlchemy()
migrate = Migrate()

def create_app():
    app = Flask(__name__)
    app.config.from_object(Config)
  
    # 初始化数据库
    db.init_app(app)
    migrate.init_app(app, db)
  
    # 注册蓝图
    from app.routes import main_bp
    app.register_blueprint(main_bp)
  
    return app

数据模型定义

创建用户模型 app/models.py：

from app import db
from datetime import datetime
from werkzeug.security import generate_password_hash, check_password_hash

class User(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))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
  
    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 to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat()
        }

数据库迁移

初始化迁移环境并创建迁移：

flask db init
flask db migrate -m "Initial migration"
flask db upgrade

用户认证系统

登录功能实现

安装认证相关依赖：

pip install flask-login flask-wtf

创建登录表单 app/forms.py：

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired, Email, Length

class LoginForm(FlaskForm):
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('登录')

class RegistrationForm(FlaskForm):
    username = StringField('用户名', validators=[DataRequired(), Length(min=2, max=64)])
    email = StringField('邮箱', validators=[DataRequired(), Email()])
    password = PasswordField('密码', validators=[DataRequired(), Length(min=6)])
    submit = SubmitField('注册')

扩展认证路由：

from flask import flash, redirect, url_for
from flask_login import login_user, logout_user, login_required, current_user
from app.forms import LoginForm, RegistrationForm
from app.models import User

@main_bp.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))
  
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user and user.check_password(form.password.data):
            login_user(user, remember=True)
            next_page = request.args.get('next')
            return redirect(next_page) if next_page else redirect(url_for('main.index'))
        else:
            flash('邮箱或密码错误', 'danger')
  
    return render_template('login.html', form=form)

@main_bp.route('/logout')
@login_required
def logout():
    logout_user()
    return redirect(url_for('main.index'))

@main_bp.route('/register', methods=['GET', 'POST'])
def register():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))
  
    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(username=form.username.data, email=form.email.data)
        user.set_password(form.password.data)
        db.session.add(user)
        db.session.commit()
        flash('注册成功！请登录。', 'success')
        return redirect(url_for('main.login'))
  
    return render_template('register.html', form=form)

RESTful API开发

API蓝图创建

创建专门的API蓝图 app/api/__init__.py：

from flask import Blueprint

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

from app.api import users, posts

用户API端点 app/api/users.py：

from flask import jsonify, request
from app import db
from app.models import User
from . import api_bp

@api_bp.route('/users', methods=['GET'])
def get_users():
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
  
    users = User.query.paginate(
        page=page, per_page=per_page, error_out=False
    )
  
    return jsonify({
        'users': [user.to_dict() for user in users.items],
        'total': users.total,
        'pages': users.pages,
        'current_page': page
    })

@api_bp.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get_or_404(user_id)
    return jsonify(user.to_dict())

@api_bp.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
  
    if User.query.filter_by(username=data.get('username')).first():
        return jsonify({'error': '用户名已存在'}), 400
  
    if User.query.filter_by(email=data.get('email')).first():
        return jsonify({'error': '邮箱已存在'}), 400
  
    user = User(
        username=data['username'],
        email=data['email']
    )
    user.set_password(data['password'])
  
    db.session.add(user)
    db.session.commit()
  
    return jsonify(user.to_dict()), 201

错误处理与日志配置

自定义错误页面

@main_bp.app_errorhandler(404)
def not_found_error(error):
    if request.path.startswith('/api/'):
        return jsonify({'error': '资源未找到'}), 404
    return render_template('404.html'), 404

@main_bp.app_errorhandler(500)
def internal_error(error):
    db.session.rollback()
    if request.path.startswith('/api/'):
        return jsonify({'error': '服务器内部错误'}), 500
    return render_template('500.html'), 500

日志配置

在应用工厂函数中添加日志配置：

import logging
from logging.handlers import RotatingFileHandler
import os

def create_app():
    app = Flask(__name__)
    # ... 其他配置
  
    if not app.debug and not app.testing:
        if not os.path.exists('logs'):
            os.mkdir('logs')
  
        file_handler = RotatingFileHandler(
            'logs/flask_app.log', maxBytes=10240, backupCount=10
        )
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
        ))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)
        app.logger.setLevel(logging.INFO)
        app.logger.info('Flask application startup')
  
    return app

部署配置

生产环境配置

创建生产配置文件 config/production.py：

import os

class ProductionConfig:
    SECRET_KEY = os.environ.get('SECRET_KEY')
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
    DEBUG = False
    TESTING = False

Gunicorn配置

创建 gunicorn.conf.py：

bind = "0.0.0.0:8000"
workers = 4
worker_class = "sync"
max_requests = 1000
timeout = 30

启动脚本

创建应用启动文件 run.py：

from app import create_app

app = create_app()

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

测试与质量保证

单元测试示例

创建测试文件 tests/test_basic.py：

import unittest
from app import create_app, db
from app.models import User

class BasicTestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app()
        self.app.config['TESTING'] = True
        self.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
        self.client = self.app.test_client()
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
  
    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)
        self.assertIn(b'欢迎使用Flask', response.data)
  
    def test_user_creation(self):
        user = User(username='testuser', email='test@example.com')
        user.set_password('password123')
        db.session.add(user)
        db.session.commit()
  
        self.assertIsNotNone(user.id)
        self.assertTrue(user.check_password('password123'))
        self.assertFalse(user.check_password('wrongpassword'))

这个 Flask 框架涵盖了从环境配置到生产部署的全流程，包括路由设计、数据库集成、用户认证、API开发、错误处理等核心概念。通过这个基础框架，开发者可以快速构建功能完整的Web应用程序，并根据项目需求进一步扩展功能。