
Python装饰器是函数式编程的重要特性,它允许在不修改原函数代码的情况下,动态地扩展函数功能。以下是10种高级用法,每种都包含说明、使用场景和代码示例。
往期阅读>>>
Python 20 个文本分析的库:效率提升 10 倍的秘密武器
Python 自动化管理Jenkins的15个实用脚本,提升效率
App2Docker:如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率
说明:装饰器本身可以接受参数,通过外层函数传递参数实现更灵活的装饰逻辑。
使用场景:需要根据配置动态调整装饰行为的场景,如不同环境的日志级别、缓存时间等。
defretry(max_attempts=3):defdecorator(func):defwrapper(*args, **kwargs):forattemptinrange(max_attempts):try:returnfunc(*args, **kwargs)exceptExceptionase:ifattempt == max_attempts-1:raiseeprint(f"Attempt {attempt + 1} failed, retrying...")returnNonereturnwrapperreturndecorator@retry(max_attempts=5)deffetch_data(url):# 模拟网络请求importrandomifrandom.random() <0.7:raiseConnectionError("Connection failed")return"Data fetched successfully"print(fetch_data("https://api.example.com"))
说明:使用类定义装饰器逻辑,通过__init__和__call__方法实现。
使用场景:需要维护状态的装饰器,如计数器、缓存管理等。
classCountCalls:def__init__(self, func):self.func = funcself.call_count = 0def__call__(self, *args, **kwargs):self.call_count += 1print(f"{self.func.__name__} has been called {self.call_count} times")returnself.func(*args, **kwargs)@CountCallsdefprocess_data(data):returndata.upper()process_data("hello")process_data("world")
说明:多个装饰器按从外到内顺序应用于同一函数。
使用场景:需要组合多个功能的场景,如日志+缓存+权限验证。
deflog_execution(func):defwrapper(*args, **kwargs):print(f"Executing {func.__name__} with args: {args}, kwargs: {kwargs}")result = func(*args, **kwargs)print(f"{func.__name__} returned: {result}")returnresultreturnwrapperdefvalidate_input(func):defwrapper(*args, **kwargs):forarginargs:ifnotisinstance(arg, (int, float)):raiseValueError("Arguments must be numbers")returnfunc(*args, **kwargs)returnwrapper@log_execution@validate_inputdefadd_numbers(a, b):returna+bprint(add_numbers(5, 3))
说明:装饰器内部使用其他装饰器,实现功能组合。
使用场景:创建可复用的装饰器组合。
deftimer_decorator(func):importtimedefwrapper(*args, **kwargs):start = time.time()result = func(*args, **kwargs)end = time.time()print(f"{func.__name__} took {end - start:.4f} seconds")returnresultreturnwrapperdeflog_decorator(func):defwrapper(*args, **kwargs):print(f"Calling {func.__name__}")returnfunc(*args, **kwargs)returnwrapperdefcombined_decorator(func):# 组合timer和log装饰器returntimer_decorator(log_decorator(func))@combined_decoratordefheavy_computation(n):importtimetime.sleep(0.5)returnsum(range(n))print(heavy_computation(1000))
说明:专门用于装饰异步函数的装饰器。
使用场景:异步编程中的性能监控、错误重试等。
importasynciodefasync_timer(func):asyncdefwrapper(*args, **kwargs):importtimestart = time.time()result = awaitfunc(*args, **kwargs)end = time.time()print(f"Async {func.__name__} took {end - start:.4f} seconds")returnresultreturnwrapper@async_timerasyncdeffetch_urls(urls):importaiohttpasyncwithaiohttp.ClientSession() assession:tasks = [session.get(url) forurlinurls]responses = awaitasyncio.gather(*tasks)return [awaitr.text() forrinresponses]# 使用示例# asyncio.run(fetch_urls(["http://example.com"]))
说明:在类方法上使用装饰器访问类属性。
使用场景:类方法的状态管理、访问控制等。
classRateLimiter:def__init__(self, max_calls=3):self.max_calls = max_callsself.call_count = 0deflimit_calls(self, func):defwrapper(*args, **kwargs):ifself.call_count>= self.max_calls:raiseException(f"Rate limit exceeded. Max calls: {self.max_calls}")self.call_count += 1returnfunc(*args, **kwargs)returnwrapperlimiter = RateLimiter(max_calls=2)classAPI:@limiter.limit_callsdefcall_endpoint(self, endpoint):returnf"Calling {endpoint}"api = API()print(api.call_endpoint("/users"))print(api.call_endpoint("/posts"))# 第三次调用会触发异常
说明:确保类只有一个实例的装饰器。
使用场景:数据库连接、配置管理、日志记录器等需要单例的场景。
defsingleton(cls):instances = {}defwrapper(*args, **kwargs):ifclsnotininstances:instances[cls] = cls(*args, **kwargs)returninstances[cls]returnwrapper@singletonclassDatabaseConnection:def__init__(self):print("Creating new database connection...")self.connection_id = id(self)defquery(self, sql):returnf"Executing: {sql}"# 测试db1 = DatabaseConnection()db2 = DatabaseConnection()print(f"db1 id: {db1.connection_id}")print(f"db2 id: {db2.connection_id}")print(f"Same instance: {db1 is db2}")
说明:在运行时检查函数参数和返回值的类型。
使用场景:调试、API接口验证、类型安全要求高的场景。
deftype_check(*arg_types, return_type=None):defdecorator(func):defwrapper(*args, **kwargs):# 检查参数类型fori, (arg, expected_type) inenumerate(zip(args, arg_types)):ifnotisinstance(arg, expected_type):raiseTypeError(f"Argument {i} must be {expected_type}, got {type(arg)}")# 执行函数result = func(*args, **kwargs)# 检查返回值类型ifreturn_typeandnotisinstance(result, return_type):raiseTypeError(f"Return value must be {return_type}, got {type(result)}")returnresultreturnwrapperreturndecorator@type_check(int, int, return_type=int)defmultiply(a, b):returna*bprint(multiply(5, 3)) # 正常# print(multiply(5, "3")) # 会抛出TypeError
说明:缓存函数结果,避免重复计算。
使用场景:计算成本高、结果不变的函数,如API调用、复杂计算等。
fromfunctoolsimportlru_cacheimporttime# 使用functools.lru_cache@lru_cache(maxsize=128)deffibonacci(n):ifn<2:returnnreturnfibonacci(n-1) +fibonacci(n-2)# 自定义缓存装饰器defcache_results(func):cache = {}defwrapper(*args, **kwargs):key = (args, tuple(kwargs.items()))ifkeynotincache:cache[key] = func(*args, **kwargs)returncache[key]returnwrapper@cache_resultsdefexpensive_computation(x):time.sleep(1) # 模拟耗时计算returnx*x# 第一次调用耗时start = time.time()print(expensive_computation(5))print(f"First call: {time.time() - start:.2f}s")# 第二次调用从缓存读取start = time.time()print(expensive_computation(5))print(f"Second call: {time.time() - start:.2f}s")
说明:验证用户权限后再执行函数。
使用场景:Web应用、API接口的权限控制。
defrequire_role(*allowed_roles):defdecorator(func):defwrapper(user, *args, **kwargs):ifuser.get('role') notinallowed_roles:raisePermissionError(f"User {user.get('name')} with role {user.get('role')} "f"is not allowed to call {func.__name__}. "f"Allowed roles: {allowed_roles}" )returnfunc(user, *args, **kwargs)returnwrapperreturndecoratorclassUserService:@require_role('admin', 'manager')defdelete_user(self, user, user_id):returnf"User {user_id} deleted by {user['name']}"@require_role('admin', 'manager', 'user')defview_profile(self, user, user_id):returnf"Viewing profile of user {user_id}"service = UserService()admin_user = {'name': 'Alice', 'role': 'admin'}regular_user = {'name': 'Bob', 'role': 'user'}print(service.delete_user(admin_user, 123)) # 正常print(service.view_profile(regular_user, 123)) # 正常# print(service.delete_user(regular_user, 123)) # 会抛出PermissionError
importtimeimportfunctoolsdefapi_decorator(func):"""API接口的完整装饰器组合"""@functools.wraps(func)defwrapper(*args, **kwargs):# 1. 日志记录print(f"[API] {time.strftime('%Y-%m-%d %H:%M:%S')} - Calling {func.__name__}")# 2. 参数验证ifnotargsandnotkwargs:print("[API] Warning: No arguments provided")# 3. 性能监控start_time = time.time()try:# 执行原函数result = func(*args, **kwargs)# 4. 响应时间记录elapsed = time.time() -start_timeprint(f"[API] {func.__name__} completed in {elapsed:.4f}s")# 5. 结果验证ifresultisNone:print("[API] Warning: Function returned None")returnresultexceptExceptionase:# 6. 错误处理print(f"[API] Error in {func.__name__}: {str(e)}")raisereturnwrapper@api_decoratordefget_user_data(user_id, include_profile=False):"""获取用户数据"""ifnotisinstance(user_id, int):raiseValueError("user_id must be an integer")# 模拟API调用time.sleep(0.1)data = {'id': user_id,'name': f'User{user_id}','email': f'user{user_id}@example.com' }ifinclude_profile:data['profile'] = {'age': 25, 'city': 'Beijing'}returndata# 使用示例try:user_data = get_user_data(123, include_profile=True)print(f"User data: {user_data}")exceptExceptionase:print(f"Failed to get user data: {e}")
Python装饰器的10种高级用法展示了其强大的灵活性和实用性:
带参数装饰器 - 提供配置灵活性
类装饰器 - 支持状态管理
装饰器链 - 实现功能组合
装饰器工厂 - 创建可复用组合
异步装饰器 - 支持现代异步编程
属性装饰器 - 类方法的状态控制
单例装饰器 - 确保唯一实例
类型检查装饰器 - 增强类型安全
缓存装饰器 - 优化性能
权限装饰器 - 实现访问控制
掌握这些高级用法可以帮助您编写更简洁、可维护、高效的Python代码,特别适用于Web开发、API设计、性能优化等场景。在实际应用中,可以根据需求组合使用这些装饰器,创建强大的功能扩展机制。
