Python基础(Flask①)

发布于:2025-08-17 ⋅ 阅读:(15) ⋅ 点赞:(0)

蓝图

在 Flask 中,蓝图(Blueprint)是一种模块化组织代码的工具,它可以将一个大型 Flask 应用拆分成多个独立的、可重用的组件。

我们可以创建一个目录结构如下:

myapp/
├── app.py          # 应用工厂函数
├── api/            # API蓝图目录
│   ├── __init__.py
│   ├── users.py    # 用户相关接口
│   └── posts.py    # 文章相关接口
└── run.py          # 启动文件

posts.py

from flask import Blueprint, jsonify

# 创建文章蓝图
posts_bp = Blueprint('posts', __name__)

# 模拟数据库
posts = [
    {'id': 1, 'title': 'Flask蓝图教程', 'content': '这是一篇关于Flask蓝图的教程'},
    {'id': 2, 'title': 'Python基础', 'content': 'Python是一种简单易学的编程语言'}
]

@posts_bp.route('', methods=['GET'])
def get_posts():
    """获取所有文章"""
    return jsonify({
        'success': True,
        'data': posts,
        'count': len(posts),
        'message': '获取文章列表成功'
    })

@posts_bp.route('/<int:post_id>', methods=['GET'])
def get_post(post_id):
    """获取单篇文章"""
    post = next((p for p in posts if p['id'] == post_id), None)
    if post:
        return jsonify({
            'success': True,
            'data': post,
            'message': f'获取ID为{post_id}的文章成功'
        })
    return jsonify({
        'success': False,
        'data': None,
        'message': f'未找到ID为{post_id}的文章'
    }), 404
    

users.py

from flask import Blueprint, jsonify, request

# 创建用户蓝图
users_bp = Blueprint('users', __name__)

# 模拟数据库
users = [
    {'id': 1, 'name': 'Alice', 'email': 'alice@example.com'},
    {'id': 2, 'name': 'Bob', 'email': 'bob@example.com'}
]

@users_bp.route('', methods=['GET'])
def get_users():
    """获取所有用户"""
    return jsonify({
        'success': True,
        'data': users,
        'message': '获取用户列表成功'
    })

@users_bp.route('/<int:user_id>', methods=['GET'])
def get_user(user_id):
    """获取单个用户"""
    user = next((u for u in users if u['id'] == user_id), None)
    if user:
        return jsonify({
            'success': True,
            'data': user,
            'message': f'获取ID为{user_id}的用户成功'
        })
    return jsonify({
        'success': False,
        'data': None,
        'message': f'未找到ID为{user_id}的用户'
    }), 404

@users_bp.route('', methods=['POST'])
def create_user():
    """创建新用户"""
    if not request.json or 'name' not in request.json:
        return jsonify({
            'success': False,
            'data': None,
            'message': '缺少用户名'
        }), 400
    
    user = {
        'id': len(users) + 1,
        'name': request.json['name'],
        'email': request.json.get('email', '')
    }
    users.append(user)
    
    return jsonify({
        'success': True,
        'data': user,
        'message': '用户创建成功'
    }), 201
    

app.py

from flask import Flask

def create_app():
    app = Flask(__name__)
    
    # 注册蓝图
    from api.users import users_bp
    from api.posts import posts_bp
    
    app.register_blueprint(users_bp, url_prefix='/api/users')
    app.register_blueprint(posts_bp, url_prefix='/api/posts')
    
    return app
    

run.py

from app import create_app

app = create_app()

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

自定义装饰器

比如 @login_required装饰器就像一个 “门卫”,在用户访问某个路由前先检查条件,不满足就拦住。

示例:自己写一个 @login_required

from flask import session, redirect, url_for
from functools import wraps

# 定义装饰器:检查用户是否登录
def login_required(f):
    @wraps(f)  # 保留原函数的信息(如函数名)
    def decorated_function(*args, **kwargs):
        # 检查 Session 中是否有 user_id(登录标识)
        if 'user_id' not in session:
            # 未登录 → 跳转到登录页
            return redirect(url_for('login'))
        # 已登录 → 正常执行原函数(如访问个人中心)
        return f(*args, **kwargs)
    return decorated_function

# 使用装饰器保护路由
@app.route('/profile')
@login_required  # 加上这行,访问 /profile 必须先登录
def profile():
    return "这是你的个人中心,只有登录后才能看"

Flask-Principal 

先定义规则:user_permission = Permission(user_role)(“要进普通厅,必须有电影票”)。
给用户发凭证:identity.provides.add(user_role)(“给张三发电影票”)。
检票放行:@user_permission.require()(“张三有电影票,允许进普通厅”)。

app.py

from flask import Flask, session, redirect, url_for
from flask_principal import (
    Principal, Permission, RoleNeed,  # 核心类
    identity_loaded, Identity, AnonymousIdentity,  # 身份相关
    PermissionDenied  # 权限不足异常
)

app = Flask(__name__)
app.secret_key = 'your-secret-key-here'  # 必须设置,用于加密session

# 1. 初始化Flask-Principal
principal = Principal(app)

# 2. 定义角色(本质是特殊的Need)
user_role = RoleNeed('user')    # 普通用户角色
admin_role = RoleNeed('admin')  # 管理员角色

# 3. 定义权限(基于角色)
# 普通用户权限:需要user角色
user_permission = Permission(user_role)
# 管理员权限:需要admin角色(包含普通用户权限)
admin_permission = Permission(admin_role)

关联用户和角色(关键步骤)

用户登录后,需要告诉系统 “这个用户是什么角色”。通过 identity_loaded 信号实现:

# 继续在app.py中添加
@identity_loaded.connect_via(app)
def on_identity_loaded(sender, identity):
    """当用户身份被加载时,给用户分配角色"""
    # 从session中获取当前登录用户ID(假设登录时已存入)
    user_id = session.get('user_id')
    
    if user_id:
        # 这里简化处理:假设id=1的是管理员,其他是普通用户
        if user_id == 1:
            # 给管理员添加admin角色
            identity.provides.add(admin_role)
        # 所有登录用户都有user角色
        identity.provides.add(user_role)
    else:
        # 未登录用户,身份为匿名
        identity = AnonymousIdentity()

 用权限保护路由

用权限装饰器限制路由访问,不满足权限就会被拒绝:

# 继续在app.py中添加
# 首页:任何人可访问
@app.route('/')
def index():
    return "首页 - 所有人可见"

# 登录模拟:实际项目中需验证账号密码
@app.route('/login/<int:user_id>')
def login(user_id):
    # 登录成功,将用户ID存入session
    session['user_id'] = user_id
    return f"登录成功(用户ID:{user_id})"

# 发帖功能:仅登录用户(user角色)可访问
@app.route('/post')
@user_permission.require()  # 要求有user权限
def post():
    return "发表帖子成功(仅登录用户可见)"

# 删帖功能:仅管理员(admin角色)可访问
@app.route('/delete')
@admin_permission.require()  # 要求有admin权限
def delete():
    return "删除帖子成功(仅管理员可见)"

# 处理权限不足的情况(返回友好提示)
@app.errorhandler(PermissionDenied)
def handle_permission_denied(error):
    return "权限不足!你没有访问该页面的权限。", 403

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

上下文

全局变量是 “所有人共用的固定容器”,适合存不常变的全局信息(如程序版本)。
上下文是 “每个场景专属的临时容器”,适合存随场景变化的动态信息(如请求数据、用户会话)。

上下文像 “你手里的购物袋”,只在你逛街这段时间(特定场景)有效,里面装的是你当前买的东西(和你相关的临时信息),换个人逛街(另一个请求),购物袋里的东西就不一样了。

应用上下文(current_app

作用:存储当前 Flask 应用的全局信息(如配置、扩展实例等)。
适用场景:需要在整个应用中共享的资源,比如数据库连接、应用配置等。
核心对象:current_app(代表当前运行的 Flask 应用实例)。

from flask import Flask, current_app

app = Flask(__name__)
app.config['MAX_CONTENT_LENGTH'] = 1024 * 1024  # 配置最大上传大小

@app.route('/')
def index():
    # 在视图中通过 current_app 获取应用配置
    max_size = current_app.config['MAX_CONTENT_LENGTH']
    return f"应用最大上传限制:{max_size} bytes"

# 在非视图函数中使用(需手动激活上下文)
def get_app_name():
    # 非视图函数中使用 current_app 前,需手动创建应用上下文
    with app.app_context():
        return current_app.name  # 返回应用名称(__name__)

请求上下文(requestsession

作用:存储当前 HTTP 请求的信息,仅在处理请求时有效。
适用场景:处理用户请求相关的数据,如表单数据、Cookie、会话信息等。
核心对象:
request:封装当前请求的所有数据(如 URL、表单、headers 等)。
session:用户会话对象,用于存储跨请求的用户状态(如登录信息)。

from flask import request, session

@app.route('/login', methods=['POST'])
def login():
    # 通过 request 获取表单数据
    username = request.form.get('username')
    password = request.form.get('password')
    
    # 验证通过后,用 session 记录用户状态
    if username == 'admin' and password == '123':
        session['user_id'] = 1  # 存储用户ID到会话
        return "登录成功"
    return "账号或密码错误"

@app.route('/user')
def user_info():
    # 通过 session 获取用户状态
    if 'user_id' in session:
        return f"当前登录用户ID:{session['user_id']}"
    return "未登录"

自定义上下文变量

模板上下文处理器(供模板使用)

代码示例:

from flask import Flask, render_template, session, appcontext_pushed
app = Flask(__name__)
app.secret_key = 'test_key'  # 必须设置,session需要

# 定义上下文处理器:向所有模板注入变量
@app.context_processor
def inject_global_vars():
    # 1. 网站名称(固定值)
    site_name = "我的博客"
    
    # 2. 当前登录用户(从session中获取,未登录则为None)
    current_user = None
    if 'user_id' in session:
        # 实际项目中这里会从数据库查用户信息,这里简化
        current_user = {"id": session['user_id'], "name": "张三"}
    
    # 返回一个字典,键就是模板中可用的变量名
    return {
        'site_name': site_name,
        'current_user': current_user
    }

# 测试视图:不用传参,模板直接能用上面的变量
@app.route('/')
def index():
    # 注意:这里没有传 site_name 或 current_user 给模板
    return render_template('index.html')

@app.route('/login')
def login():
    # 模拟登录:把用户ID存入session
    session['user_id'] = 1
    return "登录成功,<a href='/'>回首页</a>"

@app.route('/logout')
def logout():
    # 模拟登出:删除session中的用户ID
    session.pop('user_id', None)
    return "登出成功,<a href='/'>回首页</a>"

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

模板文件(templates/index.html):

<!-- 直接使用上下文处理器注入的变量 -->
<h1>{{ site_name }}</h1>  <!-- 显示“我的博客” -->

{% if current_user %}
    <p>欢迎您,{{ current_user.name }}(ID: {{ current_user.id }})</p>
    <a href="/logout">登出</a>
{% else %}
    <p>请 <a href="/login">登录</a></p>
{% endif %}

访问 / 时,模板自动显示 site_name 和 current_user,不用在 index 视图里传参。
登录 / 登出后,current_user 会自动更新,所有模板都能同步看到变化。

应用上下文变量(供代码使用)

代码示例:

from flask import Flask, g, session, redirect, url_for
app = Flask(__name__)
app.secret_key = 'test_key'

# 1. 在请求处理前,加载用户信息到 g
@app.before_request
def load_user_before_request():
    # 每次收到请求时,先执行这个函数
    g.user = None  # 初始化,避免未登录时出错
    if 'user_id' in session:
        # 模拟从数据库查询用户信息(实际项目中这里会有数据库操作)
        g.user = {
            "id": session['user_id'],
            "name": "张三",
            "role": "admin"  # 假设是管理员
        }

# 2. 视图函数中使用 g.user
@app.route('/profile')
def profile():
    if g.user is None:
        return redirect(url_for('login'))
    
    # 直接用 g.user,不用再查数据库
    return f"""
    用户资料:<br>
    ID: {g.user['id']}<br>
    姓名: {g.user['name']}<br>
    角色: {g.user['role']}
    """

# 3. 另一个函数也能用 g.user(比如权限检查)
def check_admin_permission():
    # 直接从 g 中取用户角色,不用重新查询
    return g.user and g.user['role'] == 'admin'

@app.route('/admin')
def admin_panel():
    if not check_admin_permission():
        return "没有管理员权限"
    return "管理员面板(只有管理员能看)"

# 登录/登出接口(和之前一样)
@app.route('/login')
def login():
    session['user_id'] = 1
    return "登录成功,<a href='/profile'>去个人中心</a> | <a href='/admin'>去管理员面板</a>"

@app.route('/logout')
def logout():
    session.pop('user_id', None)
    return "登出成功,<a href='/login'>重新登录</a>"

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

用户登录后访问 /profile,load_user_before_request 会先把用户信息存到 g.user,视图函数直接使用。
访问 /admin 时,check_admin_permission 函数也能直接从 g.user 取数据,不用再查数据库。
这次请求处理完后,g 里的数据会自动清空,下次请求重新加载,不会影响其他用户。