1. Python函数基础:从零开始理解代码封装
在Python编程中,函数就像厨房里的多功能料理机——你把食材(参数)放进去,选择功能(逻辑处理),就能得到加工好的食物(返回值)。这种封装思想让代码摆脱了"一锅炖"的混乱状态,成为构建复杂程序的基石。
初学者常犯的错误是写出几十行连续执行的"面条代码",而职业开发者会把功能拆解成多个函数单元。比如处理用户登录时,我们通常会分离出:验证输入格式的函数、查询数据库的函数、生成token的函数。这种模块化设计让代码具备以下优势:
- 单次定义多次调用,避免重复代码
- 每个函数专注单一功能,调试更简单
- 团队协作时接口清晰,分工明确
- 修改内部实现不影响其他代码
来看一个典型场景:电商网站的价格计算。没有函数时,每次计算都要重复写折扣逻辑、税费计算和运费规则。而使用函数后,只需调用calculate_total(price, coupon)即可。这就是为什么函数被称为"可重用代码的基石"。
关键经验:当同一段代码出现第三次时,就该考虑封装成函数了。这个"三次法则"能有效提升代码质量。
2. 函数定义深度解析:参数设计的艺术
2.1 基础定义语法
Python函数定义使用def关键字,其完整结构如下:
def function_name(parameters): """docstring""" function_body [return expression]一个计算BMI的典型示例:
def calculate_bmi(weight_kg, height_m): """ 计算身体质量指数(BMI) :param weight_kg: 体重(千克) :param height_m: 身高(米) :return: BMI值 """ if height_m <= 0: raise ValueError("身高必须大于0") return weight_kg / (height_m ** 2)2.2 参数传递的四种方式
位置参数:最常见的传参方式,按定义顺序传递
def power(base, exponent): return base ** exponent print(power(2, 3)) # 8关键字参数:明确指定参数名,顺序可变
print(power(exponent=3, base=2)) # 8默认参数:定义时指定默认值
def greet(name, message="Hello"): print(f"{message}, {name}!") greet("Alice") # Hello, Alice! greet("Bob", "Hi") # Hi, Bob!可变参数:处理不确定数量的参数
*args接收任意数量的位置参数(元组)**kwargs接收任意数量的关键字参数(字典)
def print_params(*args, **kwargs): print("位置参数:", args) print("关键字参数:", kwargs) print_params(1, 2, a=3, b=4) # 位置参数: (1, 2) # 关键字参数: {'a': 3, 'b': 4}
2.3 参数设计的最佳实践
参数顺序原则:
- 无默认值的参数在前,有默认值的在后
- 一般参数在前,可变参数在后
- 遵循:位置参数 -> *args -> 关键字参数 -> **kwargs
默认参数陷阱: 默认值只会在函数定义时计算一次,对于可变对象(如列表、字典)会导致意外共享:
# 错误示例 def add_item(item, items=[]): items.append(item) return items print(add_item(1)) # [1] print(add_item(2)) # [1, 2] 不是预期的[2] # 正确写法 def add_item(item, items=None): if items is None: items = [] items.append(item) return items类型提示增强可读性(Python 3.5+):
from typing import List, Optional def process_items(items: List[str], limit: Optional[int] = None) -> int: """ :param items: 字符串列表 :param limit: 可选的数量限制 :return: 处理后的项目数量 """ if limit is not None: items = items[:limit] return len(items)
3. 函数调用机制与作用域
3.1 调用栈与执行流程
当Python调用函数时,会创建一个新的栈帧(stack frame)用于存储:
- 局部变量
- 函数参数
- 返回地址
- 其他上下文信息
这个栈帧被压入调用栈,函数执行完毕后再弹出。理解这个过程对调试递归函数尤为重要。
3.2 变量作用域解析
Python使用LEGB规则查找变量:
- Local(局部作用域)
- Enclosing(闭包函数外的函数)
- Global(模块全局)
- Built-in(内置名称)
典型作用域示例:
x = "global" def outer(): x = "outer" def inner(): x = "inner" print(x) # inner inner() print(x) # outer outer() print(x) # global3.3 global与nonlocal关键字
global:声明使用全局变量count = 0 def increment(): global count count += 1nonlocal:修改嵌套作用域中的变量def counter(): n = 0 def inc(): nonlocal n n += 1 return n return inc
常见错误:在未声明nonlocal的情况下尝试修改外部作用域变量,会导致创建新的局部变量而非修改外部变量。
4. 高阶函数与函数式编程
4.1 函数作为一等公民
Python中函数可以:
- 被赋值给变量
- 作为参数传递
- 作为返回值
- 存储在数据结构中
def square(x): return x * x # 赋值给变量 func = square print(func(5)) # 25 # 作为参数 def apply_func(f, numbers): return [f(n) for n in numbers] print(apply_func(square, [1, 2, 3])) # [1, 4, 9]4.2 常用高阶函数
map():对可迭代对象应用函数nums = [1, 2, 3] squared = map(lambda x: x**2, nums) print(list(squared)) # [1, 4, 9]filter():过滤元素evens = filter(lambda x: x % 2 == 0, range(10)) print(list(evens)) # [0, 2, 4, 6, 8]reduce():累积计算(需从functools导入)from functools import reduce product = reduce(lambda x, y: x * y, [1, 2, 3, 4]) print(product) # 24
4.3 闭包与装饰器
闭包:引用了外部变量的内部函数
def make_multiplier(factor): def multiplier(x): return x * factor return multiplier double = make_multiplier(2) print(double(5)) # 10装饰器:修改或增强函数行为的语法糖
def log_time(func): import time def wrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) print(f"{func.__name__}执行耗时: {time.time()-start:.4f}s") return result return wrapper @log_time def heavy_computation(): # 模拟耗时计算 sum(i*i for i in range(10**6)) heavy_computation()5. 错误处理与调试技巧
5.1 异常处理机制
完整的try-except结构:
try: # 可能出错的代码 result = 10 / 0 except ZeroDivisionError as e: # 处理特定异常 print(f"除零错误: {e}") except (TypeError, ValueError) as e: # 处理多种异常 print(f"类型或值错误: {e}") except Exception as e: # 捕获所有异常 print(f"未知错误: {e}") else: # 无异常时执行 print("计算成功") finally: # 无论是否异常都执行 print("清理资源")5.2 自定义异常
创建业务相关的异常类型:
class InventoryError(Exception): """库存不足异常""" def __init__(self, item, available): self.item = item self.available = available super().__init__(f"{item}库存不足,当前仅剩{available}件") def purchase(item, quantity): stock = check_stock(item) if stock < quantity: raise InventoryError(item, stock) # 处理购买逻辑5.3 调试函数技巧
print调试法(简单直接):
def complex_calculation(a, b): print(f"输入参数: a={a}, b={b}") # 调试点1 intermediate = a * b print(f"中间结果: {intermediate}") # 调试点2 result = intermediate ** 0.5 print(f"最终结果: {result}") # 调试点3 return result使用pdb调试器:
import pdb def buggy_function(x): pdb.set_trace() # 设置断点 y = x + 1 z = y / 0 # 故意制造错误 return z日志记录(生产环境推荐):
import logging logging.basicConfig( level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) def process_data(data): logging.info("开始处理数据") try: result = complex_operation(data) logging.debug(f"处理结果: {result}") return result except Exception as e: logging.error(f"处理失败: {e}", exc_info=True) raise
6. 性能优化与高级技巧
6.1 函数缓存优化
使用functools.lru_cache缓存计算结果:
from functools import lru_cache @lru_cache(maxsize=128) def fibonacci(n): if n < 2: return n return fibonacci(n-1) + fibonacci(n-2) # 第一次计算会递归调用 print(fibonacci(50)) # 12586269025 # 第二次直接从缓存读取 print(fibonacci(50)) # 立即返回6.2 生成器函数
使用yield创建内存友好的迭代器:
def read_large_file(file_path): """逐行读取大文件,避免内存溢出""" with open(file_path, 'r') as f: for line in f: yield line.strip() # 使用示例 for line in read_large_file('huge_data.txt'): process_line(line)6.3 函数签名保留
使用functools.wraps保留原函数元信息:
from functools import wraps def my_decorator(f): @wraps(f) def wrapper(*args, **kwargs): print(f"调用函数: {f.__name__}") return f(*args, **kwargs) return wrapper @my_decorator def example(): """示例函数""" pass print(example.__name__) # 输出'example'而非'wrapper' print(example.__doc__) # 输出'示例函数'6.4 异步函数
使用async/await编写异步代码:
import asyncio async def fetch_data(url): print(f"开始获取 {url}") await asyncio.sleep(2) # 模拟IO操作 print(f"完成获取 {url}") return f"{url}的数据" async def main(): tasks = [ fetch_data("https://api1.example.com"), fetch_data("https://api2.example.com") ] results = await asyncio.gather(*tasks) print(results) asyncio.run(main())7. 实战案例:构建可重用工具函数集
7.1 文件处理工具集
import os import hashlib from pathlib import Path from typing import Union def get_file_size(filepath: Union[str, Path]) -> int: """获取文件大小(字节)""" return os.path.getsize(filepath) def calculate_md5(filepath: Union[str, Path], chunk_size=8192) -> str: """计算文件MD5哈希值""" md5 = hashlib.md5() with open(filepath, 'rb') as f: while chunk := f.read(chunk_size): md5.update(chunk) return md5.hexdigest() def backup_file(source: Union[str, Path], backup_dir: Union[str, Path]) -> Path: """备份文件到指定目录,附加时间戳""" source = Path(source) if not source.exists(): raise FileNotFoundError(f"源文件不存在: {source}") backup_dir = Path(backup_dir) backup_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = backup_dir / f"{source.stem}_{timestamp}{source.suffix}" import shutil shutil.copy2(source, backup_path) return backup_path7.2 数据处理工具集
import pandas as pd import numpy as np from typing import List, Dict, Any def clean_dataframe(df: pd.DataFrame, drop_na_threshold: float = 0.7, numeric_fill: str = 'median', categorical_fill: str = 'mode') -> pd.DataFrame: """ 自动化清理DataFrame: 1. 删除缺失值过多的列 2. 填充数值型缺失值 3. 填充类别型缺失值 """ # 删除缺失值超过阈值的列 na_ratio = df.isna().mean() cols_to_drop = na_ratio[na_ratio > drop_na_threshold].index df = df.drop(columns=cols_to_drop) # 填充数值型列 num_cols = df.select_dtypes(include=np.number).columns if numeric_fill == 'median': df[num_cols] = df[num_cols].fillna(df[num_cols].median()) elif numeric_fill == 'mean': df[num_cols] = df[num_cols].fillna(df[num_cols].mean()) # 填充类别型列 cat_cols = df.select_dtypes(exclude=np.number).columns if categorical_fill == 'mode': for col in cat_cols: df[col] = df[col].fillna(df[col].mode()[0]) return df def dicts_to_dataframe(data: List[Dict[str, Any]], columns: List[str] = None) -> pd.DataFrame: """ 将字典列表转换为DataFrame,确保列顺序一致 """ if not data: return pd.DataFrame(columns=columns or []) if columns is None: columns = list(data[0].keys()) return pd.DataFrame([ {col: item.get(col) for col in columns} for item in data ], columns=columns)7.3 Web开发工具集
import requests from urllib.parse import urljoin from typing import Optional, Dict, Any def make_api_request( base_url: str, endpoint: str, method: str = "GET", params: Optional[Dict[str, Any]] = None, data: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, timeout: int = 10 ) -> Dict[str, Any]: """ 通用API请求函数,包含错误处理和重试机制 """ url = urljoin(base_url, endpoint) session = requests.Session() for attempt in range(3): # 最多重试3次 try: response = session.request( method=method, url=url, params=params, json=data, headers=headers, timeout=timeout ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: if attempt == 2: # 最后一次尝试仍然失败 raise Exception(f"API请求失败: {str(e)}") from e time.sleep(1 * (attempt + 1)) # 指数退避 def render_template(template_path: str, context: Dict[str, Any]) -> str: """ 简易模板渲染函数,支持变量替换和简单控制结构 """ with open(template_path, 'r') as f: content = f.read() # 变量替换 for key, value in context.items(): content = content.replace(f"{{{{ {key} }}}}", str(value)) # 简单if条件 lines = [] skip = False for line in content.splitlines(): if line.strip().startswith("{% if"): expr = line.strip()[6:-3].strip() skip = not eval(expr, {}, context) continue if line.strip() == "{% endif %}": skip = False continue if not skip: lines.append(line) return '\n'.join(lines)8. 函数设计原则与最佳实践
8.1 SOLID原则在函数设计中的应用
单一职责原则(SRP): 每个函数应该只做一件事,且做好这件事。如果一个函数的注释需要用到"和"、"或"等连接词,就可能违反了SRP。
# 违反SRP def process_user_data_and_send_email(user): # 处理数据 cleaned_data = clean_data(user.data) # 发送邮件 send_email(user.email, "数据处理完成", str(cleaned_data)) # 符合SRP def process_user_data(user): return clean_data(user.data) def notify_user(user, message): send_email(user.email, "通知", message)开闭原则(OCP): 函数应该对扩展开放,对修改关闭。通过参数化和高阶函数实现。
# 硬编码的比较方式 def sort_users(users): return sorted(users, key=lambda u: u.last_name) # 可扩展的比较方式 def sort_users(users, key_func=None): key_func = key_func or (lambda u: u.last_name) return sorted(users, key=key_func)里氏替换原则(LSP): 子类中的函数应该能够替换父类中的函数而不破坏程序。在Python中主要通过鸭子类型实现。
接口隔离原则(ISP): 不要强迫客户端依赖它们不用的方法。在函数层面体现为参数设计要精确。
# 违反ISP def save_document(doc, format='pdf', encrypt=False, watermark=None): # 所有调用者都必须处理所有参数 pass # 符合ISP def save_as_pdf(doc): pass def add_watermark(doc, watermark): pass def encrypt_document(doc): pass依赖倒置原则(DIP): 高层函数不应该依赖低层函数,二者都应该依赖抽象。通过依赖注入实现。
# 直接依赖具体实现 def process_data(data): db = MySQLDatabase() db.save(data) # 依赖抽象 def process_data(data, db): db.save(data)
8.2 函数命名的艺术
好的函数名应该:
- 明确表达函数意图
- 使用动词开头
- 避免模糊的词语如"handle", "process"
- 保持一致的命名风格
命名示例:
# 好名字 def calculate_tax(income): def find_user_by_email(email): def normalize_string(text): # 不好的名字 def tax(income): # 缺少动词 def do_stuff(data): # 太模糊 def perform_operation(x, y): # 不清楚具体操作8.3 文档字符串标准
遵循PEP 257规范,推荐使用Google风格:
def calculate_distance(point1, point2): """计算两点之间的欧几里得距离。 Args: point1 (tuple): 第一个点的(x, y)坐标 point2 (tuple): 第二个点的(x, y)坐标 Returns: float: 两点之间的距离 Raises: ValueError: 如果坐标维度不一致 Examples: >>> calculate_distance((0, 0), (3, 4)) 5.0 """ if len(point1) != len(point2): raise ValueError("坐标维度不一致") return sum((a - b) ** 2 for a, b in zip(point1, point2)) ** 0.58.4 性能考量
避免不必要的函数调用: 在循环内部调用函数会有额外开销,可以的话将循环移到函数内部。
# 低效 for item in large_list: process_item(item) # 更高效 def process_all(items): for item in items: # 处理逻辑局部变量查找更快: 频繁访问的全局变量可以在函数开始赋值给局部变量。
def calculate(values): # 将全局函数赋值给局部变量 sum_func = sum sqrt = math.sqrt return [sqrt(sum_func(v)) for v in values]使用生成器表达式代替列表推导: 当不需要立即计算所有结果时。
# 立即计算所有结果 total = sum([x*x for x in range(1000000)]) # 惰性计算,内存更友好 total = sum(x*x for x in range(1000000))
9. 单元测试与函数验证
9.1 使用unittest框架
基础测试示例:
import unittest def add(a, b): return a + b class TestMathFunctions(unittest.TestCase): def test_add_positive_numbers(self): self.assertEqual(add(2, 3), 5) def test_add_negative_numbers(self): self.assertEqual(add(-1, -1), -2) def test_add_zero(self): self.assertEqual(add(0, 0), 0) if __name__ == '__main__': unittest.main()9.2 使用pytest框架
更简洁的测试写法:
# test_math.py import pytest def test_add_positive_numbers(): assert add(2, 3) == 5 def test_add_negative_numbers(): assert add(-1, -1) == -2 @pytest.mark.parametrize("a,b,expected", [ (0, 0, 0), (1, 1, 2), (2, 3, 5), ]) def test_add_multiple_cases(a, b, expected): assert add(a, b) == expected9.3 测试覆盖率
使用pytest-cov检查测试覆盖率:
pytest --cov=my_module tests/9.4 模拟外部依赖
使用unittest.mock测试有外部依赖的函数:
from unittest.mock import patch def get_user_name(user_id): # 假设这是一个会实际调用API的函数 response = requests.get(f"https://api.example.com/users/{user_id}") return response.json()['name'] class TestGetUserName(unittest.TestCase): @patch('requests.get') def test_get_user_name(self, mock_get): # 配置mock返回值 mock_get.return_value.json.return_value = {'name': 'Alice'} # 调用被测函数 result = get_user_name(123) # 验证结果 self.assertEqual(result, 'Alice') # 验证mock被正确调用 mock_get.assert_called_once_with("https://api.example.com/users/123")10. 函数版本管理与兼容性
10.1 函数版本控制策略
语义化版本:
- 主版本号:不兼容的API修改
- 次版本号:向下兼容的功能新增
- 修订号:向下兼容的问题修正
多版本共存方案:
# v1 版本 def process_data_v1(data): # 旧版实现 pass # v2 版本 def process_data_v2(data, new_feature=False): # 新版实现 pass # 兼容性包装函数 def process_data(data, version=2, **kwargs): if version == 1: return process_data_v1(data) elif version == 2: return process_data_v2(data, **kwargs) else: raise ValueError("不支持的版本")10.2 弃用警告
使用warnings模块标记即将废弃的函数:
import warnings def old_function(): warnings.warn( "old_function已废弃,将在v2.0移除,请使用new_function代替", DeprecationWarning, stacklevel=2 ) # 旧实现10.3 类型注解与静态检查
Python 3.5+支持类型注解,配合mypy进行静态检查:
from typing import List, Tuple, Optional def analyze_data( data: List[float], thresholds: Tuple[float, float] = (0.0, 1.0), normalize: bool = True ) -> Optional[Dict[str, float]]: """ 分析数据并返回统计信息 Args: data: 待分析的数据列表 thresholds: 高低阈值 normalize: 是否归一化数据 Returns: 包含统计信息的字典,如果数据为空则返回None """ if not data: return None # 分析逻辑... return {"mean": sum(data)/len(data), "max": max(data)}运行静态检查:
mypy --strict your_module.py11. 函数组合与管道操作
11.1 基础函数组合
手动实现函数组合:
def compose(*funcs): """从右到左组合多个函数""" def wrapper(arg): result = arg for f in reversed(funcs): result = f(result) return result return wrapper # 使用示例 add1 = lambda x: x + 1 square = lambda x: x * x composed = compose(square, add1) print(composed(2)) # (2 + 1)^2 = 911.2 使用第三方库
使用toolz库实现更强大的函数组合:
from toolz import compose, pipe # compose方式 func = compose(str, lambda x: x * 2, float) print(func("3.14")) # "6.28" # pipe方式(更易读) result = pipe("3.14", float, lambda x: x * 2, str) print(result) # "6.28"11.3 数据处理管道
构建数据处理流水线:
def data_pipeline(): """构建数据处理管道""" steps = [ load_data, clean_data, lambda df: normalize(df, columns=['age', 'income']), lambda df: add_features(df, ['age', 'income']), save_to_database ] return compose(*reversed(steps)) # 使用管道 process = data_pipeline() process("input.csv")12. 函数式编程实践
12.1 不可变数据结构
使用元组和命名元组实现数据不可变性:
from collections import namedtuple Point = namedtuple('Point', ['x', 'y']) def move_point(p, dx, dy): """返回新点,原对象不变""" return Point(p.x + dx, p.y + dy) p = Point(1, 2) new_p = move_point(p, 3, 4) print(p) # Point(x=1, y=2) print(new_p) # Point(x=4, y=6)12.2 纯函数实现
纯函数的特点:
- 相同输入总是返回相同输出
- 没有副作用(不修改外部状态)
- 不依赖外部状态
# 纯函数 def pure_add(a, b): return a + b # 非纯函数(依赖外部状态) total = 0 def impure_add(a): global total total += a return total12.3 柯里化技术
将多参数函数转换为单参数函数链:
from functools import partial def curry(f): """自动柯里化装饰器""" def curried(*args, **kwargs): if len(args) + len(kwargs) >= f.__code__.co_argcount: return f(*args, **kwargs) return partial(curried, *args, **kwargs) return curried @curry def add_three_numbers(a, b, c): return a + b + c # 逐步调用 add_5 = add_three_numbers(5) add_5_and_10 = add_5(10) result = add_5_and_10(15) # 3013. 元编程与动态函数
13.1 动态创建函数
使用types.FunctionType动态创建函数:
import types def create_adder(n): """创建增加n的函数""" code = compile(f'def adder(x): return x + {n}', '<string>', 'exec') namespace = {} exec(code, namespace) return namespace['adder'] add5 = create_adder(5) print(add5(10)) # 1513.2 函数属性扩展
为函数添加自定义属性:
def timer(func): """记录函数调用次数的装饰器""" func.call_count = 0 def wrapper(*args, **kwargs): func.call_count += 1 return func(*args, **kwargs) wrapper.original = func return wrapper @timer def example(): pass example() example() print(example.call_count) # 213.3 函数签名检查
使用inspect模块分析函数:
import inspect def example(a, b=1, *args, c=2, **kwargs): pass sig = inspect.signature(example) print(sig.parameters) # OrderedDict([ # ('a', <Parameter "a">), # ('b', <Parameter "b=1">), # ('args', <Parameter "*args">), # ('c', <Parameter "c=2">), # ('kwargs', <Parameter "**kwargs">) # ])14. 函数性能分析与优化
14.1 使用timeit测量执行时间
import timeit def test_func(): return sum(i*i for i in range(10000)) # 测量执行时间 time = timeit.timeit(test_func, number=1000) print(f"平均执行时间: {time/1000:.6f}秒")14.2 使用cProfile分析性能
import cProfile def slow_function(): return sum(pow(i, 3) for i in range(10000)) profiler = cProfile.Profile() profiler.enable() slow_function() profiler.disable() profiler.print_stats(sort='cumtime')14.3 性能优化技巧
使用局部变量:
def unoptimized(): return sum(math.sqrt(i) for i in range(10000)) def optimized(): sqrt = math.sqrt # 局部变量查找更快 return sum(sqrt(i) for i in range(10000))避免不必要的属性查找:
# 较慢 def slow(): for i in range(10000): if some_object.some_attribute: pass # 更快 def fast(): attr = some_object.some_attribute for i in range(10000): if attr: pass使用内置函数和库:
# 较慢的手动实现 def slow_sum(numbers): total = 0 for n in numbers: total += n return total # 使用内置sum def fast_sum(numbers): return sum(numbers)
15. 跨文件函数组织与导入
15.1 模块化组织原则
合理的函数文件组织:
project/ ├── utils/ │ ├── __init__.py │ ├── file_utils.py │ ├── math_utils.py │ └── date_utils.py ├── core/ │ ├── __init__.py │ ├── processor.py │ └── analyzer.py └── main.py15.2 导入最佳实践
绝对导入:
from utils.file_utils import read_large_file延迟导入(减少启动时间):
def expensive_operation(): import heavy_module # 在需要时才导入 return heavy_module.compute()循环导入解决方案:
- 将导入移到函数内部
- 重构代码消除循环依赖
- 使用接口抽象
15.3__init__.py技巧
控制模块的公开API:
# utils/__init__.py from .file_utils import ( read_large_file, save_to_file ) from .math_utils import calculate_stats __all__ = [ 'read_large_file', 'save_to_file', 'calculate_stats' ]这样用户可以从包根目录导入:
from utils import read_large_file16. 函数与面向对象编程
16.1 方法 vs 函数
类方法本质上也是函数,但自动接收self参数:
class Calculator: # 实例方法 def add(self, a, b): return a + b # 类方法