本文目录导读：
我来为你详细介绍Python权限装饰器的实现案例：
基础版：简单角色权限校验
from functools import wraps
def check_permission(required_permission):
"""
基础权限装饰器
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 模拟从上下文中获取当前用户权限
current_user_permissions = get_current_user_permissions()
if required_permission in current_user_permissions:
return func(*args, **kwargs)
else:
raise PermissionError(f"缺少权限: {required_permission}")
return wrapper
return decorator
# 模拟获取用户权限
def get_current_user_permissions():
# 实际项目中会从数据库或缓存中获取
return ['read', 'write', 'delete'] # 当前用户拥有这些权限
# 使用示例
@check_permission('delete')
def delete_user(user_id):
print(f"删除用户: {user_id}")
# 测试
try:
delete_user(123) # 有权限，正常执行
except PermissionError as e:
print(e)
完整版：支持多权限校验
from functools import wraps
from enum import Enum
class Permission(Enum):
READ = 'read'
WRITE = 'write'
DELETE = 'delete'
ADMIN = 'admin'
class PermissionError(Exception):
pass
class User:
def __init__(self, username, roles=None):
self.username = username
self.roles = roles or []
self.permissions = set()
self._load_permissions()
def _load_permissions(self):
# 权限映射关系
role_permissions = {
'admin': {Permission.READ, Permission.WRITE, Permission.DELETE, Permission.ADMIN},
'editor': {Permission.READ, Permission.WRITE},
'viewer': {Permission.READ}
}
for role in self.roles:
if role in role_permissions:
self.permissions.update(role_permissions[role])
def require_permissions(*required_permissions, mode='all'):
"""
权限装饰器
:param required_permissions: 需要的权限列表
:param mode: 'all' 需要所有权限，'any' 只需要任一权限
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
# 获取当前用户
current_user = get_current_user()
if not current_user:
raise PermissionError("用户未登录")
if mode == 'all':
# 需要所有权限
if all(perm in current_user.permissions for perm in required_permissions):
return func(*args, **kwargs)
else:
missing = [perm.value for perm in required_permissions
if perm not in current_user.permissions]
raise PermissionError(f"缺少权限: {', '.join(missing)}")
elif mode == 'any':
# 只需要任一权限
if any(perm in current_user.permissions for perm in required_permissions):
return func(*args, **kwargs)
else:
raise PermissionError(f"需要以下任一权限: {', '.join(p.value for p in required_permissions)}")
raise PermissionError("无效的权限校验模式")
return wrapper
return decorator
# 全局用户上下文
_current_user = None
def set_current_user(user):
global _current_user
_current_user = user
def get_current_user():
return _current_user
# 使用示例
@require_permissions(Permission.DELETE)
def delete_user(user_id):
"""删除用户 - 需要删除权限"""
print(f"用户 {get_current_user().username} 删除了用户 {user_id}")
@require_permissions(Permission.READ, Permission.WRITE, mode='all')
def edit_document(doc_id):
"""编辑文档 - 需要读写权限"""
print(f"用户 {get_current_user().username} 编辑了文档 {doc_id}")
@require_permissions(Permission.ADMIN, Permission.WRITE, mode='any')
def special_operation():
"""特殊操作 - 需要管理员或写入权限"""
print(f"用户 {get_current_user().username} 执行了特殊操作")
# 测试代码
if __name__ == "__main__":
# 创建测试用户
admin_user = User("admin_user", roles=['admin'])
editor_user = User("editor_user", roles=['editor'])
viewer_user = User("viewer_user", roles=['viewer'])
# 测试1: 管理员用户
print("=== 管理员用户测试 ===")
set_current_user(admin_user)
try:
delete_user(1)
edit_document(100)
special_operation()
except PermissionError as e:
print(f"权限错误: {e}")
# 测试2: 编辑用户
print("\n=== 编辑用户测试 ===")
set_current_user(editor_user)
try:
delete_user(2) # 应该失败
except PermissionError as e:
print(f"预期错误: {e}")
try:
edit_document(101) # 应该成功
special_operation() # 应该成功
except PermissionError as e:
print(f"权限错误: {e}")
# 测试3: 只读用户
print("\n=== 只读用户测试 ===")
set_current_user(viewer_user)
try:
delete_user(3)
except PermissionError as e:
print(f"预期错误: {e}")
try:
edit_document(102)
except PermissionError as e:
print(f"预期错误: {e}")
企业级版本：支持缓存和日志
from functools import wraps
import logging
from datetime import datetime
from typing import Set, List, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PermissionCache:
"""权限缓存"""
def __init__(self):
self._cache = {}
self._ttl = 300 # 5分钟过期
def get(self, user_id: str) -> Optional[Set]:
if user_id in self._cache:
perms, timestamp = self._cache[user_id]
if (datetime.now() - timestamp).seconds < self._ttl:
return perms
else:
del self._cache[user_id]
return None
def set(self, user_id: str, permissions: Set):
self._cache[user_id] = (permissions, datetime.now())
# 全局权限缓存
perm_cache = PermissionCache()
def enhanced_permission_check(*required_permissions,
mode='all',
cache=True,
log_denied=True):
"""
增强版权限装饰器
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
user = get_current_user()
if not user:
error_msg = "用户未登录"
logger.warning(error_msg)
raise PermissionError(error_msg)
# 从缓存获取权限
user_permissions = None
if cache:
user_permissions = perm_cache.get(user.username)
if user_permissions is None:
# 重新加载权限
user_permissions = user.permissions
if cache:
perm_cache.set(user.username, user_permissions)
# 权限校验
if mode == 'all':
if all(perm in user_permissions for perm in required_permissions):
logger.info(f"用户 {user.username} 通过权限校验: {[p.value for p in required_permissions]}")
return func(*args, **kwargs)
elif mode == 'any':
if any(perm in user_permissions for perm in required_permissions):
logger.info(f"用户 {user.username} 通过权限校验 (任一): {[p.value for p in required_permissions]}")
return func(*args, **kwargs)
# 权限不足
if log_denied:
logger.warning(f"用户 {user.username} 权限不足. 需要: {[p.value for p in required_permissions]}, "
f"当前: {[p.value for p in user_permissions]}")
raise PermissionError(f"权限不足")
return wrapper
return decorator
使用方法总结
# 基础使用
@check_permission('delete')
def delete_user(user_id):
pass
# 多权限使用
@require_permissions(Permission.READ, Permission.WRITE, mode='all')
def edit_document(doc_id):
pass
@require_permissions(Permission.ADMIN, Permission.DELETE, mode='any')
def special_operation():
pass
# 增强版使用
@enhanced_permission_check(Permission.ADMIN, cache=True)
def admin_operation():
pass
关键特性
- 灵活性强：支持单权限、多权限、任意权限校验
- 缓存机制：缓存用户权限，提高性能
- 日志记录：记录权限校验结果，便于审计
- 错误处理：清晰的权限错误提示
- 可扩展：容易集成到现有系统
这个案例可以作为实际项目中权限系统的参考实现。
