一、基础语法与数据类型 1. 变量、赋值与命名 py是动态类型语言,变量本身不固定保存某一种类型,它更像是一个名字,运行时绑定到某个对象。
1 2 3 4 name = "Alice" age = 20 pi = 3.14159 is_student = True
变量名常用命名规则:
只能由字母、数字和下划线组成,不能以数字开头
区分大小写,name和Name是两个不同的变量
变量、函数通常使用snake_case
类名通常使用PascalCase
常量没有语法层面的强制限制,约定使用UPPER_CASE
例如:
1 2 3 4 5 user_name = "Alice" MAX_RETRY = 3 class UserProfile : pass
py支持多重赋值和交换变量:
1 2 3 4 a, b = 1 , 2 a, b = b, aprint (a, b)
赋值不会自动复制对象。例如:
1 2 3 4 5 a = [1 , 2 , 3 ] b = a b.append(4 )print (a)
此时a和b都绑定到同一个列表对象。
2. 常见基本类型 常见内置类型:
类型
示例
说明
int
10
整数
float
3.14
浮点数
bool
True、False
布尔值
str
"hello"
字符串
NoneType
None
表示“没有值”
list
[1, 2]
可变序列
tuple
(1, 2)
不可变序列
dict
{"name": "Alice"}
键值映射
set
{1, 2, 3}
无重复元素集合
可以使用type()查看对象类型:
需要判断一个对象是否属于某种类型时,工程代码中通常优先使用isinstance():
1 2 3 4 value = 10 if isinstance (value, int ): print ("value是整数" )
isinstance()还会正确处理继承关系。
3. 数值运算 常见运算符:
1 2 3 4 5 6 7 8 9 10 11 12 13 a + b a - b a * b a / b a // b a % b a ** b
例如:
1 2 3 4 print (7 / 2 ) print (7 // 2 ) print (7 % 2 ) print (2 ** 3 )
复合赋值:
4. 比较、逻辑与成员运算 比较运算:
逻辑运算:
成员运算:
例如:
1 2 3 4 nums = [1 , 2 , 3 ]print (2 in nums) print (5 not in nums)
is和==不要混淆:
==通常比较两个对象的值是否相等
is比较两个变量是否指向同一个对象
1 2 3 4 5 6 7 a = [1 , 2 ] b = [1 , 2 ] c = aprint (a == b) print (a is b) print (a is c)
判断None时通常写:
而不是value == None。
5. 真值判断 在if、while等条件中,py会把对象转换为布尔意义上的真假。
常见假值包括:
1 2 3 4 5 6 7 8 9 False None 0 0.0 "" [] {}set () ()
例如:
1 2 3 4 items = []if not items: print ("列表为空" )
因此一般不需要写:
1 2 if len (items) == 0 : ...
6. 类型转换 常见转换函数:
1 2 3 4 5 6 7 int ("123" )float ("3.14" )str (100 )bool (1 )list ((1 , 2 , 3 ))tuple ([1 , 2 , 3 ])set ([1 , 1 , 2 ])
需要注意,类型转换可能失败:
因此处理外部输入时通常需要进行校验或异常处理。
7. 字符串基础与f-string 字符串可以使用单引号、双引号或三引号:
1 2 3 4 s1 = 'hello' s2 = "world" s3 = """这是一个 多行字符串"""
字符串是不可变对象,对字符串执行“修改”操作通常会产生新字符串。
现代py中推荐使用f-string进行格式化:
1 2 3 4 5 name = "Alice" age = 20 message = f"我是{name} ,今年{age} 岁" print (message)
格式控制:
1 2 pi = 3.1415926 print (f"{pi:.2 f} " )
8. 解包与星号表达式 序列解包:
1 2 point = (10 , 20 ) x, y = point
可以使用*接收剩余元素:
1 2 3 4 5 first, *middle, last = [1 , 2 , 3 , 4 , 5 ]print (first) print (middle) print (last)
*也可以展开可迭代对象:
1 2 3 4 5 a = [1 , 2 ] b = [3 , 4 ] combined = [*a, *b]print (combined)
**可以展开字典:
1 2 3 4 base = {"name" : "Alice" , "age" : 20 } updated = {**base, "age" : 21 }print (updated)
后面的键会覆盖前面同名的键。
二、常用容器 1. list 列表 列表是有顺序、可修改的序列:
1 2 3 4 nums = [10 , 20 , 30 ]print (nums[0 ]) print (nums[-1 ])
常见操作:
1 2 3 4 5 nums.append(40 ) nums.extend([50 , 60 ]) nums.insert(1 , 15 ) nums.remove(20 ) last = nums.pop()
切片:
1 2 3 4 5 6 nums = [0 , 1 , 2 , 3 , 4 , 5 ]print (nums[1 :4 ]) print (nums[:3 ]) print (nums[::2 ]) print (nums[::-1 ])
注意:普通切片会创建一个新的列表,但列表中的元素如果本身是可变对象,仍可能共享内部对象引用。
2. tuple 元组 元组是不可变序列:
单元素元组必须保留逗号:
元组常用于:
表示不应该被修改的一组值
多返回值
字典键(前提是其中元素也可哈希)
函数所谓的“返回多个值”,本质上通常是返回一个元组:
1 2 3 4 def get_point (): return 10 , 20 x, y = get_point()
3. dict 字典 字典保存键值对:
1 2 3 4 user = { "name" : "Alice" , "age" : 20 , }
访问和修改:
1 2 3 print (user["name" ]) user["age" ] = 21 user["city" ] = "Beijing"
若键可能不存在,使用get()更安全:
1 city = user.get("city" , "未知" )
遍历:
1 2 3 4 5 for key in user: print (key)for key, value in user.items(): print (key, value)
常用方法:
1 2 3 4 user.keys() user.values() user.items() user.pop("age" )
4. set 集合 集合中的元素不会重复,常用于去重和快速成员判断:
1 2 3 nums = {1 , 2 , 3 } nums.add(4 ) nums.discard(2 )
空集合必须使用:
因为:
集合运算:
1 2 3 4 5 6 7 a = {1 , 2 , 3 } b = {3 , 4 , 5 }print (a | b) print (a & b) print (a - b) print (a ^ b)
5. 可变对象与不可变对象 常见不可变对象:
int
float
bool
str
tuple(前提是内部对象本身不被修改)
frozenset
常见可变对象:
这个区别会直接影响:
函数传参后的修改效果
是否能作为dict的键
是否能放入set
浅拷贝/深拷贝行为
6. 浅拷贝与深拷贝 对于嵌套可变对象,仅复制最外层容器时,内部对象仍然会共享引用:
1 2 3 4 5 original = [[1 , 2 ], [3 , 4 ]] shallow = original.copy() shallow[0 ].append(99 )print (original)
需要递归复制内部对象时,可以使用:
1 2 3 import copy deep = copy.deepcopy(original)
但深拷贝并不是任何场景都应该默认使用;工程代码中更重要的是明确对象所有权和是否需要共享状态。
三、控制流 1. if / elif / else 1 2 3 4 5 6 7 8 score = 85 if score >= 90 : print ("A" )elif score >= 80 : print ("B" )else : print ("C" )
条件表达式(三元表达式):
1 status = "通过" if score >= 60 else "不通过"
2. for 循环 py中的for通常不是“按下标循环”,而是直接从可迭代对象中依次取值:
1 2 for item in [10 , 20 , 30 ]: print (item)
需要下标时使用enumerate():
1 2 for index, item in enumerate ([10 , 20 , 30 ]): print (index, item)
同时遍历多个序列可以使用zip():
1 2 3 4 5 names = ["Alice" , "Bob" ] scores = [90 , 80 ]for name, score in zip (names, scores): print (name, score)
3. while、break、continue 1 2 3 4 5 6 7 8 9 10 11 12 count = 0 while count < 5 : count += 1 if count == 2 : continue if count == 4 : break print (count)
break:立即结束当前循环
continue:跳过本轮剩余代码,进入下一轮
循环还支持else:
1 2 3 4 5 6 for number in [1 , 3 , 5 ]: if number % 2 == 0 : print ("找到偶数" ) break else : print ("没有找到偶数" )
这里的else只有在循环没有被break提前结束 时才执行。
4. match / case(Python 3.10+) 结构化模式匹配适合处理具有明确形状的多分支数据:
1 2 3 4 5 6 7 8 9 10 def handle_command (command ): match command: case ["quit" ]: return "退出" case ["open" , filename]: return f"打开文件:{filename} " case ["move" , x, y]: return f"移动到:{x} , {y} " case _: return "未知命令"
不要把match简单理解成其他语言的switch。它不仅能比较固定值,还可以对列表、元组、字典、类等结构进行模式匹配。
5. pass pass表示什么都不做,常用于暂时占位:
四、函数参数:*args 与 **kwargs 1. 可变位置参数 *args *args将传入的位置参数打包为一个元组
1 2 3 4 5 6 def sum_all (*args ): print (f"args的类型: {type (args)} " ) print (f"args的内容: {args} " ) return sum (args)print (sum_all(1 , 2 , 3 , 4 ))
1 2 3 4 5 6 7 8 9 def print_names (*args ): for i, name in enumerate (args, 1 ): print (f"{i} . {name} " ) print_names("小明" , "小红" , "小华" )
注:enumerate() 是 Python 内置函数,用于在遍历可迭代对象时同时获取索引和值。
2. 可变关键字参数 **kwargs **kwargs 将传入的关键字参数打包成一个字典
1 2 3 4 5 6 7 8 9 10 11 12 def show_info (**kwargs ): print (f"kwargs的类型: {type (kwargs)} " ) print (f"kwargs的内容: {kwargs} " ) for key, value in kwargs.items(): print (f"{key} : {value} " ) show_info(name="张三" , age=25 , city="北京" )
3. 混合使用与参数顺序 在这个例子中,参数顺序是:普通参数 → *args → 仅限关键字参数 → **kwargs
其中,写在*args后面的default只能通过关键字传入,因此属于仅限关键字参数。
1 2 3 4 5 6 7 8 9 10 11 12 def mixed (param1, param2, *args, default="默认值" , **kwargs ): print (f"普通参数: {param1} , {param2} " ) print (f"*args: {args} " ) print (f"默认参数: {default} " ) print (f"**kwargs: {kwargs} " ) mixed(1 , 2 , 3 , 4 , 5 , default="自定义" , name="李四" , age=30 )
4. 位置参数、关键字参数与 /、* 除了*args和**kwargs,py还可以显式限制参数的传递方式。
位置参数与关键字参数 1 2 3 4 5 6 7 8 def greet (name, message="你好" ): print (f"{message} ,{name} " ) greet("Alice" , "欢迎" ) greet(name="Alice" , message="欢迎" )
仅限位置参数:/ /前面的参数只能按位置传入:
1 2 3 4 5 def divide (a, b, / ): return a / b divide(10 , 2 )
仅限关键字参数:* 单独的*之后的参数只能通过关键字传入:
1 2 3 4 def connect (host, *, timeout=5 , retry=3 ): print (host, timeout, retry) connect("example.com" , timeout=10 , retry=2 )
完整参数顺序可以理解为:
1 仅限位置参数 / 普通参数 / *args / 仅限关键字参数 / **kwargs
实际代码中不一定需要把所有形式同时用上,但读标准库、FastAPI或第三方SDK时经常会看到/和*。
5. 调用函数时的 * 与 ** 解包 *args和**kwargs不仅可以出现在函数定义中,也可以用于函数调用时展开参数:
1 2 3 4 5 def add (a, b, c ): return a + b + c nums = [1 , 2 , 3 ]print (add(*nums))
字典可以通过**展开成关键字参数:
1 2 3 4 5 def show (name, age ): print (name, age) info = {"name" : "Alice" , "age" : 20 } show(**info)
五、函数进阶 1. 函数也是对象 在py中,函数是一个真正的对象。能够在后面加()进行调用的对象,称为可调用对象。
所以函数可以作为值赋给变量,也可以作为参数传递,也可以作为返回值。
装饰器能够成立,其根本原因为:
函数可以接收另一个函数,也可以返回一个新函数
2. lambda 表达式(匿名函数) lambda真正适合的场景是:
某个地方需要传入一个简单函数,而这个函数只用一次。
注: 下面部分代码主要用于展示lambda支持的各种参数形式。实际编程中,如果一个函数需要起名字并反复使用,通常更推荐使用def;lambda更适合临时作为其他函数的参数使用。
基本语法
冒号后面的表达式计算出什么,函数就自动返回什么。
注:函数调用也属于表达式,所以你也可以在返回值表达式里调用一个函数,那么lambda就会返回这个函数的返回值。
因为py中函数也是对象,所以lambda还可以返回一个函数,就像下面这样:
1 create_multiplier = lambda number: lambda value: value * number
这个写法会形成闭包,闭包将在后面详细讲解。它等价于:
1 2 3 4 5 def create_multiplier (number ): def multiply (value ): return value * number return multiply
lambda 函数的调用方式 a. 先保存,再调用 1 2 3 4 5 double = lambda x: x * 2 result = double(10 )print (result)
b. 创建后立即调用 1 2 3 result = (lambda x: x * 2 )(10 )print (result)
参数形式 a. 单个参数 1 2 3 4 5 6 7 8 9 10 11 12 square = lambda x: x ** 2 print (square(5 )) to_upper = lambda s: s.upper()print (to_upper("hello" )) to_int = lambda s: int (s) if s.isdigit() else 0 print (to_int("123" )) print (to_int("abc" ))
b. 多个参数 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 add = lambda x, y: x + yprint (add(10 , 20 )) multiply_three = lambda x, y, z: x * y * zprint (multiply_three(2 , 3 , 4 )) distance = lambda x1, y1, x2, y2: ((x2 - x1)**2 + (y2 - y1)**2 )**0.5 print (distance(0 , 0 , 3 , 4 )) bmi = lambda weight, height: weight / (height ** 2 )print (bmi(70 , 1.75 ))
c. 默认参数 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 greet = lambda name, greeting="Hello" : f"{greeting} , {name} !" print (greet("Alice" )) print (greet("Bob" , "Hi" )) power = lambda x, exp=2 : x ** expprint (power(5 )) print (power(2 , 3 )) def add_item (item=0 , lst=[] ): lst.append(item) return lstprint (add_item(3 , [1 , 2 ])) print (add_item(item=5 )) print (add_item(item=10 ))
d. 无参数 1 2 3 4 5 6 7 8 9 10 11 12 get_five = lambda : 5 print (get_five()) import time get_current_time = lambda : time.time()print (get_current_time()) result = (lambda : 3.14159 )()print (result)
3. 闭包 闭包就是:
内部函数引用了外部函数的变量,即使外部函数已经执行完毕,内部函数仍然保留着这些变量所处的环境。
从更准确的角度说,闭包是:
内部函数,加上它所记住的外部变量。
例如:
1 2 3 4 5 6 7 8 9 10 11 def outer (): message = "你好" def inner (): print (message) return inner func = outer() func()
执行过程:
调用外层函数:
py此时进入outer():
然后创建内部函数:
1 2 def inner (): print (message)
最后:
注意到不是return inner(),inner()表示立即调用函数,而inner表示函数对象本身
故func = outer()实际上可以理解为func = inner
调用内部函数: func()相当于调用inner() ,而inner()中需要使用message,但是inner自己内部没有定义message,于是它会去外层寻找,然后找到message = "你好"。 虽然outer()已经运行结束,但是inner仍然保存着对这个变量的引用,因此可以继续使用它。
闭包的基本条件 一般需要同时满足下面三个条件:
存在一个外层函数
外层函数里面定义了一个内层函数
内层函数使用了外层函数的变量。被返回或传递出去不是闭包成立的必要条件,但这样最能体现外层函数结束后变量仍被保存
nonlocal 的作用1 2 3 4 5 6 7 8 9 10 def create_counter (): count = 0 def counter (): count += 1 return count return counter
原因是只要函数内部出现:count = ···,py就会默认认为count是这个函数自己的局部变量,但执行右边的count+1时,这个局部变量还没有值,所以报错。
这个时候加上nonlocal count,就是告诉py,这里的count不是内层函数自己的变量,而是外层函数里的那个count。
1 2 3 4 5 6 7 8 9 def create_counter (): count = 0 def counter (): nonlocal count count += 1 return count return counter
为什么每个闭包能保存自己的数据 仍然以上面的例子举例:
1 2 3 4 5 6 7 8 9 def create_counter (): count = 0 def counter (): nonlocal count count += 1 return count return counter
假如此时分别创建两个计数器:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 counter1 = create_counter() counter2 = create_counter()print (counter1())print (counter1())print (counter2())print (counter1())print (counter2())
因为每调用一次create_counter()都会创建一套新的环境,和使用类创建多个对象有点类似,每个对象拥有自己的属性,而闭包则是每个函数拥有自己记住的外层变量。
闭包的延迟绑定陷阱 1 2 3 4 5 6 7 8 functions = []for i in range (3 ): functions.append(lambda : i)print (functions[0 ]())print (functions[1 ]())print (functions[2 ]())
输出不是012,而是222。
因为这些lambda记住的不是创建函数那一刻i当前的具体值,而是变量i本身,是对变量i的引用。
延迟绑定可以理解为:
函数不是在创建时马上把i的值抄下来,而是在真正执行时再去查i现在是多少。
解决方法:利用默认参数:
1 2 3 4 functions = []for i in range (3 ): functions.append(lambda i=i: i)
使用默认参数来保存每次创建函数时变量i的值。
4. 装饰器 装饰器适合用来处理当多个函数都需要添加某个公共功能的情况,这样就不用在每个函数里重复写相同代码。
基本语法 创建一个装饰器:
1 2 3 4 5 6 7 8 9 10 11 def decorator (func ): def wrapper (*args, **kwargs ): ... result = func(*args, **kwargs) ... return result return wrapper
注意到最后返回的是wrapper而不是wrapper()。装饰完成后,外部的原函数名会重新指向wrapper,而wrapper中的func仍然指向原函数。
装饰器的核心公式:
比如说下面这个timer函数,就是一个装饰器。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 import timedef timer (func ): """一个计时装饰器,测量函数的执行时间""" def wrapper (*args, **kwargs ): start = time.perf_counter() result = func(*args, **kwargs) end = time.perf_counter() print (f"{func.__name__} 执行耗时:{end - start:.4 f} 秒" ) return result return wrapperdef slow_add (a, b ): time.sleep(0.1 ) return a + b slow_add = timer(slow_add) result = slow_add(1 , 2 )print (f"结果:{result} " )
timer接收一个函数func,返回一个新函数wrapper,wrapper在调用func前后加了计时逻辑,最后slow_add = timer(slow_add)把原函数替换成包装后的版本。
@ 语法糖1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 import timedef timer (func ): def wrapper (*args, **kwargs ): start = time.perf_counter() result = func(*args, **kwargs) end = time.perf_counter() print (f"{func.__name__} 执行耗时:{end - start:.4 f} 秒" ) return result return wrapper@timer def slow_add (a, b ): time.sleep(0.1 ) return a + b@timer def slow_multiply (a, b ): time.sleep(0.2 ) return a * b r1 = slow_add(1 , 2 )print (r1) r2 = slow_multiply(3 , 4 )print (r2)
@timer写在函数定义上方,py在定义函数后自动执行slow_add = timer(slow_add)
为什么装饰器需要嵌套函数 如果直接写:
1 2 3 4 5 6 def timer (func ): start = time.perf_counter() result = func() end = time.perf_counter() print (end - start) return result
的话,这样在执行timer(slow_add)的时候,就会立即调用原函数。
我们真正需要的是,timer(slow_add)先生成一个新函数,等以后调用这个新函数时,再执行计时逻辑。所以需要分成两个阶段:
1 2 3 4 5 6 7 装饰阶段: slow_add = timer(slow_add) 生成 wrapper,但不执行原函数 调用阶段: slow_add(1, 2) 执行 wrapper,再由 wrapper 调用原函数
装饰器实际上运用了闭包,此处的闭包为:闭包 = wrapper函数 + 它记住的原函数func,而此处的func指向原来的slow_add。
装饰器有个小问题:包装之后,原函数的名字和文档字符串会丢失。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 def my_decorator (func ): def wrapper (*args, **kwargs ): return func(*args, **kwargs) return wrapper@my_decorator def hello (): """这是 hello 函数的文档""" print ("Hello!" )print (hello.__name__)print (hello.__doc__)
因为hello实际上已经被替换成了wrapper,所以__name__和__doc__都是wrapper的,这在调试的时候会造成误解。
解决方法是用functools.wraps装饰wrapper:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 import functoolsdef my_decorator (func ): @functools.wraps(func ) def wrapper (*args, **kwargs ): return func(*args, **kwargs) return wrapper@my_decorator def hello (): """这是hello函数的文档""" print ("Hello!" )print (hello.__name__)print (hello.__doc__)
写装饰器时在wrapper上方加上这行,就能把原函数的__name__、__doc__等属性复制过来。
常用装饰器
@functools.cache(Python 3.9+):无限制记忆化 会缓存函数的返回值,相同的参数只计算一次。刷算法题做记忆化搜索时,如果不需要限制缓存大小,写起来最直接。
@functools.lru_cache:带LRU策略的缓存 可以通过maxsize限制缓存大小;@functools.lru_cache(maxsize=None)的效果与@functools.cache类似。
@staticmethod 和 @classmethod: 这两个用在类里面,调整方法和类的关系。
带参数的装饰器 如果你想让装饰器本身也能传参数,那么需要再套一层函数:
1 2 3 4 5 6 7 8 9 def 装饰器工厂 (装饰器参数 ): def 装饰器 (原函数 ): def 包装函数 (*args, **kwargs ): ... return 原函数(*args, **kwargs) return 包装函数 return 装饰器
例如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 from functools import wrapsdef repeat (n ): """让被装饰的函数重复执行n次""" def decorator (func ): @wraps(func ) def wrapper (*args, **kwargs ): result = None for _ in range (n): result = func(*args, **kwargs) return result return wrapper return decorator@repeat(3 ) def say_hello (name ): print (f"你好,{name} !" ) say_hello("Alice" )
整个过程实际上分为两步:
decorator = repeat(3)
say_hello = decorator(say_hello)
即:
1 2 3 4 5 6 7 8 9 10 repeat(times) │ ├── 接收装饰器自己的参数 times │ └── decorator(func) │ ├── 接收被装饰函数 │ └── wrapper(*args, **kwargs) └── 每次函数调用时真正执行
最外层接收装饰器的参数,中间层接收被装饰的函数,最内层是实际执行的包装函数。
多个装饰器的执行顺序 1 2 3 4 @decorator_a @decorator_b def func (): pass
等价于:
1 func = decorator_a(decorator_b(func))
调用时表现为从外向内:
1 2 3 4 5 6 7 8 9 进入 decorator_a 的 wrapper ↓ 进入 decorator_b 的 wrapper ↓ 执行原函数 ↓ 退出 decorator_b ↓ 退出 decorator_a
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 from functools import wrapsdef outer (func ): @wraps(func ) def wrapper (*args, **kwargs ): print ("outer 开始" ) result = func(*args, **kwargs) print ("outer 结束" ) return result return wrapperdef inner (func ): @wraps(func ) def wrapper (*args, **kwargs ): print ("inner 开始" ) result = func(*args, **kwargs) print ("inner 结束" ) return result return wrapper@outer @inner def work (): print ("执行 work" )
最后输出:
1 2 3 4 5 outer 开始 inner 开始 执行 work inner 结束 outer 结束
小结
函数装饰器通常利用闭包保存被装饰的原函数。装饰时只创建并返回包装函数,不立即调用原函数;等以后调用包装后的函数时,包装函数才通过闭包中保存的 func 调用原函数。这属于延迟执行。
六、迭代器与生成器 1. 可迭代对象与迭代器 可迭代对象是能够传给iter()并获得迭代器的对象,例如列表、元组、字符串和range。
迭代器是能够通过next()依次取值的对象;迭代器本身也是可迭代对象,iter(iterator)通常会返回它自己。
迭代器只能依次消费
迭代器内部会记录“当前走到哪里了”。每取出一个元素,位置就向后移动一步;已经取出的元素通常不能回退,全部取完后,迭代器就耗尽了。
即迭代器已经把这个元素取出来了,并把当前位置向后移动了。
“消费”指的是消耗迭代器中的遍历进度,原对象中的原始数据并没有被改变。
for 循环也在消费迭代器 for循环的底层逻辑大概是:
1 2 3 4 5 6 7 8 iterator = iter (z)while True : try : item = next (iterator) print (item) except StopIteration: break
也就是说,for 循环也会不断调用 next()。
list()、map() 和 filter() 的消费时机比如:
list(z)不是查看一下z,而是不断调用next(z),直到取完为止,近似相当于:
1 2 3 4 5 6 7 8 result = []while True : try : item = next (z) result.append(item) except StopIteration: break
因此这会造成,两次print(list(z))之后,第二次打印出来的是空列表。
map()和filter()本身返回惰性迭代器,创建它们时通常不会立刻取完输入;当它们被next()、for或list()继续迭代时,才会逐步消费输入迭代器。
为什么 zip() 不直接返回列表? 因为返回迭代器有一个很大的优点:不需要一开始就把所有结果全部存进内存,而是需要一个时才生成一个。
这种方法也叫惰性计算 ,也就是用到时再计算。优点是节省内存,适合大量数据,代价是它一般只能向前走,不能随意回退。
2. 生成器 生成器是py中一种按需产生数据的迭代器,每次需要一个值时,才计算并返回一个值;下次继续从上次暂停的位置运行。即惰性计算 。
可以把生成器理解成:
一个“可以暂停,可以继续执行,并且每次只吐出一个值”的函数执行过程。
例如:
1 2 3 4 def generate (): yield 1 yield 2 yield 3
此时generate是一个生成器函数,而g = generate()得到的是一个生成器对象。
调用生成器函数时,函数体不会立即执行 1 2 3 4 5 6 7 8 9 10 11 def generate (): print ("函数开始" ) yield 10 print ("继续执行" ) yield 20 print ("函数结束" ) g = generate()print ("创建完毕" )
此时输出只会有创建完毕。而:
仅仅是在创建一个生成器对象。如果这时候加入:
此时py才会真正开始执行generate函数,在yield 10执行完之后暂停。第二次再调用next()的时候,才会继续执行yield 10后面的代码。
为什么for循环可以直接遍历生成器? 例如:
1 2 3 4 5 6 7 8 def generate (): yield 10 yield 20 yield 30 for x in generate(): print (x)
结果为10 20 30.
因为前面说过for循环本质上大概就是不停地next(),直到生成器耗尽:
1 2 3 4 5 6 7 8 iterator = iter (generate())while True : try : x = next (iterator) print (x) except StopIteration: break
生成器的主要创建方法
生成器表达式
带有yield的生成器函数
生成器表达式 比如:
1 number = [x * x for x in range (1_000_000 )]
这是列表推导式,它会计算0到100万-1的平方,然后把100万个结果全部存起来。
而把[]换成()就是生成器表达式:
1 numbers = (x * x for x in range (1_000_000 ))
只有逐步执行next(numbers)的时候,才会逐步计算从0到100万-1的平方。
七、推导式 1. 列表推导式的基本语法
可以理解为:
从“可迭代对象”中依次取出每个元素,交给“变量”,计算“表达式”,然后把每次计算的结果放进新列表。
若变量在前面的表达式中未使用到,那么可以理解为:可迭代对象每产生一个元素,就把前面的表达式计算一次。执行次数由可迭代对象中的元素数量决定。
2. 条件筛选与条件表达式 带筛选条件:
1 [表达式 for 变量 in 可迭代对象 if 条件]
if-else放在前面:将符合条件的元素转换为相应结果
1 [结果1 if 条件 else 结果2 for 变量 in 可迭代对象]
3. 在列表推导式中调用函数 可以在表达式部分调用函数,例如:
1 2 3 4 def square (x ): return x * x result = [square(i) for i in range (5 )]
4. 二维列表的共享引用陷阱 创建二维列表千万不能用 [[0]*4]*3。因为 *3 复制的是引用,三行指向的是同一个列表对象,修改一行其他行也会跟着变。用列表推导式 [[0]*cols for _ in range(rows)] 才是正确做法,每行都是独立的列表对象。
例如:
1 2 3 4 5 grid = [[0 ] * 4 ] * 3 grid[0 ][1 ] = 5 print (grid)
输出:
`[[0, 5, 0, 0],
[0, 5, 0, 0],
[0, 5, 0, 0]]`
5. 集合推导式与字典推导式 集合推导式:
1 2 squares = {x * x for x in range (5 )}print (squares)
字典推导式:
1 2 squares = {x: x * x for x in range (5 )}print (squares)
也可以配合条件:
1 2 3 4 5 even_squares = { x: x * x for x in range (10 ) if x % 2 == 0 }
列表、集合、字典推导式适合表达简单的“遍历 + 转换/筛选”。如果逻辑已经包含很多层条件、异常处理或副作用,通常改用普通for循环会更清晰。
6. 推导式中的作用域 Python 3中,推导式内部的循环变量拥有自己的局部作用域,不会直接泄漏到外层:
1 2 3 4 5 x = 100 nums = [x * 2 for x in range (3 )]print (x) print (nums)
八、常用内置函数与数学工具 1. range() 生成一个整数范围,常用于for循环,包含范围为前闭后开
1 2 3 range (stop)range (start, stop)range (start, stop, step)
返回值:返回一个range对象
步长step不能为0
2. enumerate() 同时获得下标和元素
基本语法 1 enumerate (iterable, start=0 )
参数:
iterable:需要遍历的可迭代对象
start:编号从哪个数字开始,默认从 0 开始
返回值:
一个enumerate迭代器对象
遍历时,每次产生一个(编号,元素)元组
3. zip() 基本用法 zip() 是 Python 中用来把多个可迭代对象中相同位置的元素配对组合的函数。
最常见的场景是:同时遍历两个或多个列表。可以把它理解成把两个列表“拉上拉链”。
1 2 3 4 5 6 7 ["小明", "小红", "小刚"] ↓ ↓ ↓ [ 90, 85, 92 ] 组合后: [("小明", 90), ("小红", 85), ("小刚", 92)]
注:
zip()返回的是一个迭代器对象
可以组合不同类型的可迭代对象
配合 for 循环使用 1 2 3 4 5 names = ["小明" , "小红" , "小刚" ] scores = [90 , 85 , 92 ]for name, score in zip (names, scores): print (name, score)
同时组合三个或更多可迭代对象 1 2 3 4 5 6 7 names = ["小明" , "小红" , "小刚" ] scores = [90 , 85 , 92 ] ages = [18 , 19 , 18 ] result = list (zip (names, scores, ages))print (result)
可迭代对象长度不同时 zip()会以最短的可迭代对象为准
解压操作 例如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 students = [ ("小明", 90), ("小红", 85), ("小刚", 92) ] names, scores = zip(*students) print(names) print(scores) 得到: ("小明", "小红", "小刚") (90, 85, 92)
通过这一解压操作还能用于实现矩阵转置 :
1 2 3 4 5 6 7 8 9 10 11 matrix = [ [1, 2, 3], [4, 5, 6] ] result = list(zip(*matrix)) print(result) 得到: [(1, 4), (2, 5), (3, 6)]
若想让结果仍为二维列表:
1 2 3 4 5 6 7 8 9 10 result = [list(row) for row in zip(*matrix)] print(result) 得到: [ [1, 4], [2, 5], [3, 6] ]
4. map() 将指定函数依次应用于可迭代对象中的每一个元素
基本语法
返回值:一个map迭代器对象
可处理多个可迭代对象 1 map (function, iterable1, iterable2, ...)
例如:
1 2 3 4 5 6 numbers1 = [1 , 2 , 3 ] numbers2 = [10 , 20 , 30 ] result = map (lambda x, y: x + y, numbers1, numbers2)print (list (result))
若多个可迭代对象长度不同,map()会在最短的可迭代对象结束时停止。
示例 map(int, str(n))会依次把字符串中的每个字符转换为整数。
1 2 3 4 5 n = 124 str (n) map (int , str (n)) sorted (..., reverse=True )
5. filter() 对每个元素进行条件判断,只保留判断结果为真的元素。
基本语法 1 filter (function, iterable)
判断函数function应该接受一个元素,并返回一个布尔值,但判断函数返回的也不一定非得是布尔值,因为py会根据一个值的“真假”判断是否保留元素,常见假值包括:
1 2 3 4 5 6 7 8 False None 0 0.0 "" [] {}set ()
其他大多数值都是真值。
注: filter()的第一个参数还可以直接写成None,此时会自动删除所有假值。
6. sorted() 对可迭代对象进行排序,返回一个新的列表
1 sorted (iterable, key=None , reverse=False )
参数:key为指定排序依据,reverse为是否降序,默认False表示升序
返回值:返回一个新的列表
注:sorted()与列表的.sort()不同,后者会修改原列表,且返回None。
7. reversed() 反向遍历一个序列,不会修改原序列
参数:sequence为列表、元组、字符串、range等可逆序对象
返回值:返回一个反向迭代器
8. round() 对数字进行舍入
1 round (number, ndigits=None )
参数:number为需要舍入的数字,ndigits为保留的小数位数,省略时取整
返回值:返回舍入后的数字
9. divmod() 同时计算整除的商和余数
参数:a为被除数,b为除数
返回值:返回一个二元组(商,余数)
10. ord() 把单个Unicode字符转换为对应的整数码点
11. chr() 把整数码点转换为对应的Unicode字符
1 2 3 4 5 6 7 8 9 10 11 12 text = "abracadabra" freq = [0 ] * 26 for ch in text: freq[ord (ch) - ord ('a' )] += 1 for i in range (26 ): if freq[i] > 0 : print (f"{chr (i + ord ('a' ))} : {freq[i]} " , end=" " )print ()
ord(ch) - ord('a') 这个技巧把小写字母映射到 0~25,非常适合用数组代替哈希表来统计字母频率。刷题时很常用。
12. max()、min() 返回可迭代对象中的最值,或多个位置参数中的最值
基本语法 1 2 3 4 5 max(iterable, *, key=None, default=...) 或者: max(arg1, arg2, *args, key=None)
参数:
iterable:需要查找最大值的可迭代对象,例如列表、元组、集合等
arg1, arg2, …:也可以直接传入多个值进行比较
key:指定比较规则,默认直接比较元素本身
default:当 iterable 为空时返回的默认值
返回值:
返回最大的那个原元素
如果使用 key,根据 key(element) 的结果进行比较,但返回的仍然是原元素 ,例如:
1 max ([-10 , 3 , 5 ], key=abs )
min()同理
13. pow() 进行幂运算
基本语法 1 2 3 4 5 普通幂: pow(base, exp) 带模运算: pow(base, exp, mod)
返回值:
快速幂取模 1 2 3 4 5 6 7 8 MOD = 10 **9 + 7 result = pow (2 , 100 , MOD)print (f"2^100 mod (10^9+7) = {result} " )
14. math 模块 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 import mathprint (f"sqrt(9) = {math.sqrt(9 )} " )print (f"sqrt(2) = {math.sqrt(2 ):.4 f} " )print (f"ceil(3.2) = {math.ceil(3.2 )} " ) print (f"floor(3.8) = {math.floor(3.8 )} " ) print (f"gcd(12, 8) = {math.gcd(12 , 8 )} " ) print (f"gcd(12, 8, 6) = {math.gcd(12 , 8 , 6 )} " ) print (f"log2(8) = {math.log2(8 )} " ) print (f"log(e) = {math.log(math.e)} " ) print (f"log10(1000) = {math.log10(1000 )} " ) print (f"math.inf = {math.inf} " )print (f"5 < math.inf ? {5 < math.inf} " )
注:
python中表示无穷大还有另一种方式float('inf'),表示负无穷可以写成float('-inf')或-math.inf
python中的整除//是向下取整
15. 进制转换相关函数 bin()将整数转换为二进制字符串 ,不改变原数字
例如:
1 2 3 4 5 6 7 8 x = bin (42 )print (x)print (type (x))
注:
0b是二进制前缀
负数也可以,只不过返回的是带负号的字符串,不是计算机内部补码表示
oct()将整数转换为八进制字符串
注:
0o表示八进制
hex()将整数转换为十六进制字符串
注:
0x表示十六进制
显示时默认显示小写字母
int()创建或转换成整数
1 2 3 int (x)int (string, base)
参数:
base:字符串当前使用的是几进制注 :若base=0,则py会根据前缀自动判断进制
返回值:
九、常用对象方法 1. str.strip() 删除字符串开头和结尾的空白字符或指定字符,即只删除两端,不删除中间。
1 string.strip(chars=None )
参数:chars为要删除的字符集合,默认删除空格、换行、制表符等
返回值:返回处理后的新字符串
相关方法:
1 2 text.lstrip() text.rstrip()
2. str.split() 按照指定分隔符拆分字符串
1 string.split(sep=None , maxsplit=-1 )
参数:sep为分隔符,默认按照任意空白字符拆分;maxsplit为最多拆分多少次,默认全部拆分
返回值:返回一个字符串列表
3. str.replace() 将字符串中的指定内容替换成新内容
1 string.replace(old, new, count=-1 )
参数:old为要被替换的内容,new为替换后的内容,count为最多替换多少次,默认全部替换
4. str.join() 使用指定字符串,将多个字符串连接起来
1 separator.join(iterable)
参数:separator为元素之间使用的连接符;iterable为包含字符串的可迭代对象
返回值:返回连接后的新字符串
5. list.sort() 使用元组进行多级排序 元组在比较时会先比第一个元素,相等再比第二个,以此类推。对于数字字段,如果想让某一项降序 ,可以对这一项取负号
例如:
1 2 3 4 students = [("Alice" , 88 ), ("Bob" , 95 ), ("Charlie" , 88 ), ("Diana" , 95 )] students.sort(key=lambda x: (-x[1 ], x[0 ]))print (students)
6. dict.get() 根据键获取字典中的值;如果键不存在,不会报错,而是返回一个默认值。
参数:key为要查找的键,必填;default为键不存在时返回的默认值,可选,默认是None
返回值:若键存在,返回该键对应的值;若键不存在,返回default,未传入default时返回None
频率统计的经典写法 1 freq[ch] = freq.get(ch, 0 ) + 1
十、异常处理 1. try / except / else / finally 完整结构 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 try : result = 10 / 0 except ZeroDivisionError: print ("除数不能为零" ) except (TypeError, ValueError) as e: print (f"发生错误: {e} " ) except Exception as e: print (f"未知错误: {e} " ) else : print ("计算成功!" ) finally : print ("这是最后执行的代码" )
2. 常见异常类型
异常类型
说明
示例
ZeroDivisionError
除零错误
1/0
TypeError
类型错误
‘1’ + 1
ValueError
值错误
int(‘abc’)
IndexError
索引错误
[1][10]
KeyError
键错误
{}[‘key’]
FileNotFoundError
文件不存在
open(‘nofile.txt’)
1 2 3 4 5 6 7 8 9 10 BaseException ├── SystemExit ├── KeyboardInterrupt ├── GeneratorExit └── Exception # ← 我们通常处理的异常 ├── TypeError # 类型错误 ├── ValueError # 值错误 ├── KeyError # 键错误 ├── FileNotFoundError └── ... (其他所有异常)
3. 自定义异常 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 class AgeError (Exception ): """年龄异常类""" pass class UnderAgeError (AgeError ): """年龄太小异常""" pass class OverAgeError (AgeError ): """年龄太大异常""" pass def check_age (age ): if age < 0 : raise AgeError("年龄不能为负数" ) elif age < 18 : raise UnderAgeError("年龄太小,需要满18岁" ) elif age > 120 : raise OverAgeError("年龄太大,超出范围" ) else : print (f"年龄 {age} 验证通过" )try : check_age(15 )except UnderAgeError as e: print (f"未成年: {e} " )except OverAgeError as e: print (f"超龄: {e} " )except AgeError as e: print (f"年龄错误: {e} " )
十一、面向对象基础 面向对象的思路是:
把一组相关的数据以及操作这些数据的函数,封装为一个整体
这个整体称为对象,对象所属的类型称为类
1. 类与对象 用class关键字定义一个类,类就是一个模版,定义了这类对象有什么属性(数据)和什么方法(操作)
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 class Student : def __init__ (self, name, age ): self .name = name self .age = age self .scores = [] def add_score (self, score ): self .scores.append(score) def average (self ): if not self .scores: return 0 return sum (self .scores) / len (self .scores) alice = Student("Alice" , 20 ) bob = Student("Bob" , 21 ) alice.add_score(90 ) alice.add_score(85 ) alice.add_score(92 ) bob.add_score(78 ) bob.add_score(88 )print (f"{alice.name} 的平均分:{alice.average()} " )print (f"{bob.name} 的平均分:{bob.average()} " )
普通的py自定义类默认允许随时读取和修改属性,也可以动态添加新属性;但__slots__、property等机制可以对属性访问进行限制
实例属性与类属性 a. 实例属性与类属性的区别
实例属性:通过self.xxx设置,每个实例都有独立的数据
类属性:通常用于所有实例共享的数据
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 class Student : school = "BUPT" def __init__ (self, name ): self .name = name self .scores = [] alice = Student("Alice" ) bob = Student("Bob" ) alice.scores.append(100 )print (alice.scores) print (bob.scores) print (Student.school) print (alice.school) print (bob.school)
实例变量通常用于保存每个对象各自独有的数据
b. 实例同名属性与类属性 1 2 3 4 5 alice.school = "Alice 的学校" print (alice.school) print (bob.school) print (Student.school)
此处并未修改 Student.school,而是给 alice 新增了一个同名实例属性。
属性查找大致是:
先找 alice 自己的实例属性 ↓ 找不到再找 Student 类属性 ↓ 再找父类中的属性
当实例属性与类属性同名时,实例属性优先
c. 可变类属性的共享问题 类属性如果是列表、字典、集合等可变对象,会被实例共享。如果本来就需要所有实例共享,这样写没有问题;如果希望每个实例各自拥有一份,就应该放在__init__中。
1 2 3 4 5 6 class Student : def __init__ (self, name ): self .name = name self .scores = []
属性访问与“封装”
单下划线_name: 表示这是内部属性,外部代码最好不要直接操作,但实际上仍可以访问。仅为约定,不是强制限制。
双下划线__name : 双下划线主要是为了避免子类属性重名,不是安全机制,也不等于真正私有。
2. Python 中的三种方法 实例方法 第一个参数是self,用于操作某个具体对象
适用场景:方法需要读取或修改实例属性
类方法 使用@classmethod,第一个参数通常写成cls,表示当前类
1 2 3 4 5 6 7 8 9 10 11 class Student : school = "BUPT" def __init__ (self, name, age ): self .name = name self .age = age @classmethod def from_text (cls, text ): name, age = text.split("," ) return cls(name, int (age))
调用:
1 2 3 4 alice = Student.from_text("Alice,20" )print (alice.name) print (alice.age)
整个过程详细解释:
因为 from_text 被 @classmethod 修饰,Python 会自动把 Student 类传给第一个参数 cls。所以这里的cls本质上就是Student,即 1 2 cls = Student text = "Alice, 20"
然后拆分字符串,转换年龄类型
接着创建对象,return cls(name, int(age))等价于return Student("Alice", 20),这会自动调用Student.__init__(新对象, "Alice", 20)
最终得到: 1 2 student.name student.age
a. 为什么不直接写 return Student(name, int(age)) 使用cls的核心优势是:
谁调用这个类方法,就创建谁的对象
静态方法 使用@staticmethod,不会自动接收 self 或 cls。
1 2 3 4 class Student : @staticmethod def is_valid_score (score ): return 0 <= score <= 100
调用:
1 2 print (Student.is_valid_score(90 )) print (Student.is_valid_score(120 ))
它本质上更像一个普通函数,只是逻辑上和Student类关系密切,所以放在类里面。
3. 继承与 super() 基础继承 继承就是在已有类的基础上扩展。子类可以按照属性查找规则使用父类中可继承的属性和方法,还可以添加新的或者重写已有的。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 class Animal : def __init__ (self, name ): self .name = name def speak (self ): return "..." class Cat (Animal ): def speak (self ): return "喵~" class Dog (Animal ): def speak (self ): return "汪!" cat = Cat("小花" ) dog = Dog("旺财" )print (f"{cat.name} 说:{cat.speak()} " )print (f"{dog.name} 说:{dog.speak()} " )
Cat(Animal) 表示 Cat 继承自 Animal。Cat 没有定义 __init__,所以自动用 Animal 的。
Cat 重写了 speak() 方法,调用时会用 Cat 自己的版本。
扩展 __init__ 如果子类需要在父类的基础上扩展 __init__,可以用 super() 调用父类的方法:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 class Animal : def __init__ (self, name ): self .name = name def info (self ): return f"名字:{self.name} " class Pet (Animal ): def __init__ (self, name, owner ): super ().__init__(name) self .owner = owner def info (self ): base_info = super ().info() return f"{base_info} ,主人:{self.owner} " pet = Pet("小白" , "张三" )print (pet.info())
super() 返回一个代理对象,通过它可以按照方法解析顺序MRO继续查找方法。在这个单继承例子中,super().__init__(name) 就是调用父类 Animal 的 __init__,确保 name 属性被正确设置,然后子类再添加自己的 owner 属性。
4. 魔术方法 以双下划线开头和结尾的特殊方法,由py在特定场景下自动调用
__call__() 与类装饰器对象实现__call__()后,对象本身就可以像函数一样使用。
1 2 3 4 5 6 7 8 9 class Timer : def __init__ (self, func ): self .func = func def __call__ (self, *args, **kwargs ): print ("开始执行" ) result = self .func(*args, **kwargs) print ("执行结束" ) return result
使用:
1 2 3 @Timer def slow_add (a, b ): return a + b
这里没有使用函数闭包,而是把原函数保存在self.func里面。
注: 这个例子主要用于理解__call__()和类装饰器的基本原理。实际工程中的通用类装饰器还可能需要考虑函数元数据、实例方法绑定等问题。
5. @property 普通属性可以直接读取和修改:
1 2 3 class User : def __init__ (self, age ): self .age = age
如果以后希望在读取或修改属性时加入校验逻辑,可以使用property,同时仍然保持obj.age这样的访问形式:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 class User : def __init__ (self, age ): self .age = age @property def age (self ): return self ._age @age.setter def age (self, value ): if value < 0 : raise ValueError("年龄不能为负数" ) self ._age = value user = User(20 )print (user.age) user.age = 21
这里真正保存数据的是_age,而age对外表现得仍像普通属性。
注意:不要为了“面向对象”就把所有属性都机械地写成property。只有在确实需要校验、计算或兼容旧接口时再使用。
6. dataclass 当一个类主要用于保存数据时,手写大量__init__、__repr__、__eq__通常很重复,可以使用@dataclass。
dataclass不是说这个类只能用来存数据,它仍然是一个正常的py类,只不过@dataclass会帮你自动生成一些特别常见的样板代码。
@dataclass是一个类装饰器,专门帮“主要用来存数据的类”自动生成__init__/__repr__、__eq__等重复代码:
1 2 3 4 5 6 7 8 9 10 from dataclasses import dataclass@dataclass class User : name: str age: int user = User("Alice" , 20 )print (user)
带默认值:
1 2 3 4 @dataclass class Config : host: str = "localhost" port: int = 8000
但注意,有默认值的字段必须放后面。
这里的:
会被dataclass看作字段,然后自动生成__init__(self, name, age)。
而name:str是py的类型注解 ,py的普通类型注解默认不会强制进行运行时类型检查。
类型注解主要用于:
阅读代码
IDE提示
静态类型检查器
dataclass判断哪些名称属于字段
__repr__()这个魔术方法和__str__()坐一桌,用来规定一个对象应该以什么字符串形式表示出来。比如普通类如果没有专门定义对象的字符串表示,通常就会看到类似这样的输出:
1 2 3 4 5 6 7 8 9 10 11 class Student : def __init__ (self, name, age ): self .name = name self .age = age s = Student("小明" , 18 )print (s)
如果此时定义__repr__():
1 2 3 4 5 6 7 8 9 class Student : def __init__ (self, name, age ): self .name = name self .age = age def __repr__ (self ): return f"Student(name={self.name!r} , age={self.age!r} )"
基本语法 1 2 def __repr__ (self ): return "表示这个对象的字符串"
__repr__()必须返回字符串str。
可以用print(repr(s))或者在交互解释器里直接输入s来触发,使用repr来表示对象。
!r表示使用这个值的repr()结果,例如:
1 2 3 4 5 6 7 8 name = "小明" print (f"{name} " )print (f"{name!r} " )
__repr__()通常是给程序员,调试器看的对象表示,所以一般希望它信息明确,能看出对象类型,能看出重要属性,如果方便,尽量长得像创建这个对象的代码。
__eq__()这个方法用来规定:两个对象使用==比较时,什么情况下认为它们相等。
例如:
1 2 3 4 5 6 7 8 9 10 11 12 class Student : def __init__ (self, name, age ): self .name = name self .age = age s1 = Student("小明" , 18 ) s2 = Student("小明" , 18 )print (s1 == s2)
虽然s1.name == s2.name s1.age == s2.age都是True,但是s1和s2指向的仍然是两个不同的Student对象,如果没有自定义适当的__eq__(),py不会自动认为属性一样就是相等的。
例如此时我们可以规定,姓名和年龄都相同,就认为两个学生是相等的:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 class Student : def __init__ (self, name, age ): self .name = name self .age = age def __eq__ (self, other ): if not isinstance (other, Student): return NotImplemented return (self .name == other.name and self .age == other.age) s1 = Student("小明" , 18 ) s2 = Student("小明" , 18 ) s3 = Student("小红" , 18 )print (s1 == s2)print (s1 == s3)
列表默认值不能直接写[] 可变默认值不能直接写成[]或{},应该使用default_factory:
1 2 3 4 5 6 from dataclasses import dataclass, field@dataclass class Student : name: str scores: list = field(default_factory=list )
这里的:
1 field(default_factory=list )
表示:每创建一个Student对象,都重新调用一次list(),产生一个新的空列表。
这和普通函数的“可变默认参数”陷阱本质上类似:不应该让多个实例意外共享同一个可变对象。
field()用于对某个dataclass字段进行更加详细的配置。
__post_init__有时候仅靠自动生成的__init__不够,还想再初始化后执行一些额外逻辑时,可以使用__post_init__。
1 2 3 4 5 6 7 8 @dataclass class Student : name:str age:int def __post_init__ (self ): if self .age < 0 : raise ValueError("年龄不能小于0" )
执行Student("小明", -1)就会报错,此时的执行顺序可以理解为:
1 2 3 4 5 6 7 8 Student("小明" , 18 ) ↓ 自动生成的 __init__ ↓self .name = "小明" self .age = 18 ↓ 自动调用 __post_init__()
在__post_init__中是可以创建计算属性的,例如:
1 2 3 4 5 6 7 8 9 10 11 @dataclass class Rectangle : width:float height:float def __post_init__ (self ): self .area = self .width * self .height r = Rectangle(10 , 20 )print (r.area)
@dataclass的主要参数常见写法:
1 2 3 4 5 6 7 8 9 10 @dataclass( init=True , repr =True , eq=True , order=False , frozen=False , slots=False )class Student : ...
a.order参数:是否生成大小比较方法 默认dataclass支持==, !=,但不自动支持<, >, <=, >=,如果把order参数设置为True,那么就可以让对象按照字段定义顺序进行比较。比如:
1 2 3 4 5 6 7 8 9 10 @dataclass(order=True ) class Student : score: int name: str s1 = Student(80 , "小明" ) s2 = Student(90 , "小红" )print (s1 < s2)
这里就是先比较score,如果相同,就会继续比较name,以此类推。
b.frozen参数:是否禁止正常修改字段 这个参数可以创建近似不可变的数据类:
1 2 3 4 5 6 7 @dataclass(frozen = True ) class Point : x:int y:int p = Point(1 , 2 )print (p.x)
但如果p.x = 100的话,就会报错,因此可以理解成创建出来之后,不允许正常地重新修改字段,适合以下字段:
c.slots参数:是否使用slots 现代py中还会经常看到:
1 2 3 4 @dataclass(slots=True ) class Student : name:str age:int
这会使用__slots__机制,可以简单理解为限制对象主要只能拥有预先声明好的字段,并可能减少大量对象的内存开销。
3. __slots__(了解) 普通自定义类实例通常有__dict__,因此可以动态添加属性:
1 2 3 4 5 class User : pass user = User() user.name = "Alice"
某些场景可以使用__slots__限制实例允许拥有的属性,并减少大量实例的部分内存开销:
1 2 3 4 5 6 class Point : __slots__ = ("x" , "y" ) def __init__ (self, x, y ): self .x = x self .y = y
一般业务代码不用为了“优化”主动到处使用__slots__,知道它改变了实例属性存储方式即可。
十二、模块、包与依赖管理 1. 模块与 import 一个.py文件通常就是一个模块。
假设:
1 2 3 project/ ├── main.py └── math_utils.py
math_utils.py:
1 2 def add (a, b ): return a + b
main.py:
1 2 3 import math_utilsprint (math_utils.add(1 , 2 ))
也可以只导入某个名字:
1 2 3 from math_utils import addprint (add(1 , 2 ))
可以起别名:
不推荐:
1 from math_utils import *
因为这会让当前命名空间中突然出现很多名字,难以判断一个变量或函数从哪里来。
2. __name__ 与程序入口 每个模块都有__name__变量。
当文件被直接运行时:
当它被其他模块导入时,__name__通常是模块名。
因此常见写法:
1 2 3 4 5 6 def main (): print ("程序开始" )if __name__ == "__main__" : main()
这样导入这个模块时不会自动执行main()中的程序入口逻辑。
3. 包与 __init__.py 多个模块可以组成一个包:
1 2 3 4 5 6 project/ ├── main.py └── app/ ├── __init__.py ├── service.py └── utils.py
例如:
1 from app.service import run
现代Python支持namespace package,因此某些场景没有__init__.py也能形成包;但普通项目中保留__init__.py仍然很常见,而且可以明确表达这个目录是Python包。
4. 绝对导入与相对导入 绝对导入:
1 from app.utils import parse_data
包内部也可以使用相对导入:
1 2 from .utils import parse_datafrom ..config import settings
.表示当前包,..表示上一级包。
实际工程中通常更推荐清晰的绝对导入;相对导入适合包内部关系明确的场景。
5. 虚拟环境与依赖 不同项目可能依赖不同版本的第三方库,因此一般不要把所有库都装到同一个全局Python环境中。
使用venv:
Windows PowerShell中激活:
1 .\.venv\Scripts\Activate.ps1
安装依赖:
1 python -m pip install fastapi
查看依赖:
现代项目还经常使用pyproject.toml统一保存项目元数据、构建配置和工具配置,并使用uv等工具管理环境和依赖。第一次学习时先理解“项目环境彼此隔离”这个核心概念即可。
十三、类型标注 typing 类型标注本质上是给程序员、IDE、类型检查器、框架看的“类型说明”,
1. 基本类型标注 Python是动态类型语言,类型标注不会把它变成静态类型语言;它主要用于:
提高可读性
IDE自动补全
静态类型检查
框架根据类型生成校验或Schema
例如:
1 2 def add (a: int , b: int ) -> int : return a + b
变量也可以标注:
1 2 name: str = "Alice" age: int = 20
2. 容器类型 Python 3.9+通常可以直接使用内置泛型:
1 2 3 4 names: list [str ] = ["Alice" , "Bob" ] scores: dict [str , int ] = {"Alice" : 90 } point: tuple [int , int ] = (10 , 20 ) tags: set [str ] = {"python" , "agent" }
list[str]说明这是一个“元素应该是字符串”的列表,其余类型同理。
3. 联合类型与可空值 Python 3.10+可以使用|表示联合类型 ,比如说此处的value,就是既可以是int,也可以是str:
1 2 def normalize (value: int | str ) -> str : return str (value)
可能为None:
1 2 3 4 5 def find_user (user_id: int ) -> str | None : if user_id == 1 : return "Alice" return None
所以此处调用find_user之后,不能想当然地认为返回就一定是str,还要考虑返回值为None时代码应该如何处理。
旧代码里还会看到:
1 2 3 4 from typing import Optional , Union Optional [str ] Union [int , str ]
4. Any 与 object Any基本相当于告诉类型检查器:“这里暂时不要检查”:
1 2 3 4 from typing import Any def handle (value: Any ) -> Any : return value.foo.bar()
但不要无脑把所有复杂类型都写成Any,否则类型标注的价值会被抵消。
object则表示“可以接收任何Python对象”,但读取具体属性前仍然需要缩窄类型,因为object并没有告诉类型检查器这个对象具体有哪些能力。
5. Callable 与 Awaitable 函作为参数传递的时候也可以进行类型标注:
1 2 3 4 from collections.abc import Callable def apply_twice (func: Callable [[int ], int ], value: int ) -> int : return func(func(value))
1 2 3 4 Callable[ [int], ← 参数列表 int ← 返回值 ]
意思就是func必须是一个接收一个int,返回一个int的可调用对象。
异步回调中还可能看到:
1 2 3 from collections.abc import Awaitable, Callable AsyncHandler = Callable [[str ], Awaitable[str ]]
其中,Callable[[str], Awaitable[str]]表示接收一个str,返回一个可以被await的对象,并且最终得到str。
所以这里的AsyncHandler大致描述一个异步处理函数,输入字符串,最终异步得到字符串。
这类类型在Agent工具、回调、Middleware和异步SDK中很常见。
6. TypedDict:给字典定义Schema 普通字典中不同键可以有不同的值类型,比如说:
1 2 3 4 user = { "name" : "Alice" , "age" : 20 }
类型标注如果只写dict,或者dict[str, object]的话,信息太少,而TypedDict可以描述这种固定结构:
1 2 3 4 5 6 7 8 9 from typing import TypedDictclass UserData (TypedDict ): name: str age: int def greet (user: UserData ) -> str : return f"你好,{user['name' ]} "
注意:运行时它仍然只是普通dict,TypedDict主要服务于静态类型检查。
7. Protocol:Python的接口 Protocol用于描述“只要拥有这些方法/属性就算满足要求”的结构化接口:
1 2 3 4 5 6 7 8 9 from typing import Protocolclass Reader (Protocol ): def read (self ) -> str : ...def load (reader: Reader ) -> str : return reader.read()
即:一个对象只要拥有符合要求的read() ->str方法,我就可以把它当作Reader。
此时我们定义:
1 2 3 class FileReader : def read (self ) -> str : return "文件内容"
注意到并不是class FileReader(Reader),但load(FileReader())从静态类型角度照样可以成立。因为FileReader刚好满足Reader的def read(self) -> str:要求,这就是所谓的结构化类型 。
调用load()的对象不一定需要显式继承Reader,只要它拥有兼容的read()方法,就可以通过静态类型检查。
这种方式和py的“鸭子类型”很契合:我不管你是不是“鸭子类”;你会鸭子叫、会鸭子走,我就把你当鸭子。
8. TypeVar 与 Generic(了解后会用) 当输入和输出之间存在同一种抽象类型关系时,可以使用泛型:
1 2 3 4 5 6 from typing import TypeVar T = TypeVar("T" )def first (items: list [T] ) -> T: return items[0 ]
T可以理解为:先不知道具体是什么类型,但同一次使用过程中必须保持一致的“类型占位符”。
这样:
1 2 first([1 , 2 , 3 ]) first(["a" , "b" ])
所以T最核心的意义是:表达不同位置之间的类型关系。
复杂泛型不需要刚开始就钻得很深,能够看懂SDK中常见的T、Generic[T]即可。
9. 类型标注不是运行时校验 例如:
1 2 def add (a: int , b: int ) -> int : return a + b
直接调用:
Python运行时不会仅因为标注写了int就自动拒绝它。
FastAPI/Pydantic之所以能根据类型做运行时校验,是因为框架主动读取了这些类型信息并执行了额外逻辑。
十四、文件、路径与 JSON 1. 文件读写 最推荐使用with自动管理文件关闭:
1 2 with open ("data.txt" , "r" , encoding="utf-8" ) as file: content = file.read()
写文件:
1 2 with open ("result.txt" , "w" , encoding="utf-8" ) as file: file.write("hello\n" )
常见模式:
"r":读取
"w":覆盖写入
"a":追加
"b":二进制模式,例如"rb"
逐行读取大型文本时,不一定需要一次性read()到内存,否则会把整个大文件内容一次性加载到内存:
1 2 3 with open ("data.txt" , encoding="utf-8" ) as file: for line in file: print (line.rstrip("\n" ))
文件对象本身可以迭代,这样一次读取一行,处理完再继续读取下一行,内存占用就会小很多。
2. pathlib 以前很多人写路径:
1 2 base = "data" path = base + "/" + "users.json"
这种字符串拼接容易产生平台和路径细节问题。
相比大量手工拼接路径字符串,现代Python中通常更推荐pathlib.Path:
1 2 3 4 5 6 7 8 from pathlib import Path base = Path("data" ) file_path = base / "users.json" print (file_path.name)print (file_path.suffix)print (file_path.exists())
这里的/被Path重载了,表示路径拼接,即data/users.json。
创建目录:
1 Path("output/logs" ).mkdir(parents=True , exist_ok=True )
这两个参数表示:如果父目录不存在就一起创建,目录已经存在也不要报错。
读取和写入小型文本文件:
1 2 3 path = Path("hello.txt" ) path.write_text("你好" , encoding="utf-8" ) content = path.read_text(encoding="utf-8" )
3. JSON JSON经常用于HTTP API、配置文件和数据交换。
Python对象 → JSON字符串,使用dumps():
1 2 3 4 import json data = {"name" : "Alice" , "age" : 20 } text = json.dumps(data, ensure_ascii=False )
JSON字符串 → Python对象,使用loads():
直接处理文件,使用dump()和load():
1 2 3 4 5 6 7 import jsonwith open ("user.json" , "w" , encoding="utf-8" ) as file: json.dump({"name" : "Alice" }, file, ensure_ascii=False , indent=2 )with open ("user.json" , "r" , encoding="utf-8" ) as file: user = json.load(file)
这里的ensure_ascii=False是让中文不要写成\u4f60\u597d,而直接保存你好,而indent=2则是格式化缩进。
常见对应关系:
JSON
Python
object
dict
array
list
string
str
number
int/float
true/false
True/False
null
None
4. os 与环境变量 很多工程配置、API Key不会直接硬编码进代码,而是通过环境变量传入:
1 2 3 import os api_key = os.getenv("API_KEY" )
操作系统环境中:API_KEY=xxxxx程序读取它,如果不存在,则返回None。
如果变量是必须的,也可以显式检查:
1 api_key = os.environ["API_KEY" ]
不存在时会抛出KeyError,因此能更早暴露配置缺失。
实际项目中常见.env文件,但.env不是Python标准库功能,一般由python-dotenv、Pydantic Settings等工具负责读取;敏感.env文件通常不应该提交到Git仓库。
十五、with 语句与上下文管理器 通常打开一个文件之后需要关闭:
1 2 3 file = open ("data.txt" , "r" , encoding="utf-8" ) content = file.read() file.close()
但如果读取过程中发生了异常,比如说:
1 2 3 4 5 6 file = open ("data.txt" , "r" , encoding="utf-8" ) content = file.read() result = 1 / 0 file.close()
则file.close()不会执行。
更安全的写法是使用try-finally语句:
1 2 3 4 5 6 file = open ("data.txt" , "r" , encoding="utf-8" )try : content = file.read()finally : file.close()
而with语句把这种模式简化为:
1 2 with open ("data.txt" , "r" , encoding="utf-8" ) as file: content = file.read()
with open(···) as f:做了两件事:先由open()打开文件并返回文件对象,进入with代码块时把上下文管理器的进入结果赋值给f;离开代码块时自动关闭文件,不管代码块里是正常执行完毕还是抛出异常,文件都会被关闭。
with语句帮你省掉了异常语句和手动关闭文件,代码更简洁,也更不容易忘记释放资源。
1. 基本语法 1 2 with 上下文管理器 [as 变量]: 代码块
此处as变量可以省略,不加as的意思是:只执行上下文管理器的进入和退出逻辑,不需要接收__enter__()的返回值;使用@contextmanager时,这个值就是yield产生的值。
2. with 的工作原理 with语句并不是只能用于文件操作,任何实现了__enter__和__exit__魔术方法的对象都可以用with语句。
__enter__()作用:进入with代码块之前,py会自动调用该魔术方法。
它通常负责:
打开或获取资源
执行初始化操作
返回as后面要使用的对象
a. 基本语法 1 2 3 def __enter__ (self ): return 返回值
参数:self
返回值:该方法的返回值会赋给as后面的变量。
1 2 3 4 5 with 对象 as 变量: #近似等价于: 变量 = 对象.__enter__()
__exit__()作用:离开with代码块时,py会自动调用该方法。
通常负责:
关闭文件
断开数据库连接
释放锁
清理资源
处理异常
即使with内部发生异常,__exit__()通常也会执行。
a. 基本语法 1 2 3 def __exit__ (self, exc_type, exc_value, traceback ): return False
参数:
self:当前上下文管理器对象
exc_type:异常类型
exc_value:具体的异常对象,也就是错误信息。
traceback:异常的调用栈信息,用于定位错误发生的位置
返回值:
若返回True:表示异常已经处理,不再继续抛出
若返回False或None:表示异常没有处理,继续向外抛出
注:不要随便返回True,否则错误会被静默吞掉,不容易发现问题,一般只在确定自己能够处理某种异常时才返回True
当你写with obj as x时,py会先调用obj.__enter__(),把返回值赋给x,然后执行with代码块,代码块结束后,不管是否会出错,都会调用obj.__exit__()做清理。
比如说下面这个例子,临时修改某个状态,用完自动恢复:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 class TempWorkDir : """临时切换工作目录,离开with块后自动恢复""" def __init__ (self, path ): self .path = path def __enter__ (self ): import os self .old_dir = os.getcwd() os.chdir(self .path) print (f"切换到:{os.getcwd()} " ) return self def __exit__ (self, exc_type, exc_val, traceback ): import os os.chdir(self .old_dir) print (f"恢复到:{os.getcwd()} " ) return False import osimport tempfileprint (f"当前目录:{os.getcwd()} " )with TempWorkDir(tempfile.gettempdir()): print (f"with块内:{os.getcwd()} " )print (f"with块外:{os.getcwd()} " )
3. contextlib.contextmanager 如果每次都要创建一个类来实现enter和exit两个魔术方法的话似乎有些麻烦,于是py的contextlib模块提供了一个装饰器,能够用更简洁的方式来写上下文管理器。
yield的作用是:
返回一个值,并暂停函数;下次继续执行时,从暂停的位置往后运行
只要函数中出现了yield,这个函数就不再是普通函数,而是一个生成器函数。
yield的作用和return类似,只不过return的作用是返回结果并结束函数,而yield的作用是产生一个结果并暂停函数。调用包含yield的生成器函数时会得到一个生成器对象;生成器恢复执行后,可以从上次暂停的位置继续运行。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from contextlib import contextmanagerimport time@contextmanager def timer (label="代码块" ): start = time.perf_counter() print (f"[{label} ] 开始计时···" ) try : yield finally : elapsed = time.perf_counter() - start print (f"[{label} ] 耗时:{elapsed:.4 f} 秒" )with timer("求和" ): total = sum (range (1000000 )) print (f"结果:{total} " )
yield把函数分成两半:yield之前的代码在进入with块时执行,相当于__enter__,yield之后的代码在离开with块时执行,相当于__exit__,这样就不用每次都写一个带enter和exit的类了。
1 2 3 4 5 6 7 8 9 10 11 @contextmanager def manager (): resource = acquire_resource() try : yield resource finally : release_resource(resource)
流程:
1 2 3 4 5 6 7 8 9 10 11 执行 yield 之前的代码 ↓ yield resource ↓ resource 绑定给 as 后面的变量 ↓ 执行 with 代码块 ↓ 恢复生成器 ↓ 执行 yield 之后的代码
再看一个例子,临时修改列表并自动恢复:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from contextlib import contextmanager@contextmanager def temp_append (lst, item ): """临时往列表末尾加一个元素,离开后自动移除""" lst.append(item) print (f"添加了{item} , 当前列表:{lst} " ) try : yield lst finally : lst.pop() print (f"移除了{item} ,当前列表:{lst} " ) data = [1 ,2 ,3 ]print (f"原始列表:{data} " )with temp_append(data, 99 ) as d: print (f"with块内:{d} " )print (f"with块外:{data} " )
注: 这个例子默认with代码块内部不会继续改变列表末尾,否则pop()移除的可能不是最开始添加的item。
整个过程大概是这样的:
首先创建一个列表对象,而data指向这个列表,接着print。
执行with temp_append(data, 99) as d,先计算temp_append(data, 99),因此lst也指向data指向的列表对象,item赋值为99。
因为函数被@contextmanager这个装饰器修饰,所以temp_append(data, 99)被包装成一个可以配合with使用的上下文管理器,进入with时,py开始执行temp_append(),一直运行到yield为止。
接着开始执行yield之前的代码,给data指向的那个列表对象最后加上99。
然后执行yield lst,第一步,函数暂停在这里,暂时不执行后面的finally,第二步,把lst交给as d,即让d指向lst,因此现在d, lst, data指向的都是同一个列表。
接着开始执行with代码块内的代码,即print(f"with块内:{d}"),然后离开with代码块,从yield lst后面的代码开始执行,也就是finally部分的代码。
finally部分执行完之后,此时上下文管理器已经结束,最后执行with代码块后面的代码,即print(f"with块外:{data}")。
contextmanager 装饰的生成器必须正常地只 yield 一次;通常要用 try/finally 保证清理代码一定执行。若 with 中发生异常,异常会在生成器暂停的 yield 位置重新抛入。
注: py传参时,形参会绑定到实参所指向的同一个对象,不会自动复制对象;更准确地说,是对象引用按值传递,这适用于所有对象。
十六、算法题常用技巧 1. ACM 模式 I/O 模板 输入处理 最基础的套路:input() 读一行,split() 按空格切分,map(int, ...) 批量转整数:
1 2 3 4 5 6 7 8 9 10 11 n = int (input ())for i in range (n): a, b = map (int , input ().split()) print (a + b)
如果一行有很多数字需要存到列表里:
1 nums = list (map (int , input ().split()))
sys.stdin 快速读取(大数据量)当输入数据量很大(比如几十万行),input() 可能会超时。这时候需要用 sys.stdin 来加速读取。方法很简单,只需要在代码开头加一行:
1 2 3 4 5 6 7 8 9 import sysinput = sys.stdin.readline n = int (input ())for i in range (n): a, b = map (int , input ().split()) print (a + b)
注: input = sys.stdin.readline是算法竞赛中很常见的简写,但它会覆盖Python内置的input()名称,适合局部刷题代码;一般工程代码中不建议这样覆盖内置名称。
注意: sys.stdin.readline 读取的字符串末尾会带一个换行符 \n,但 split() 和 int() 会自动处理掉,所以大部分情况不需要额外操心。只有在直接拿字符串比较的时候要注意用 strip() 去掉末尾的换行。
多组输入 / EOF 有些题目不告诉你有多少组输入,而是读到文件末尾(EOF)为止。这时候用 try-except 捕获异常。try 块中的代码如果发生ValueError,程序不会崩溃,而是跳到 except 块执行。这里利用这个机制检测输入结束:没有更多输入时,空字符串拆分后无法完成变量拆包,会被 except 捕获并执行 break 退出循环:
1 2 3 4 5 6 7 8 9 10 11 12 13 import sysinput = sys.stdin.readlinewhile True : try : a, b = map (int , input ().split()) print (a + b) except ValueError: break
当 input() 读到 EOF 时,会返回空字符串,split() 得到空列表,map(int, …) 拆包就会抛出ValueError,被 except 捕获后 break 跳出循环。注意,这种写法也会把格式错误导致的ValueError当成输入结束,因此下面直接遍历sys.stdin的写法通常更清晰。
还有一种更简洁的写法,直接用 sys.stdin 遍历每一行:
1 2 3 4 import sysfor line in sys.stdin: a, b = map (int , line.split()) print (a + b)
sys.stdin 本身就是一个可迭代对象,读完所有行自动结束,连 try-except 都不用写。对于按EOF读取多组输入,这通常是更清晰的首选写法。
2. 排序 cmp_to_keya. 适用于必须两两比较才能定义顺序的场景 python中的排序通常使用key参数,大部分排序用 key 就够了,但有些场景需要两两比较才能定义顺序,比如经典的”最大数拼接”问题。这时候需要 cmp_to_key。
cmp_to_key函数就是把自定义的比较函数转化为key函数,便于排序。
一个标准的比较函数通常写成:
1 2 3 4 5 6 7 def cmp (a, b ): if a < b: return 负数 elif a > b: return 正数 else : return 0
返回值:
b. 示例:最大数拼接 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from functools import cmp_to_key nums = [3 , 30 , 34 , 5 , 9 ]def compare (a, b ): if a + b > b + a: return -1 elif a + b < b + a: return 1 else : return 0 str_nums = [str (x) for x in nums] str_nums.sort(key=cmp_to_key(compare)) result = "" .join(str_nums)print (f"最大拼接数:{result} " )
十七、并发与异步编程基础 1. 多线程:threading py的threading模块提供了多线程编程的能力,最基本的用法是创建Thread对象,传入一个目标函数,然后调用start()启动线程。
其中:
1 2 3 4 5 6 7 threading.Thread( target=target, args=args, kwargs=kwargs, name=name, daemon=daemon, )
参数含义:
target:线程要执行的函数;
args:传给目标函数的位置参数;
kwargs:传给目标函数的关键字参数;
name:线程名称;
daemon:是否为守护线程。
表示真正启动一个新线程,在新线程中调用目标函数。
每个Thread对象只能调用一次start()
join()的作用是让当前线程“等待”目标线程执行完毕。如果不调用join(),主线程会直接往下运行;但只要仍有非守护线程存活,整个Python进程通常不会立即退出。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 import threadingimport timedef worker (name, seconds ): print (f"线程 {name} 开始工作" ) time.sleep(seconds) print (f"线程 {name} 完成,耗时 {seconds} 秒" ) t1 = threading.Thread(target=worker, args=("A" , 2 )) t2 = threading.Thread(target=worker, args=("B" , 1 )) start = time.perf_counter() t1.start() t2.start() t1.join() t2.join() end = time.perf_counter()print (f"全部完成,总耗时 {end - start:.1 f} 秒" )
再例如,用多线程模拟并发的网络请求:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 import threadingimport timedef fetch_data (url, delay ): print (f"开始请求{url} ..." ) time.sleep(delay) print (f"请求{url} 完成,耗时{delay} 秒" ) urls = [ ("api/users" , 2 ), ("api/orders" , 3 ), ("api/products" , 1 ), ] start = time.perf_counter()for url, delay in urls: fetch_data(url, delay) sync_time = time.perf_counter() - startprint (f"同步总耗时:{sync_time:.1 f} 秒\n" ) start = time.perf_counter() threads = []for url, delay in urls: t = threading.Thread(target=fetch_data, args=(url, delay)) threads.append(t) t.start()for t in threads: t.join() thread_time = time.perf_counter() - startprint (f"多线程总耗时:{thread_time:.1 f} 秒" )
输出:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 开始请求api/users... 请求api/users完成,耗时2秒 开始请求api/orders... 请求api/orders完成,耗时3秒 开始请求api/products... 请求api/products完成,耗时1秒 同步总耗时:6.0秒 开始请求api/users... 开始请求api/orders... 开始请求api/products... 请求api/products完成,耗时1秒 请求api/users完成,耗时2秒 请求api/orders完成,耗时3秒 多线程总耗时:3.0 秒
2. 线程安全、竞态条件与锁 多线程确实很方便,但当多个线程同时修改同一份数据时,可能会出错,这也叫做竞态条件:
程序结果取决于多个执行单元不可预测的执行顺序。
不能因为某次运行恰好得到正确结果,就认为代码是安全的。
例如,两个线程各自给一个全局变量加1一万次,最终结果理论上应该是两万,但实际按下面这样操作的话,结果不一定会是两万。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 import threading counter = 0 def increment (n ): global counter for _ in range (n): counter += 1 t1 = threading.Thread(target=increment, args=(10000 ,)) t2 = threading.Thread(target=increment, args=(10000 ,)) t1.start() t2.start() t1.join() t2.join()print (f"期望值:20000" )print (f"实际值:{counter} " )print (f"结果正确?{counter == 20000 } " )
你可能发现某次运行结果恰好是正确的20000,但这不能说明代码安全。CPython的GIL(全局解释器锁,后面会详细讲)不保证counter += 1这样的复合操作具有原子性;换一个Python版本、实现或稍微改变操作方式,竞态就可能暴露出来。
因为counter += 1看起来是一步,实际上分为三步:
1 2 3 读取 counter 计算 counter + 1 把新值写回 counter
当两个线程共同执行的时候,可能会发生这样的情况:
1 2 3 4 5 6 7 8 线程 A 读取 counter = 10 线程 B 读取 counter = 10 线程 A 计算出 11 线程 B 计算出 11 线程 A 写入 11 线程 B 写入 11
所以只要涉及多个线程共同访问共享可变数据,就需要使用合适的同步方式,例如锁或Queue。锁保证同一时刻只有一个线程能执行被保护的代码。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 import threading counter = 0 lock = threading.Lock()def increment (n ): global counter for _ in range (n): with lock: counter += 1 t1 = threading.Thread(target=increment, args=(10000 ,)) t2 = threading.Thread(target=increment, args=(10000 ,)) t1.start() t2.start() t1.join() t2.join()print (f"期望值:20000" )print (f"实际值:{counter} " )print (f"结果正确?{counter == 20000 } " )
with lock:会在进入代码块时自动获取锁,离开时自动释放,是实际编程中更推荐的写法。
如果手动管理锁,更准确的写法应使用try/finally保证发生异常时也能释放锁:
1 2 3 4 5 lock.acquire()try : counter += 1 finally : lock.release()
临界区与锁粒度 被锁保护的代码称为临界区。
锁的范围不应该无脑扩大,这样其他线程会长时间等待,基本失去并发意义。
不过,加锁虽然保证了安全,但也降低了性能,线程需要排队等锁,就不能真正并发执行被锁保护的那段代码了,所以锁只应该保护必须互斥访问的最小代码段。
通常应该只保护共享数据操作:
1 2 3 4 result = calculate_something()with lock: shared_results.append(result)
死锁 假设线程A:
1 2 3 with lock_a: with lock_b: ...
线程B:
1 2 3 with lock_b: with lock_a: ...
这样可能就会出现:
1 2 线程 A 拿到了 lock_a,等待 lock_b 线程 B 拿到了 lock_b,等待 lock_a
双方互相等待,永远无法继续,这就是死锁。
线程间数据传递:优先考虑 Queue 比起多个线程共同修改一个列表,更安全的方式常常是使用:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 from queue import Queuedef producer (queue ): for number in range (5 ): queue.put(number) queue.put(None )def consumer (queue ): while True : item = queue.get() try : if item is None : break print ("处理:" , item) finally : queue.task_done()
queue.Queue内部已经实现了必要的同步机制,专门用于在线程之间安全传递任务或数据。
有一句并发编程中的老话很实用:
能传递消息,就尽量不要共享可变状态。
更高层的线程池:ThreadPoolExecutor 前面的threading.Thread适合用来理解线程的创建、启动和等待。如果实际需求只是“把一批独立任务交给线程并发执行”,标准库concurrent.futures提供了更高层的线程池接口:
1 2 3 4 5 6 7 8 9 10 11 from concurrent.futures import ThreadPoolExecutorwith ThreadPoolExecutor(max_workers=3 ) as executor: futures = [ executor.submit(fetch_data, url, delay) for url, delay in urls ] for future in futures: future.result()
ThreadPoolExecutor会复用线程,不需要自己逐个创建和管理Thread对象。学习线程原理时仍然应该先理解Thread、start()和join(),实际批量执行I/O任务时则可以优先考虑线程池。
3. 全局解释器锁(GIL)及其影响 在默认启用GIL的CPython构建中,GIL即全局解释器锁,它通常限制同一进程内同时只有一个线程执行py字节码。Python从3.13开始还提供可选的free-threaded构建,可以禁用GIL,但它并不是传统默认构建。
在默认启用GIL的CPython中,对于纯py的CPU密集型任务(大量计算),多线程通常不会更快,甚至可能更慢。
因为即使你开了多个线程,同一时刻也只有一个线程在执行py代码,其他线程在等着。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 import threadingimport timedef cpu_work (n ): """CPU密集任务:计算累加""" total = 0 for i in range (n): total += i return total N = 5_000_000 start = time.perf_counter() cpu_work(N) cpu_work(N) single_time = time.perf_counter() - startprint (f"单线程耗时:{single_time:.2 f} 秒" ) start = time.perf_counter() t1 = threading.Thread(target=cpu_work, args=(N,)) t2 = threading.Thread(target=cpu_work, args=(N,)) t1.start() t2.start() t1.join() t2.join() multi_time = time.perf_counter() - startprint (f"多线程耗时:{multi_time:.2 f} 秒" )print (f"多线程 vs 单线程:{multi_time / single_time:.2 f} x" )
输出:
1 2 3 单线程耗时:0.26 秒 多线程耗时:0.25 秒 多线程 vs 单线程:0.96x
但是对于I/O密集型任务,如网络请求、文件读写、数据库查询等,多线程依然有效,因为GIL在线程执行I/O操作时会释放,让其他线程有机会执行。
如果你确实需要并行计算CPU密集型任务,可以用multiprocessing模块,它会启动真正的子进程,每个进程有自己的独立的py解释器和GIL,可以充分利用多核CPU。
1 2 3 4 5 6 7 8 9 10 from multiprocessing import Processif __name__ == "__main__" : p1 = Process(target=cpu_work, args=(N,)) p2 = Process(target=cpu_work, args=(N,)) p1.start() p2.start() p1.join() p2.join()
在Windows等使用spawn方式创建子进程的环境中,应当使用if __name__ == "__main__":保护创建进程的代码,避免子进程重新导入主模块时再次创建进程。
更高层的进程池:ProcessPoolExecutor 如果实际需求是把一批彼此独立的CPU任务分发到多个进程,concurrent.futures也提供了更高层的进程池接口:
1 2 3 4 5 from concurrent.futures import ProcessPoolExecutorif __name__ == "__main__" : with ProcessPoolExecutor() as executor: results = list (executor.map (cpu_work, [N, N]))
它和ThreadPoolExecutor使用方式相近,但底层使用多个进程,适合默认启用GIL的CPython中的纯Python CPU密集型任务。
4. 异步编程:asyncio 除了多线程,py还有另一种并发方式:异步编程。典型的asyncio事件循环可以在一个线程内实现并发,靠的是“协程”这个概念。
协程不是线程,而是“可暂停的任务”。一个任务等待时,暂时把执行权交给其他任务。
也叫协作式并发。
协程是一段可以执行到一半暂停,之后再从原位置继续执行的代码。
py用async def定义协程函数 ,用await等待一个可等待对象。await是潜在的暂停点:如果等待结果还没完成,当前协程可以暂停并让事件循环执行其他任务;如果结果已经完成,也可能直接继续运行。
协程对象 :调用协程函数得到的对象。
此时一般还没有正式开始运行。
任务Task :将协程提交给事件循环调度后,可以包装成任务:
1 task = asyncio.create_task(download())
Task可以理解为:已经登记到事件循环中,准备并发执行的协程。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 import asyncioimport timeasync def say_hello (name, delay ): print (f"{name} 开始" ) await asyncio.sleep(delay) print (f"{name} 完成,等了 {delay} 秒" )async def main (): start = time.perf_counter() await say_hello("A" , 2 ) await say_hello("B" , 1 ) end = time.perf_counter() print (f"顺序执行总耗时:{end - start:.1 f} 秒" ) asyncio.run(main())
asyncio.run()是启动异步程序的入口,它会创建一个事件循环并运行传入的协程,等待协程执行完成之后,关闭事件循环。
上面的例子中,await say_hello(“A”, 2) 会等 A 完成再执行 B,所以总耗时还是 3 秒,跟同步没什么区别。
asyncio.create_task() 与 Task1 2 3 4 5 6 7 8 async def main (): task_a = asyncio.create_task(work("A" , 2 )) task_b = asyncio.create_task(work("B" , 1 )) print ("任务已经创建" ) await task_a await task_b
区别是:
只是创建协程对象,不会自动调度。
而:
1 task = asyncio.create_task(coro)
会把协程包装成Task并安排它尽快运行。
官方文档把Task定义为被调度并发运行的协程,并建议保存create_task()返回的引用。
事件循环 事件循环的过程大致如下:
1 2 3 4 5 6 7 8 9 10 11 检查哪些协程现在可以执行 ↓ 选择一个协程运行 ↓ 协程遇到 await,暂停 ↓ 切换到另一个可执行协程 ↓ 某个协程等待的结果完成 ↓ 把对应协程重新放回可执行队列
在一个典型的单线程事件循环中,同一瞬间只有一个任务在执行py代码,任务执行到能够挂起的await时,事件循环才有机会调度其他任务。
使用 asyncio.gather() 并发运行 要并发执行多个协程,可以使用asyncio.gather():
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 import asyncioimport timeasync def fetch_data (url, delay ): print (f"开始请求 {url} ..." ) await asyncio.sleep(delay) print (f"请求 {url} 完成,耗时 {delay} 秒" ) return f"{url} 的数据" async def main (): start = time.perf_counter() results = await asyncio.gather( fetch_data("api/users" , 2 ), fetch_data("api/orders" , 3 ), fetch_data("api/products" , 1 ), ) end = time.perf_counter() print (f"\n并发总耗时:{end - start:.1 f} 秒" ) print (f"返回结果:{results} " ) asyncio.run(main())
asyncio.gather()接收多个协程,并把它们并发调度运行。返回结果时,会按照传入参数的顺序排列。
当一个协程遇到await进入等待时,事件循环会自动切换到其它协程继续执行。对于彼此独立、等待时间可以重叠的任务,总耗时通常接近最慢的那个,而不是所有耗时相加。
asyncio 和多线程有什么区别?在典型的单线程事件循环中,协程在await处协作式地让出执行权;多线程则由操作系统在线程之间进行调度。
但你的代码必须是”异步风格”的,普通函数不能直接变成协程,阻塞操作(比如 time.sleep())也不能直接用在协程里(要用 asyncio.sleep())。
结构化并发:TaskGroup 从py 3.11开始,还可以使用结构化并发:
1 2 3 4 5 6 7 8 import asyncioasync def main (): async with asyncio.TaskGroup() as group: task_a = group.create_task(work("A" , 2 )) task_b = group.create_task(work("B" , 1 )) print ("所有任务都已经结束" )
而退出async with asyncio.TaskGroup()时,会自动等待组内任务。如果其中一个任务异常失败,TaskGroup会取消其他相关任务,并在退出时统一处理异常。相比散落的create_task(),它更容易保证任务不会被遗忘。对于一组生命周期彼此相关的任务,现代asyncio代码通常优先考虑这种结构化并发写法。
5. 并发方案如何选择
I/O 密集型任务 (网络请求、文件读写、数据库查询):asyncio 和 threading 都行。asyncio 在高并发场景下性能更好(单线程没有线程切换开销),但需要用异步风格编写代码。threading 更直观,现有的同步代码改造成本低。
CPU 密集型任务 (大量数值计算、图像处理、加密解密):在默认启用GIL的CPython中,通常使用multiprocessing实现多核并行。threading对纯py CPU密集任务通常没有加速效果;释放GIL的扩展代码或free-threaded构建属于例外。
简单场景 :用 threading 最省心,API 简单直观,几行代码就能用起来。
高并发 I/O 场景 :用 asyncio。比如同时发起上百个网络请求,asyncio 的性能比开上百个线程好得多。
6. asyncio 常见注意事项 异步代码中的阻塞函数 错误示例:
1 2 3 4 5 6 import timeasync def work (): time.sleep(5 )
应该使用:
1 2 3 async def work (): await asyncio.sleep(5 )
同理,在异步函数里直接调用阻塞式网络库、数据库驱动或耗时 CPU 函数,也可能把整个事件循环卡住。官方文档明确警告不要直接在事件循环线程中运行阻塞或 CPU 密集代码。
假如一定要调用同步阻塞函数,那么可以使用:
1 await asyncio.to_thread(sync_function, arg1, arg2)
例如:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 import asyncioimport timedef blocking_work (): time.sleep(2 ) return "完成" async def main (): result = await asyncio.to_thread(blocking_work) print (result) asyncio.run(main())
这会把同步阻塞函数放到另一个线程中执行,避免堵住事件循环。
适合:
暂时无法替换的同步库
阻塞式文件操作
阻塞式网络API
已有同步项目异步迁移
对于纯Python CPU密集计算,在默认CPython下更适合使用进程池,而不是简单地to_thread。
异步代码中的竞态条件 asyncio是单线程,但这并不意味着它没有竞态条件,单线程意味着两个协程不会在任意机器指令中间被操作系统强制同时运行,但只要代码在共享状态操作过程中出现await,其他协程就可以插进来。
例如:
1 2 3 4 5 6 7 8 9 10 import asyncio counter = 0 async def increase (): global counter old_value = counter await asyncio.sleep(0 ) counter = old_value + 1
运行两个任务:
1 2 3 4 5 6 7 8 9 async def main (): await asyncio.gather( increase(), increase(), ) print (counter) asyncio.run(main())
可能过程是:
1 2 3 4 5 6 7 8 任务 A 读取 counter = 0 任务 A 遇到 await 任务 B 读取 counter = 0 任务 B 遇到 await 任务 A 写入 1 任务 B 写入 1
因此异步代码也有asyncio.Lock()
1 2 3 4 5 6 7 lock = asyncio.Lock()async def increase (): global counter async with lock: counter += 1
Python 标准库专门提供了 asyncio.Lock、Event、Semaphore 等同步原语,这本身就说明异步任务也存在共享状态协调需求;这些原语用于协程之间,不用于操作系统线程之间。
async with 与异步上下文管理器普通上下文管理器实现:
异步上下文管理器实现:
1 2 __aenter__() __aexit__()
使用:
1 2 async with resource: ...
适用于进入或退出上下文时本身需要等待的资源,例如:
异步数据库连接
异步HTTP会话
异步锁
TaskGroup
异步文件或网络资源
也可以使用:
1 from contextlib import asynccontextmanager
例如:
1 2 3 4 5 6 7 8 9 10 from contextlib import asynccontextmanager@asynccontextmanager async def database_connection (): connection = await open_connection() try : yield connection finally : await connection.close()
使用:
1 2 async with database_connection() as connection: await connection.query(...)
官方contextlib提供了asynccontextmanager,用于通过异步生成器实现async with。
再看一个例子:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 import asynciofrom functools import wrapsfrom time import perf_counterdef async_timer (func ): @wraps(func ) async def wrapper (*args, **kwargs ): start = perf_counter() try : return await func(*args, **kwargs) finally : elapsed = perf_counter() - start print (f"{func.__name__} 耗时:{elapsed:.3 f} 秒" ) return wrapper@async_timer async def async_work (): await asyncio.sleep(2 ) asyncio.run(async_work())
这段代码实现的是一个异步计时装饰器,给异步函数加上计时功能,并且不改变原异步函数的调用方式。
@wraps(func)是带参数的装饰器,代码:
1 2 3 @wraps(func ) async def wrapper (*args, **kwargs ): ...
等价于:
1 2 3 4 async def wrapper (*args, **kwargs ): ... wrapper = wraps(func)(wrapper)
这里看起来有两次调用,可以拆成:
1 2 decorator = wraps(func) wrapper = decorator(wrapper)
也就是:
wraps(func)根据原函数func创建一个装饰器
再用这个装饰器处理wrapper
把处理后的函数重新赋值给wrapper
为什么是@wraps(func),不是@wraps?因为wraps需要先知道要保留哪一个原函数的信息,就可以把func的名称,文档等信息复制给下面的wrapper。
异步函数的装饰器 假设:
1 2 3 @timer async def async_work (): await asyncio.sleep(2 )
如果timer的包装函数是普通函数:
1 2 def wrapper (*args, **kwargs ): return func(*args, **kwargs)
那么它得到的只是协程对象,并没有等待协程真正执行。因此计时结果可能接近0
异步计时装饰器的实现如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 import asynciofrom functools import wrapsfrom time import perf_counterdef async_timer (func ): @wraps(func ) async def wrapper (*args, **kwargs ): start = perf_counter() try : return await func(*args, **kwargs) finally : elapsed = perf_counter() - start print (f"{func.__name__} 耗时:{elapsed:.3 f} 秒" ) return wrapper@async_timer async def async_work (): await asyncio.sleep(2 ) asyncio.run(async_work())
其中perf_counter()用来精确测量时间间隔:
1 2 3 start = perf_counter() elapsed = perf_counter() - start
这里得到的elapsed就是这段代码经过的实际时间,perf_counter()比time.time()更适合性能计时。
上述过程大致为:
程序加载阶段: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 1. 导入 wraps 和 perf_counter 2. 定义 async_timer 只创建函数,不执行函数体 3. 定义原来的 async_work 4. 发现 @async_timer 5. 执行: async_work = async_timer(原来的 async_work) 6. async_timer 返回 wrapper 7. 现在: async_work → wrapper wrapper 通过闭包记住原函数 func
程序运行阶段: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 执行:asyncio.run(async_work()) 过程是: 1. async_work() 实际调用 wrapper() 创建 wrapper 协程对象 2. asyncio.run() 创建事件循环 3. 事件循环开始运行 wrapper 4. wrapper 记录开始时间 5. 执行 await func() 调用原来的 async_work 6. 原函数执行 await asyncio.sleep(2) 7. 原函数暂停两秒 8. 两秒后恢复原函数 9. 原函数执行结束,返回 None 10. wrapper 准备返回 None 11. 先执行 finally 12. 计算并打印耗时 13. wrapper 返回 None 14. asyncio.run() 结束并关闭事件循环 最终输出类似为: async_work 耗时:2.002 秒
为什么必须使用await:因为调用异步函数只会创建协程对象,不会等待它执行完成,因此,如果直接写成return func(*args, **kwargs)的话,会直接返回协程对象,计时器也无法覆盖原函数真正执行的过程。
写成return await func(*args, **kwargs),就是运行异步函数,并等待它执行完成,再取得它的返回值。
7. 异步迭代器与异步生成器 普通for循环对应同步迭代协议,而某些数据本身需要“等待之后才能产生下一个值”,例如:
流式LLM响应
WebSocket消息
异步数据库游标
分页网络数据流
这个时候会使用async for。
异步生成器 只要在async def中使用yield,就会得到异步生成器函数:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 import asyncioasync def count (): for i in range (3 ): await asyncio.sleep(0.5 ) yield iasync def main (): async for value in count(): print (value) asyncio.run(main())
它可以理解为:
普通生成器负责“按需产生值”,异步生成器则允许在产生下一个值之前先执行异步等待。
异步迭代协议 异步可迭代对象通常实现:
async for会不断等待:
1 await iterator.__anext__()
直到抛出StopAsyncIteration。
正常开发中一般不需要频繁手写__aiter__和__anext__,异步生成器通常更加简洁。
8. asyncio.timeout()(Python 3.11+) 网络请求、Agent工具等操作不能无限等待,可以给一段异步代码设置超时:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 import asyncioasync def slow_work (): await asyncio.sleep(10 )async def main (): try : async with asyncio.timeout(2 ): await slow_work() except TimeoutError: print ("操作超时" ) asyncio.run(main())
在Agent和后端工程中,“超时”本身就是正常的失败模式之一,不应该默认让一个外部调用永远阻塞整个任务。