Python函数与模块
函数用于复用逻辑,模块用于拆分文件。初学者写 Python 时,最常见的问题是把所有代码堆在一个文件里。随着功能变多,代码会越来越难读、难改、难测试。
好的 Python 项目通常由小函数、清晰模块和明确入口组成。
函数是什么
函数是一段有名字、可重复调用的逻辑。
def add(a, b):
return a + b
result = add(1, 2)
print(result)执行流程:
flowchart TD
A["调用函数"] --> B["绑定参数"]
B --> C["执行函数体"]
C --> D{"是否执行 return"}
D -- "是" --> E["返回指定结果"]
D -- "否" --> F["默认返回 None"]类型提示
def add(a: int, b: int) -> int:
return a + b类型提示不会在运行时强制检查,但有三个好处:
- 读代码时更容易知道参数含义。
- IDE 能提供更准确的补全。
- 静态检查工具能提前发现错误。
类型提示不是装饰品。项目变大后,函数签名就是“函数契约”:调用方要传什么、函数会返回什么、异常情况下怎么处理,都应该尽量明确。
def normalize_asset_code(raw_code: str) -> str:
code = raw_code.strip().upper()
if not code:
raise ValueError("asset code is required")
return code这个函数的契约是:
- 输入必须是字符串。
- 返回标准化后的字符串。
- 空字符串不是合法资产编号,会抛异常。
如果不写类型和校验,错误数据可能会一路传到数据库或接口调用时才暴露,排查成本会高很多。
参数类型
位置参数
def create_user(name: str, age: int):
print(name, age)
create_user("Tom", 18)关键字参数
create_user(name="Tom", age=18)关键字参数更清晰,尤其适合参数较多的函数。
默认参数
def connect(host: str, port: int = 3306):
print(host, port)默认参数适合提供常用值。不要使用可变对象作为默认参数。
为什么不要这样写:
def append_error(message: str, errors: list[str] = []) -> list[str]:
errors.append(message)
return errors默认参数在函数定义时创建一次,不是每次调用都创建。多次调用会共享同一个 errors。
正确写法:
def append_error(message: str, errors: list[str] | None = None) -> list[str]:
if errors is None:
errors = []
errors.append(message)
return errorsflowchart TD
A["定义函数"] --> B["创建默认参数对象"]
B --> C["第一次调用使用它"]
B --> D["第二次调用仍使用它"]
C --> E["append 修改同一个 list"]
D --> E可变参数
def total(*nums: int) -> int:
return sum(nums)
print(total(1, 2, 3))关键字可变参数
def print_info(**kwargs):
for key, value in kwargs.items():
print(key, value)
print_info(name="Tom", age=18)return 的作用
def check_score(score: int) -> str:
if score >= 60:
return "pass"
return "fail"return 会结束函数执行。利用提前返回可以减少嵌套:
def get_discount(user):
if user is None:
return 0
if not user.get("vip"):
return 0
return 0.8提前返回特别适合校验型代码:
def can_import_asset(user: dict, asset: dict) -> bool:
if not user.get("active"):
return False
if asset.get("hospital_id") not in user.get("hospital_ids", []):
return False
if "asset:import" not in user.get("permissions", []):
return False
return True如果全部写成多层嵌套,代码会越来越难读。函数设计里一个重要原则是:先处理异常和边界,再写主流程。
作用域:变量到底从哪里找
Python 查找变量时有一套作用域规则,常用 LEGB 理解:
| 层级 | 含义 | 示例 |
|---|---|---|
| Local | 当前函数内部 | 函数里的局部变量 |
| Enclosing | 外层函数作用域 | 闭包里的外层变量 |
| Global | 模块全局变量 | .py 文件顶层变量 |
| Built-in | 内置作用域 | len、dict、print |
flowchart TD
A["使用变量 name"] --> B{"Local 有吗"}
B -- "有" --> C["使用 Local"]
B -- "没有" --> D{"Enclosing 有吗"}
D -- "有" --> E["使用 Enclosing"]
D -- "没有" --> F{"Global 有吗"}
F -- "有" --> G["使用 Global"]
F -- "没有" --> H{"Built-in 有吗"}
H -- "有" --> I["使用 Built-in"]
H -- "没有" --> J["NameError"]示例:
rate = 0.8
def calculate(price: float) -> float:
tax = 1.06
return price * rate * taxtax 来自 Local,rate 来自 Global。
不要随便在函数里修改全局变量:
count = 0
def increase() -> None:
global count
count += 1这样会让函数依赖外部状态,测试和并发都更麻烦。更推荐把状态作为参数传入,或封装到对象里。
闭包:函数记住外部状态
闭包是指内部函数引用了外部函数的变量,即使外部函数已经执行完,内部函数仍然能使用那些变量。
def make_discount(rate: float):
def apply(price: float) -> float:
return price * rate
return apply
vip_discount = make_discount(0.8)
print(vip_discount(100)) # 80.0流程:
flowchart TD
A["调用 make_discount(0.8)"] --> B["创建 rate=0.8"]
B --> C["定义内部函数 apply"]
C --> D["apply 引用 rate"]
D --> E["返回 apply"]
E --> F["后续调用 apply 时仍能使用 rate"]闭包适合生成带配置的小函数,例如折扣函数、字段清洗函数、校验函数。但不要把复杂业务状态都藏进闭包,否则调试困难。
装饰器:在不改函数主体的情况下增强函数
装饰器本质上是一个“接收函数、返回新函数”的函数。它常用于日志、耗时统计、权限校验、重试、缓存。
最小示例:
from functools import wraps
def log_time(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(f"start {func.__name__}")
result = func(*args, **kwargs)
print(f"finish {func.__name__}")
return result
return wrapper
@log_time
def import_asset(file_name: str) -> int:
print(f"import {file_name}")
return 100@log_time 等价于:
import_asset = log_time(import_asset)执行流程:
flowchart TD
A["定义 import_asset"] --> B["传给 log_time"]
B --> C["返回 wrapper"]
C --> D["import_asset 指向 wrapper"]
D --> E["调用 import_asset"]
E --> F["wrapper 中执行增强逻辑"]
F --> G["调用原始函数"]functools.wraps 很重要。它会保留原函数名、文档等元信息。不加它时,调试、日志、FastAPI 文档、测试报告可能都只看到 wrapper。
商业装饰器 Demo:接口重试
调用外部医院接口可能偶发超时,可以用装饰器给函数加重试。但重试必须谨慎:查询类接口通常可以重试,创建订单、扣款、写数据这类非幂等操作不能随便重试。
from functools import wraps
from time import sleep
from typing import Callable, TypeVar
T = TypeVar("T")
def retry(times: int, interval_seconds: float):
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@wraps(func)
def wrapper(*args, **kwargs) -> T:
last_error: Exception | None = None
for index in range(times):
try:
return func(*args, **kwargs)
except Exception as exc:
last_error = exc
if index < times - 1:
sleep(interval_seconds)
assert last_error is not None
raise last_error
return wrapper
return decorator使用:
@retry(times=3, interval_seconds=0.5)
def fetch_asset(asset_code: str) -> dict:
# 这里可以放 requests.get(..., timeout=3)
return {"asset_code": asset_code}这个装饰器背后的原则:
| 设计 | 原因 |
|---|---|
| 限制次数 | 避免无限重试拖垮系统 |
| 间隔等待 | 给下游恢复时间 |
| 最终抛出异常 | 不吞掉失败,方便上层记录和补偿 |
| 只用于幂等操作 | 避免重复写入或重复扣款 |
lambda
lambda 适合非常短的匿名函数:
nums = [1, 2, 3]
result = list(map(lambda x: x * 2, nums))复杂逻辑不要写成 lambda,直接定义函数更清楚。
错误示例:
rows.sort(key=lambda row: row["name"].strip().lower() if row.get("name") else "")这类逻辑已经不短,建议拆函数:
def sort_name(row: dict[str, str]) -> str:
name = row.get("name")
if not name:
return ""
return name.strip().lower()
rows.sort(key=sort_name)代码不是越短越好,越容易读、越容易测才是生产项目更看重的。
模块是什么
一个 .py 文件就是一个模块。
# math_utils.py
def add(a: int, b: int) -> int:
return a + b导入模块:
from math_utils import add
print(add(1, 2))import 执行流程
flowchart TD
A["执行 import"] --> B["按 sys.path 查找模块"]
B --> C["加载模块文件"]
C --> D["执行模块顶层代码"]
D --> E["缓存到 sys.modules"]注意:模块第一次被导入时,顶层代码会执行。因此不要在模块顶层写复杂业务逻辑。
import 为什么会执行顶层代码
假设有文件 config_loader.py:
print("load config")
def get_config() -> dict:
return {"env": "dev"}另一个文件:
import config_loader导入时会打印 load config。因为 Python 加载模块时会执行模块顶层语句,把函数、类、变量都创建出来,然后缓存到 sys.modules。
这就是为什么下面这种写法危险:
# bad_module.py
import requests
response = requests.get("https://example.com/config", timeout=10)
CONFIG = response.json()任何地方只要 import 这个模块,就会发网络请求。启动慢、测试慢、网络失败导致 import 失败,都会出现。
推荐:
def load_config() -> dict:
# 真正需要时再加载
return {"env": "dev"}__name__ == "__main__" 为什么重要
脚本入口要放在这个判断里:
def main() -> None:
print("run task")
if __name__ == "__main__":
main()原因:
| 场景 | __name__ 值 | 是否执行 main |
|---|---|---|
| 直接运行当前文件 | "__main__" | 是 |
| 被其他模块 import | 模块名 | 否 |
如果没有这个保护,测试文件或其他模块导入它时,脚本会自动执行,可能误删文件、误发接口请求、误跑批处理。
包是什么
包是包含多个模块的目录,通常包含 __init__.py。
project/
app/
__init__.py
main.py
config.py
services/
__init__.py
user_service.py
repositories/
__init__.py
user_repository.py常见分层:
| 目录 | 职责 |
|---|---|
api | 路由和接口 |
schemas | 请求和响应结构 |
services | 业务逻辑 |
repositories | 数据访问 |
config.py | 配置 |
tests | 测试 |
从脚本到工程:怎么拆模块
很多新手最开始会写一个 main.py:
# main.py
# 读取文件、清洗、校验、请求接口、写数据库、打印日志全部在这里这在几十行时还能接受,到了几百行会出现:
- 很难找到某段逻辑。
- 函数之间互相依赖。
- 想测试清洗逻辑,必须准备文件、数据库、网络。
- 任何小修改都可能影响全局。
推荐按职责拆:
asset_importer/
app/
__init__.py
main.py
config.py
reader.py
validator.py
cleaner.py
writer.py
service.py
tests/
test_validator.py
test_cleaner.py职责:
| 文件 | 职责 |
|---|---|
main.py | 程序入口,解析参数,调用 service |
config.py | 配置读取 |
reader.py | 读取 CSV/Excel/JSON |
validator.py | 校验必填、格式、唯一性 |
cleaner.py | 字段标准化、脱敏 |
writer.py | 写结果文件或数据库 |
service.py | 编排完整导入流程 |
拆模块的标准不是“文件越多越好”,而是每个模块有清晰职责,能单独测试。
循环导入
循环导入是初学者常见问题:
a.py 导入 b.py
b.py 又导入 a.py解决思路:
- 抽出公共逻辑到第三个模块。
- 减少模块顶层执行逻辑。
- 重新划分职责,避免两个模块互相依赖。
循环导入为什么会报错
假设:
# a.py
from b import func_b
def func_a():
return "a"# b.py
from a import func_a
def func_b():
return "b"加载流程:
flowchart TD
A["导入 a.py"] --> B["a.py 导入 b.py"]
B --> C["开始加载 b.py"]
C --> D["b.py 又导入 a.py"]
D --> E["a.py 还没加载完"]
E --> F["拿不到完整对象,报错或行为异常"]解决方式一:抽公共模块。
# common.py
def normalize_name(name: str) -> str:
return name.strip()a.py 和 b.py 都依赖 common.py,不要互相依赖。
解决方式二:把 import 放到函数内部,只在真正执行时导入。但这通常是临时方案,根本上还是要调整模块职责。
一个小项目示例
todo_app/
app/
__init__.py
main.py
todo_service.py
tests/
test_todo_service.pytodo_service.py:
def add_todo(todos: list[str], title: str) -> list[str]:
if not title.strip():
raise ValueError("title不能为空")
todos.append(title)
return todosmain.py:
from app.todo_service import add_todo
def main():
todos = []
add_todo(todos, "学习 Python")
print(todos)
if __name__ == "__main__":
main()商业场景 Demo:资产导入工具
下面把函数和模块放进一个真实场景:写一个资产导入预处理工具。它读取 CSV 行,校验必填字段,标准化资产编号,输出可导入的数据和错误数据。
目录结构
asset_importer/
app/
__init__.py
main.py
validator.py
cleaner.py
service.py
tests/
test_validator.pyvalidator.py
def validate_required(row: dict[str, str]) -> list[str]:
errors: list[str] = []
if not row.get("asset_code", "").strip():
errors.append("资产编号不能为空")
if not row.get("asset_name", "").strip():
errors.append("资产名称不能为空")
if not row.get("dept_code", "").strip():
errors.append("科室不能为空")
return errorscleaner.py
def normalize_asset(row: dict[str, str]) -> dict[str, str]:
return {
"asset_code": row["asset_code"].strip().upper(),
"asset_name": row["asset_name"].strip(),
"dept_code": row["dept_code"].strip().upper(),
}service.py
from app.cleaner import normalize_asset
from app.validator import validate_required
def preview_import(rows: list[dict[str, str]]) -> tuple[list[dict[str, str]], list[dict[str, str]]]:
valid_rows: list[dict[str, str]] = []
error_rows: list[dict[str, str]] = []
for index, row in enumerate(rows, start=1):
errors = validate_required(row)
if errors:
error_row = dict(row)
error_row["line_no"] = str(index)
error_row["errors"] = ";".join(errors)
error_rows.append(error_row)
continue
valid_rows.append(normalize_asset(row))
return valid_rows, error_rowsmain.py
from app.service import preview_import
def main() -> None:
rows = [
{"asset_code": " a001 ", "asset_name": " CT ", "dept_code": " d01 "},
{"asset_code": "", "asset_name": "MRI", "dept_code": "D02"},
]
valid_rows, error_rows = preview_import(rows)
print("valid:", valid_rows)
print("errors:", error_rows)
if __name__ == "__main__":
main()为什么这样拆:
| 设计 | 好处 |
|---|---|
| 校验函数独立 | 可以单独测试,不依赖文件 |
| 清洗函数独立 | 字段标准化逻辑集中 |
| service 编排流程 | 主流程清楚,便于扩展 |
| main 只做入口 | 被测试导入时不会自动执行 |
测试:
from app.validator import validate_required
def test_validate_required_should_return_error_when_asset_code_empty():
row = {"asset_code": "", "asset_name": "CT", "dept_code": "D01"}
errors = validate_required(row)
assert "资产编号不能为空" in errors这个 Demo 体现了函数和模块的核心价值:不是为了拆而拆,而是让业务逻辑可读、可测、可替换。
函数设计原则
| 原则 | 解释 | 不这样会怎样 |
|---|---|---|
| 单一职责 | 一个函数只做一件明确的事 | 函数越来越长,测试困难 |
| 输入输出明确 | 参数和返回值清楚 | 调用方不知道怎么用 |
| 少依赖全局状态 | 尽量通过参数传入 | 测试难、并发不安全 |
| 边界先返回 | 先处理异常情况 | 嵌套越来越深 |
| 异常不吞掉 | 失败要保留原因 | 线上只看到结果错误 |
| 纯函数优先 | 同输入同输出,不改外部状态 | 副作用难追踪 |
纯函数示例:
def calculate_total(price: float, count: int) -> float:
return price * count有副作用的函数:
def save_order(order: dict) -> None:
# 写数据库,这是副作用
pass副作用不是不能有,而是要集中在边界层,例如数据库、文件、网络调用。核心规则函数尽量保持纯,这样更容易测试。
开发建议
- 一个函数只做一件明确的事。
- 函数长度过长时,优先拆小函数。
- 重要函数加类型提示和必要 docstring。
- 模块命名使用小写加下划线。
- 避免在模块顶层执行数据库连接、网络请求等复杂逻辑。
- 脚本入口统一放到
main()。
生产排查:函数和模块相关问题
flowchart TD
A["Python 项目异常"] --> B{"是否导入时报错"}
B -- "是" --> C["检查循环导入和顶层副作用"]
B -- "否" --> D{"是否函数结果不稳定"}
D -- "是" --> E["检查全局变量、可变默认参数、副作用"]
D -- "否" --> F{"是否难以测试"}
F -- "是" --> G["拆小函数,隔离 IO 和业务规则"]
F -- "否" --> H["继续查日志、数据和外部依赖"]常见问题:
| 现象 | 可能原因 | 处理 |
|---|---|---|
| import 就发请求 | 模块顶层有网络调用 | 改到函数里,入口显式调用 |
cannot import name | 循环导入或模块未加载完 | 抽公共模块,调整依赖方向 |
| 测试一跑就执行脚本 | 缺少 if __name__ == "__main__" | 入口保护 |
| 多次调用结果互相污染 | 可变默认参数或全局变量 | 默认值用 None,减少全局状态 |
| 函数很难测 | 函数里混了文件、网络、数据库 | 拆出纯业务函数 |
面试标准回答
函数和模块的作用
函数用于封装可复用逻辑,模块用于把代码按文件和职责组织起来。项目变大后,如果所有代码都写在一个文件里,会难以阅读、测试和维护。好的 Python 项目会把入口、配置、业务规则、数据访问、文件处理和测试拆开,函数签名尽量清晰,并用类型提示表达输入输出契约。Python import 做了什么
import 会按 sys.path 查找模块,加载模块文件,执行模块顶层代码,并把模块对象缓存到 sys.modules。后续再次 import 通常直接复用缓存。所以模块顶层不要写数据库连接、网络请求、批处理执行等复杂逻辑,否则只要被导入就会产生副作用。循环导入怎么解决
循环导入通常说明模块职责划分有问题。a.py 导入 b.py,b.py 又导入 a.py 时,可能出现某个模块还没加载完就被另一个模块使用。解决方式是抽公共逻辑到第三个模块,减少顶层执行逻辑,或重新划分依赖方向。把 import 放到函数内部可以临时缓解,但根本上还是要拆清职责。装饰器是什么
装饰器本质是接收函数并返回新函数的函数。它可以在不修改原函数主体的情况下增加日志、耗时统计、权限校验、重试、缓存等能力。@decorator 等价于 func = decorator(func)。写装饰器时通常要用 functools.wraps 保留原函数元信息。练习
- 写一个
calculate_area(width, height)函数并加类型提示。 - 写一个带默认参数的连接函数。
- 把一个脚本拆成
main.py和utils.py两个模块。 - 故意制造循环导入,再尝试通过抽公共模块解决。
- 给函数增加输入校验,非法输入抛出
ValueError。
小结
函数让逻辑可复用,模块让文件可组织。写 Python 项目时,先从小函数开始,再按职责拆分模块。避免顶层复杂逻辑、避免循环导入、给关键函数添加类型提示,是入门到工程化的关键一步。
