[Python3高阶编程] - 高阶函数一：常见的高阶函数

一、常见的高阶函数Python 中的高阶函数Higher-Order Function是指接收函数作为参数或返回一个函数的函数。Python 内置了大量高阶函数下面按来源分类尽量列全1. 内置高阶函数函数作用map(func, *iterables)将函数逐个作用于可迭代对象元素返回迭代器filter(func, iterable)用函数筛选可迭代对象中符合条件的元素sorted(iterable, *, key, reverse)排序key接收一个函数决定排序依据max(iterable, *, key)/min(iterable, *, key)找最大/最小值key指定比较依据zip(*iterables)虽不直接接收函数但常与map搭配使用2.functools模块函数/装饰器作用reduce(func, iterable[, init])累积归约将函数反复作用到序列上得到单一值partial(func, *args, **kwargs)固定部分参数生成新函数lru_cache(maxsize)缓存装饰器自动记忆函数返回值cachelru_cache(maxsizeNone)的简洁写法Python 3.9wraps(wrapped)装饰器辅助工具保留原函数元信息cmp_to_key(func)将旧式两两比较函数转为key函数singledispatch单分派泛函数装饰器按第一个参数类型选择实现total_ordering类装饰器补全比较方法cached_property将方法转为只计算一次的属性Python 3.83.itertools模块函数式工具库函数作用filterfalse(pred, iterable)与filter相反——保留pred返回False的元素starmap(func, iterable)map的变体将元素解包为位置参数传给函数takewhile(pred, iterable)遇到第一个不满足pred的元素时停止dropwhile(pred, iterable)跳过开头不满足pred的元素之后全保留compress(data, selectors)按布尔选择器筛选数据4.operator模块替代 lambda 的标准操作符函数函数说明add / sub / mul / truediv / floordiv / mod / pow对应 - * / // % **eq / ne / lt / le / gt / ge对应 ! and_ / or_ / xor / not_ / invert逻辑和位运算itemgetter(*items)按索引/键取值常用于sorted的keyattrgetter(*attrs)按属性取值methodcaller(name, *args)调用对象的方法5. 装饰器——最典型的高阶函数装饰器本质上就是func → func的变换函数常见内置和标准库装饰器装饰器作用staticmethod静态方法classmethod类方法property属性访问器abstractmethod抽象方法dataclass数据类Python 3.7contextmanager用生成器构造上下文管理器atexit.register程序退出时执行timer/retry/cache自定义装饰器下面会讲二、高阶函数详解场景 用法1.map()— 一对一变换# 基础 nums [1, 2, 3, 4] squares list(map(lambda x: x ** 2, nums)) # [1, 4, 9, 16] # 多序列并行 a [1, 2, 3] b [10, 20, 30] list(map(lambda x, y: x y, a, b)) # [11, 22, 33] # 场景类型批量转换 str_nums [1, 2, 3] list(map(int, str_nums)) # [1, 2, 3] # 场景字符串清洗 raw [ hello , world , foo ] list(map(str.strip, raw)) # [hello, world, foo]2.filter()— 条件筛选# 基础 nums range(1, 11) list(filter(lambda x: x % 3 0, nums)) # [3, 6, 9] # 场景过滤 None / 空值 data [0, , None, hello, [], [1, 2], 42] list(filter(None, data)) # [hello, [1, 2], 42] # 场景按条件过滤对象 class User: def __init__(self, name, age): self.name name self.age age users [User(Alice, 25), User(Bob, 17), User(Carol, 30)] adults list(filter(lambda u: u.age 18, users))3.reduce()— 累积归约from functools import reduce # 求积 reduce(lambda a, b: a * b, [1, 2, 3, 4, 5]) # 120 # 找最长字符串 words [a, abc, ab, abcdef] reduce(lambda a, b: a if len(a) len(b) else b, words) # abcdef # 场景扁平化嵌套列表 nested [[1, 2], [3, 4], [5]] reduce(lambda a, b: a b, nested) # [1, 2, 3, 4, 5] # 场景构建深层字典访问 data {a: {b: {c: 42}}} reduce(dict.get, [a, b, c], data) # 42 # 场景URL 路径拼接 parts [http://example.com, api, v1, users] reduce(lambda a, b: f{a}/{b}, parts) # http://example.com/api/v1/users4.sorted()key— 自定义排序# 按字符串长度 words [banana, pie, Washington] sorted(words, keylen) # [pie, banana, Washington] # 按字典值 students [ {name: Alice, score: 85}, {name: Bob, score: 92}, {name: Carol, score: 78}, ] sorted(students, keylambda s: s[score], reverseTrue) # 按对象属性 class Person: def __init__(self, name, age): self.name, self.age name, age people [Person(Alice, 30), Person(Bob, 25), Person(Carol, 30)] # 用 itemgetter from operator import itemgetter sorted(people, keyattrgetter(age, name)) # 多级排序先按 age 升序再按 name 降序 sorted(people, keylambda p: (p.age, [-ord(c) for c in p.name]))5.max()/min()keywords [apple, strawberry, fig, banana] max(words, keylen) # strawberry min(words, keylen) # fig # 场景找到成绩最高的学生 max(students, keylambda s: s[score]) # {name: Bob, score: 92}6.partial()— 参数预设from functools import partial # 基础预设 base 的二进制转换 int2 partial(int, base2) int2(1101) # 13 # 场景日志快捷函数 import logging log_warning partial(logging.log, logging.WARNING) log_error partial(logging.log, logging.ERROR) log_warning(Disk usage above 80%) log_error(Connection timeout) # 场景HTTP 请求预设配置 import requests get partial(requests.get, timeout10, verifyTrue) post partial(requests.post, timeout10, verifyTrue) # 场景GUI 回调预设参数 from tkinter import Button def on_click(button_id): print(fButton {button_id} clicked) # 用 partial 绑定参数 btn Button(textOK, commandpartial(on_click, 42))7.lru_cache/cache— 自动记忆化from functools import lru_cache, cache # 基础斐波那契无缓存 O(2^n)有缓存 O(n) lru_cache(maxsize128) def fib(n): if n 2: return n return fib(n-1) fib(n-2) fib(100) # 瞬间返回 # 场景昂贵的 API 调用缓存 import requests lru_cache(maxsize64) def get_user_info(user_id): return requests.get(fhttps://api.example.com/users/{user_id}).json() # 同一 user_id 只请求一次 get_user_info(1) # 网络请求 get_user_info(1) # 缓存命中 # Python 3.9 简洁写法 cache def expensive_computation(x): # ... 复杂计算 return result # 查看缓存命中率 fib.cache_info() # CacheInfo(hits98, misses101, maxsize128, currsize101)8.singledispatch— 函数重载按类型分派from functools import singledispatch singledispatch def process(data): raise TypeError(fUnsupported type: {type(data)}) process.register def _(data: str): return fProcessing string: {data.upper()} process.register def _(data: int): return fProcessing int: {data * 2} process.register def _(data: list): return fProcessing list with {len(data)} items process(hello) # Processing string: HELLO process(42) # Processing int: 84 process([1, 2, 3]) # Processing list with 3 items9. 装饰器 — 函数增强from functools import wraps import time # --- 计时装饰器 --- def timer(func): wraps(func) def wrapper(*args, **kwargs): start time.perf_counter() result func(*args, **kwargs) elapsed time.perf_counter() - start print(f{func.__name__} took {elapsed:.4f}s) return result return wrapper # --- 重试装饰器 --- def retry(max_attempts3, delay1): import time as _time def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_attempts): try: return func(*args, **kwargs) except Exception as e: if attempt max_attempts - 1: raise print(fAttempt {attempt1} failed: {e}) _time.sleep(delay) return wrapper return decorator # --- 权限检查装饰器 --- def require_role(role): def decorator(func): wraps(func) def wrapper(user, *args, **kwargs): if user.role ! role: raise PermissionError(fRequires {role} role) return func(user, *args, **kwargs) return wrapper return decorator # 使用 timer retry(max_attempts3, delay2) def fetch_data(url): return requests.get(url).json() require_role(admin) def delete_user(admin, user_id): # 只有 admin 能执行 pass10.itertools中的高阶函数from itertools import filterfalse, starmap, takewhile, dropwhile # filterfalse — 保留不满足条件的 list(filterfalse(lambda x: x % 2, range(10))) # [0, 2, 4, 6, 8] ← 全是偶数不满足 % 2 为 True # starmap — 解包传参 from operator import pow list(starmap(pow, [(2, 3), (3, 2), (10, 3)])) # [8, 9, 1000] ← 等价于 2**3, 3**2, 10**3 # takewhile — 取到不满足为止 list(takewhile(lambda x: x 5, [1, 3, 4, 6, 2, 8])) # [1, 3, 4] # dropwhile — 跳过开头不满足的 list(dropwhile(lambda x: x 5, [1, 3, 4, 6, 2, 8])) # [6, 2, 8]11.operator模块 — 替代 lambdafrom operator import add, mul, itemgetter, attrgetter, methodcaller # 替代 lambda 做 reduce reduce(add, [1, 2, 3, 4, 5]) # 15替代 lambda a, b: a b reduce(mul, [1, 2, 3, 4, 5]) # 120 # itemgetter — 字典/元组取值 students [(Alice, 85), (Bob, 92), (Carol, 78)] sorted(students, keyitemgetter(1)) # [(Carol, 78), (Alice, 85), (Bob, 92)] # 多键取值 get_name_and_score itemgetter(name, score) get_name_and_score({name: Alice, score: 85}) # (Alice, 85) # attrgetter — 对象属性取值 class Point: def __init__(self, x, y): self.x, self.y x, y points [Point(3, 1), Point(1, 4), Point(2, 2)] sorted(points, keyattrgetter(x)) # methodcaller — 调用方法 names [alice, BOB, Carol] sorted(names, keymethodcaller(lower))12. 返回函数 — 工厂 闭包# 工厂函数 def make_power(n): def power(x): return x ** n return power square make_power(2) cube make_power(3) square(5) # 25 cube(5) # 125 # 场景构建中间件链Flask 风格 def middleware(handler): def wrapper(request): request[logged] True return handler(request) return wrapper def build_pipeline(*middlewares): def final_handler(request): return {response: OK, **request} for mw in reversed(middlewares): final_handler mw(final_handler) return final_handler # 场景参数化查询构建 def build_query(table): def where(**conditions): clauses AND .join(f{k} {v} for k, v in conditions.items()) return fSELECT * FROM {table} WHERE {clauses} return where query_user build_query(users) query_user(nameAlice, age25) # SELECT * FROM users WHERE name Alice AND age 2513.contextmanager— 生成器变上下from contextlib import contextmanager contextmanager def timer_context(label): start time.perf_counter() try: yield # 此处进入 with 体 finally: elapsed time.perf_counter() - start print(f{label} took {elapsed:.4f}s) with timer_context(Database query): time.sleep(1) # Database query took 1.0012s # 场景临时修改工作目录 import os contextmanager def cd(path): old os.getcwd() os.chdir(path) try: yield finally: os.chdir(old) with cd(/tmp): print(os.getcwd()) # /tmp print(os.getcwd()) # 回到原目录 # 场景临时修改环境变量 contextmanager def env_vars(**kwargs): old {k: os.environ.get(k) for k in kwargs} os.environ.update(kwargs) try: yield finally: for k, v in old.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] v14.cmp_to_key— 旧式比较函数适配from functools import cmp_to_key # 旧式比较返回 -1/0/1 def compare_version(v1, v2): a, b map(int, v1.split(.)), map(int, v2.split(.)) for x, y in zip(a, b): if x ! y: return -1 if x y else 1 return 0 versions [1.2, 1.10, 1.1, 2.0] sorted(versions, keycmp_to_key(compare_version)) # [1.1, 1.2, 1.10, 2.0]