Flask实现用户封禁功能完整代码示例

根据您的问题，我推测您想要一个实现“封禁/拉黑用户”功能的代码示例。这里提供一个基于Python和Flask的简单实现，包含用户管理和封禁逻辑。 ## 🚫 用户封禁功能实现（Python + Flask） ### 📦 依赖安装 ```bash pip install flask flask-sqlalchemy ``` ### ⚙️ 数据库模型设计 ```python from flask import Flask, request, jsonify from flask_sqlalchemy import SQLAlchemy from datetime import datetime, timedelta app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///users.db' db = SQLAlchemy(app) class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) is_banned = db.Column(db.Boolean, default=False) ban_expiry = db.Column(db.DateTime, nullable=True) # 封禁到期时间 created_at = db.Column(db.DateTime, default=datetime.utcnow) ``` ### 👤 用户注册与封禁API ```python @app.route('/register', methods=['POST']) def register(): data = request.json user = User(username=data['username']) db.session.add(user) db.session.commit() return jsonify({'message': '注册成功', 'user_id': user.id}) @app.route('/ban/<int:user_id>', methods=['POST']) def ban_user(user_id): user = User.query.get_or_404(user_id) data = request.json # 永久封禁 if data.get('permanent', False): user.is_banned = True user.ban_expiry = None else: # 临时封禁（默认24小时） hours = data.get('hours', 24) user.is_banned = True user.ban_expiry = datetime.utcnow() + timedelta(hours=hours) db.session.commit() return jsonify({ 'message': f'用户 {user.username} 已被封禁', 'expiry': str(user.ban_expiry) }) @app.route('/unban/<int:user_id>', methods=['POST']) def unban_user(user_id): user = User.query.get_or_404(user_id) user.is_banned = False user.ban_expiry = None db.session.commit() return jsonify({'message': f'用户 {user.username} 已解封'}) ``` ### 🔒 验证用户状态中间件 ```python from functools import wraps from flask import abort def check_ban_status(f): @wraps(f) def decorated(*args, **kwargs): # 假设通过 token 获取当前用户 user_id = kwargs.get('user_id') user = User.query.get(user_id) if user and user.is_banned: if user.ban_expiry and datetime.utcnow() > user.ban_expiry: # 自动解封过期的封禁 user.is_banned = False user.ban_expiry = None db.session.commit() else: abort(403, description='该用户已被封禁') return f(*args, **kwargs) return decorated # 应用中间件 @app.route('/protected/<int:user_id>') @check_ban_status def protected_resource(user_id): return jsonify({'message': '访问成功'}) ``` ### 🚀 运行测试 ```python if __name__ == '__main__': with app.app_context(): db.create_all() app.run(debug=True) ``` ## ⚠️ 开发注意要点 1. **安全性考虑**：实际应用中应使用JWT或Session验证用户身份，而非直接传递user_id 2. **数据库选择**：生产环境建议使用PostgreSQL或MySQL替代SQLite 3. **异步任务**：封禁通知可以通过Celery等异步任务发送给用户 4. **审计日志**：记录封禁操作的详细日志，包括操作人、时间、原因等 5. **批量操作**：如果支持批量封禁，需特别注意事务处理和性能优化 6. **时间处理**：建议统一使用UTC时间存储，显示时转换为用户本地时间 ######[AI问答 | 959点数解答 | 2026-07-18 04:07:45]