Python函数式编程技巧

一、函数是一等公民

在Python中,函数是一等对象,可以赋值给变量、作为参数传递、作为返回值、存储在数据结构中。

def add(x, y):
return x + y

def subtract(x, y):
return x - y

# 函数存储在字典中
operations = {
'+': add,
'-': subtract,
'*': lambda x, y: x * y,
'/': lambda x, y: x / y,
}

def calculate(op, x, y):
return operations[op](x, y)

print(calculate('+', 10, 3)) # 13
print(calculate('*', 4, 5)) # 20


二、高阶函数

高阶函数是接受函数作为参数或返回函数的函数。

2.1 map、filter、reduce

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# map:对每个元素应用函数
squares = list(map(lambda x: x**2, numbers))
# [1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

# filter:筛选满足条件的元素
evens = list(filter(lambda x: x % 2 == 0, numbers))
# [2, 4, 6, 8, 10]

# reduce:累积计算
from functools import reduce
product = reduce(lambda acc, x: acc * x, numbers)
# 3628800

2.2 自定义高阶函数

def retry(max_attempts, exceptions=(Exception,)):
"""返回一个重试装饰器"""
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts - 1:
raise
print(f"尝试 {attempt + 1} 失败: {e}")
return wrapper
return decorator

@retry(max_attempts=3, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
# 可能失败的网络请求
pass


三、闭包

闭包是引用了外部函数变量的内部函数,即使外部函数已经返回。

def make_counter(start=0):
count = start

def counter():
nonlocal count
count += 1
return count

def reset():
nonlocal count
count = start

counter.reset = reset
return counter

c = make_counter(10)
print(c()) # 11
print(c()) # 12
print(c()) # 13
c.reset()
print(c()) # 11

闭包的实际应用:

def make_validator(min_length, max_length, pattern=None):
"""创建字符串验证器"""
import re
compiled_pattern = re.compile(pattern) if pattern else None

def validate(value):
if not isinstance(value, str):
return False, "必须是字符串"
if len(value) < min_length:
return False, f"长度不能少于{min_length}"
if len(value) > max_length:
return False, f"长度不能超过{max_length}"
if compiled_pattern and not compiled_pattern.match(value):
return False, "格式不正确"
return True, "验证通过"

return validate

email_validator = make_validator(5, 100, r'^[\w.+-]+@[\w-]+\.[\w.-]+$')
username_validator = make_validator(3, 20, r'^[a-zA-Z0-9_]+$')

print(email_validator("user@example.com")) # (True, '验证通过')
print(username_validator("ab")) # (False, '长度不能少于3')


四、偏函数 functools.partial

from functools import partial

def power(base, exponent):
return base ** exponent

square = partial(power, exponent=2)
cube = partial(power, exponent=3)

print(square(5)) # 25
print(cube(3)) # 27

# 实际应用:简化回调函数
import json

# 创建特定配置的JSON序列化器
compact_json = partial(json.dumps, separators=(',', ':'))
pretty_json = partial(json.dumps, indent=2, ensure_ascii=False)

data = {"name": "张三", "age": 30}
print(compact_json(data)) # {"name":"张三","age":30}
print(pretty_json(data)) # 格式化输出


五、函数组合

def compose(*functions):
"""从右到左组合函数"""
def composed(x):
result = x
for func in reversed(functions):
result = func(result)
return result
return composed

def pipe(*functions):
"""从左到右组合函数(管道)"""
def piped(x):
result = x
for func in functions:
result = func(result)
return result
return piped

# 数据处理管道
clean_text = pipe(
str.strip,
str.lower,
lambda s: s.replace('\n', ' '),
lambda s: ' '.join(s.split()), # 合并多余空格
)

text = " Hello World \n Python "
print(clean_text(text)) # "hello world python"

# 数值转换管道
process_number = pipe(
abs,
lambda x: round(x, 2),
str,
lambda s: f"${s}",
)

print(process_number(-123.456)) # "$123.46"


六、不可变数据与纯函数

纯函数:相同输入总是产生相同输出,没有副作用。

# 非纯函数(修改外部状态)
total = 0
def add_to_total(x):
global total
total += x
return total

# 纯函数(无副作用)
def add_numbers(numbers):
return sum(numbers)

# 使用不可变数据结构
from typing import NamedTuple

class Point(NamedTuple):
x: float
y: float

def move_point(point, dx, dy):
"""返回新的Point,不修改原对象"""
return Point(point.x + dx, point.y + dy)

p1 = Point(1, 2)
p2 = move_point(p1, 3, 4)
print(p1) # Point(x=1, y=2) 未被修改
print(p2) # Point(x=4, y=6)


七、递归与尾递归优化

def flatten_deep(nested):
"""递归展平嵌套结构"""
result = []
for item in nested:
if isinstance(item, (list, tuple)):
result.extend(flatten_deep(item))
else:
result.append(item)
return result

# Python不支持尾递归优化,但可以用迭代模拟
def flatten_iterative(nested):
"""迭代方式展平"""
stack = list(reversed(nested))
result = []
while stack:
item = stack.pop()
if isinstance(item, (list, tuple)):
stack.extend(reversed(item))
else:
result.append(item)
return result

# 使用 lru_cache 优化递归
from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(100)) # 瞬间计算出结果


八、函数式工具集

8.1 operator模块

import operator
from functools import reduce

# 替代lambda的运算符函数
numbers = [3, 1, 4, 1, 5, 9, 2, 6]

# 排序
sorted_nums = sorted(numbers, key=operator.neg) # 降序

# 属性获取
class Student:
def __init__(self, name, grade):
self.name = name
self.grade = grade

students = [Student("Alice", 85), Student("Bob", 92), Student("Charlie", 78)]
top_student = max(students, key=operator.attrgetter('grade'))

# 字典项获取
records = [{'name': 'A', 'score': 90}, {'name': 'B', 'score': 85}]
sorted_records = sorted(records, key=operator.itemgetter('score'))

8.2 itertools函数式工具

import itertools

# takewhile / dropwhile
numbers = [2, 4, 6, 7, 8, 10]
before_odd = list(itertools.takewhile(lambda x: x % 2 == 0, numbers))
# [2, 4, 6]

# starmap:展开参数
pairs = [(1, 2), (3, 4), (5, 6)]
sums = list(itertools.starmap(operator.add, pairs))
# [3, 7, 11]

# groupby
data = sorted(['apple', 'avocado', 'banana', 'blueberry', 'cherry'],
key=lambda x: x[0])
for letter, group in itertools.groupby(data, key=lambda x: x[0]):
print(f"{letter}: {list(group)}")


九、实际应用:数据转换管道

from functools import reduce, partial
from operator import itemgetter

# 模拟数据
orders = [
{'product': 'laptop', 'price': 999, 'quantity': 2, 'category': 'electronics'},
{'product': 'book', 'price': 15, 'quantity': 5, 'category': 'education'},
{'product': 'phone', 'price': 699, 'quantity': 1, 'category': 'electronics'},
{'product': 'pen', 'price': 3, 'quantity': 100, 'category': 'office'},
{'product': 'tablet', 'price': 499, 'quantity': 3, 'category': 'electronics'},
]

# 函数式数据处理
def filter_by(key, value):
return lambda items: [item for item in items if item[key] == value]

def add_field(name, func):
return lambda items: [{**item, name: func(item)} for item in items]

def sort_by(key, reverse=False):
return lambda items: sorted(items, key=itemgetter(key), reverse=reverse)

def take(n):
return lambda items: items[:n]

# 组合管道
top_electronics = pipe(
filter_by('category', 'electronics'),
add_field('total', lambda o: o['price'] * o['quantity']),
sort_by('total', reverse=True),
take(2),
)

result = top_electronics(orders)
for item in result:
print(f"{item['product']}: ${item['total']}")
# laptop: $1998
# tablet: $1497


十、注意事项

1. Python不是纯函数式语言,不必强制所有代码都函数式
2. 列表推导式通常比map/filter更Pythonic
3. 过度使用lambda会降低可读性,复杂逻辑应定义具名函数
4. 递归深度有限(默认1000),深递归应改用迭代
5. 函数式风格适合数据转换管道,命令式风格适合有副作用的操作

总结:函数式编程是Python工具箱中的重要组成部分。高阶函数、闭包、函数组合等技巧能让代码更简洁、更易测试。关键是在合适的场景选择合适的范式,而非教条地追求纯函数式。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐