Python权限装饰器案例如何校验权限

本文目录导读：

1. 基础版：简单角色权限校验
2. 完整版：支持多权限校验
3. 企业级版本：支持缓存和日志
4. 使用方法总结
5. 关键特性

我来为你详细介绍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

关键特性

1. 灵活性强：支持单权限、多权限、任意权限校验
2. 缓存机制：缓存用户权限，提高性能
3. 日志记录：记录权限校验结果，便于审计
4. 错误处理：清晰的权限错误提示
5. 可扩展：容易集成到现有系统

这个案例可以作为实际项目中权限系统的参考实现。