语句
赋值与名称作用域
LEGB作用域
Python 使用 LEGB 规则查找名称：
L (Local)：当前函数作用域
E (Enclosing)：外层函数作用域
G (Global)：模块全局作用域
B (Built-in)：内置作用域
读取一个名称(NAME)时，Python 按 L → E → G → B 的顺序查找，如果某一内层作用域没有对该名称进行绑定，则继续向外层作用域继续查找，找到后停止
一旦某一内层作用域绑定了该变量，Python将不再往外查找，所有外层作用域都将被遮蔽
x = 10
def f():
print(x) #输出10
f()
2. 一旦函数内也定义了变量，外层作用域将被屏蔽，内层将无法访问
x = 10
def f():
print(x) # 抛出UnboundLocalError，由于内层也定义了x(CPython编译阶段发现)，因此外层的x会被屏蔽，但内层尚未定义该变量就访问(无变量提升)，因此抛错
x = 20
print(x) # 输出20
f()
global/nonlocal
global 和 nonlocal 用于声明名称的绑定作用域，主要用于函数内部的名称绑定
- global：将名称绑定到全局作用域
- nonlocal：用于嵌套函数，声明名称绑定到最近的外层函数作用域，而不是当前函数的局部作用域
x = 10
def f():
x = 20 # 这里是新的局部变量
print(x)
f() # 20
print(x) # 10 全局的x没变
想让函数内变量的作用域绑定到全局，可以使用global
x = 10
def f():
global x
x = 20
f()
print(x) # 20
def f1():
x=10
def f2():
x=100
f2()
print(x) # 输出10，内层的x=100并没有改掉该值
f1()
通过nonlocal将作用范围绑定到函数外层
def f1():
x=10
def f2():
nonlocal x
x=100
f2()
print(x) # 输出100
f1()
注解赋值
注解赋值（Annotated Assignment）是 Python 3.6+ 引入的语法，允许在变量赋值时附加类型注解，其语法为变量名:类型 = 值 ，主要用于为变量、函数参数、返回值提供类型说明，方便IDE工具做类型检查，但注解默认不强制类型，Python 不会因类型不符报错，如注解赋值：x: int = 10，但如果为 x 赋值 x = "hello" 也不会报错，只是类型提示失效
del
del删除语句用于解除名称与对象的绑定，或删除可变容器中的元素。它删除的是"引用"，而不会直接销毁对象本身：
- 删除只会解除这些名称到对应Python对象的绑定，对象本身是否被销毁取决于是否还有其他引用(取决于GC机制)
- 删除变量：如：del a,b,c
- 删除可变容器内的元素，包括列表list、字典dict(按键删除)，集合set需要用remove()或discard()，如：del list[1:3]
- 删除函数名/类名的绑定：如：def func(): pass 删除函数名的绑定 del func
- 删除类的属性：如：class C: x = 10 删除属性 del C.x ，相应地也可以删除实例化对象身上的属性：c=C() del c.x
- 删除模块中的名称：import math 解除模块名绑定 del math
流程控制
Python使用:标记复合语句头的结束，并通过缩进(通常 4 个空格)确定语句块范围，不使用{}
continue、break、pass
- continue：结束当前一轮的循环，进入下一轮循环
- break：结束当前整个循环
- pass：什么都不做，继续执行后面的语句，类似于Bash shell中的空操作符:，常用于占位符，它不止用于循环，还用于函数、类等
if语句
Python中的if语句与Bash shell语法类似：
while语句
for语句
- else语句可选，它会在 while/for 循环正常结束(没有被 break 打断离开)时执行，如：通过循环查找某个数据，但是循环结束依旧没找到，此时可在 else 中进行提醒/善后
- Python 的 if、for、while 本身都不会产生新的局部作用域。因此在这些语句块中创建的名称(i、j等)，依旧指向被绑定的对象，依旧可以在循环/语句块结束后访问
# 查找0-5中的质数
for j in range(0,6):
for i in range(2,j):
if j%i==0:
print(f"{j}不是质数")
break
else:
print(f"{j}是质数")
print(f"i,j会在循环结束时保留值：i={i},j={j}")
match模式匹配
match 是 Python 3.10 引入的一种结构化模式匹配（Structural Pattern Matching）语法。主要用于根据数据的结构、类型和内容进行分支匹配，功能类似于其他语言中的 switch-case，但远比 switch-case 强大，它不仅能匹配值，还能匹配类型、序列结构、字典结构、类对象属性等，其语法格式为：
- 匹配到任意pattern，执行完对应的case语句块后，match将退出，即便后续pattern命中，也不再执行逻辑，不需要显式给定 break
- _会匹配任意内容，通常用作所有模式都未命中时的兜底(类似于switch语句的default)，Python仅允许_用于最后一个case (给定一个变量也可以匹配并捕获任意内容，作用类似，参考下文)
以下是 match 可以捕获的模式（pattern）：
字面量模式
case 后可以指定字面量值，并且可以通过或|指定多个值
def http_status(code):
match code:
case 201 | 200 | 204:
return "Success"
case 401 |400 | 403:
return "Client Error"
case _:
return "Other"
print(http_status(404)) # Not Found捕获模式
case语句可以捕获变量并绑定值，如：point=(0,100) 通过 case (0,y): 由于x坐标都为0，因此匹配成功，且y将被赋值为100。注意如果指定单个变量，该变量会捕获所有值(如：case x: 则x = (0,100))，导致后面所有规则失效，作用等同于_，因此Python不允许匹配单个变量的case语句置于第一个
point = (0,12)
match point:
case (0, 0):
print("原点")
case (x, 0):
print(f"X轴上，x={x}")
case (0, y):
print(f"Y轴上，y={y}") # 匹配该项，且y被赋值12
case (x, y): #可以改写为 case x:同样匹配任意值
print(f"任意点 ({x}, {y})")
AS模式
使用as语句可以将整个匹配对象绑定给某个变量，这在需要拆分元素和保留完整对象时很有用
类型模式
case 可以根据内置类型名匹配数据类型，如：case int() | float()，且可以用 as 重新指定变量名，如：case int() as n
data = "text"
match data:
case int() as n: #可以移除as n，只保留case int()
print(f"整数: {n}")
case str() as s:
print(f"字符串: {s}")
case list() as lst:
print(f"列表，长度={len(lst)}")
case _:
print("其他类型")序列模式
用于匹配具有序列结构的内容(列表、元组、range对象内容)，匹配规则为：通过[ ]指定需要匹配的元素(匹配元组元素也不需要改为())，支持通过_捕获单个匿名元素，通过* 捕获0个或多个剩余元素(捕获为列表)
- [a,b] 将严格匹配有2个元素的序列，并将元素赋值给a,b
- [a,"text"]则严格匹配有两个元素，且第二个元素一定需要为字符串"text"的序列，并将第一个元素赋值给a
- [a, *rest]匹配至少一个元素，*rest收集剩余元素
- [a , _ ,*rest]匹配至少两个元素，第一个元素赋值给a，第二个元素匿名(它无法通过_访问)，*rest收集剩余元素
- [first, *middle, last] 匹配至少两个元素，first、last捕获第一个、倒数第一个元素，*middle捕获剩余元素
list1=[100,"Tom",(1,2)]
list2=[100,"Jay",(3,)]
for li in [list1,list2]:
match li:
case [100,"Tom"]:
print("仅匹配两个元素")
case [100,*a,(3,)]:
print(f"匹配到{li}")
case [100,a,b]:
print(f"a={a},b={b}")
case _:
print("未符合任何匹配模式")
映射模式
映射模式用于匹配字典，特点为：
- 匹配的是字典的键
- match只会匹配字典中是否存在这些键，字典中可以有多余键
- 可以使用**捕获剩余的键
stu={"name":"Jay","age":10,"scores":[10,20,30,40]}
match stu:
case {"job":job}:
print("匹配到已工作")
case {"name":name}: # 命中匹配
print(f"匹配到有name，值为{name}")
case _:
print("未匹配到内容")
可将匹配语句中添加**来捕获剩余的键
case {"name":name,**reset}:
print(f"匹配到有name，值为{name}，其他内容为{reset}")
# 输出为：匹配到有name，值为Jay，其他内容为{'age': 10, 'scores': [10, 20, 30, 40]}
类模式
类模式可以匹配对象及其属性，其中具体位置参数如何映射到属性，由类中的__match_args__决定
class Point:
__match_args__ = ("x", "y")
def __init__(self, x, y):
self.x = x
self.y = y
p = Point(0, 5) # ← 必须实例化，p 是 Point 的实例
match p:
case Point(0, 0):
print("原点")
case Point(x=0, y=y):
print(f"Y轴上 y={y}")
case Point(x, y):
print(f"点 ({x}, {y})")
守卫模式if
通过if语句在 pattern 的基础上进一步做判断
for x in (-10,2,0,10,-2.2):
match x:
case n if n > 0:
print("positive")
case n if n < 0:
print("negative")
case 0:
print("zero")
组合
上述模式可以组合
#综合示例
events = [
{"type": "click", "pos": (100, 650)},
{"type": "key", "code": 52},
{"type": "key", "code": 13},
{"type": "block-dev", "protocol": "USB", "data": b"\x01\x02\x03"},
{"type": "error", "code": 404},
{"type": "unknown"},
]
for event in events:
match event:
# 映射模式 + 序列模式
case {"type": "click", "pos": (x, y)}:
print(f"点击 ({x}, {y})")
# 映射模式 + 类模式 + as 模式
case {"type": "key", "code": int() as code}:
print(f"按键 {code}")
# 映射模式 + 捕获模式
case {"type": "block-dev", "protocol": protocol, "data": data}:
print(f"块设备：{protocol}，数据：{data.hex()}")
# 映射模式 + 字面量 OR 模式
case {"type": "key", "code": 13 | 27}:
print("Enter 或 Escape")
# 捕获模式 + guard
case {"type": "error", "code": code} if 400 <= code < 500:
print(f"客户端错误：{code}")
# 通配符模式
case _:
print("其他事件")
assert断言
assert用于检查程序运行过程中应该成立的条件。如果条件不成立，Python 抛出AssertionError，其语法为：
- assert 条件表达式
- assert 条件表达式, '错误信息'
断言主要用来在调试阶段捕获逻辑错误，做程序内部的正确性检查，该语句可能会被优化掉，Python 以 -O 或 -OO 参数运行时，所有 assert 会被完全移除，因此assert不适用于：
- 处理数据校验、权限检查等安全相关逻辑
- 检查用户输入
- 执行异常处理逻辑
# assert适合做以下工作
# 变量值检查
assert x == 1, "x 应为 1"
# 类型检查
assert isinstance(x, int)
# 检查越界
assert 0 <= x <= 100, f"x 越界: {x}"
# 标记不可达
assert False, "不应执行到这里"
函数
定义函数使用def关键字，后跟函数名与形参列表
- 函数的第一条语句是字符串时，该字符串为函数的文档字符串 docstring，Python shell中开发者可用函数名.__doc__直接查看该部分内容
- 第一行应为对象用途的简短摘要
- 第二行为空白行，将摘要与其余描述分开，后面的行可包含若干段落，描述对象的调用约定、副作用等
- 默认情况下，函数返回 None，可用 return 指定返回值
特点
Python中，函数与整数、字符串、字典等对象一样，是一等公民对象：
- 在运行时创建
- 本身是对象，可以添加、读取属性，如：可以通过函数名.__doc__查看开发者为该函数写好的文档字符串
- 可以赋值、传递、返回和存入容器
- 函数的这些特性使得python中的函数支持高阶函数（指接受函数作为参数，或者可以返回一个函数的函数）、回调、闭包、装饰器等机制
fruits=['strawberry','fig','apple','cherry','raspberry','banana']
def reverse(word):
return word[::-1]
print(sorted(fruits,key=reverse))
#输出为 ['banana', 'apple', 'fig', 'raspberry', 'strawberry', 'cherry']
借助lambda函数可以精简为
sorted(fruits,key=lambda word:word[::-1])
返回值
默认情况下，函数返回 None，通过 return 语句可以返回指定的内容：
- 可以返回各种类型的数据结构，如：return {"name":"Jay","age":10}
- 返回多个值时，返回的是一个元组：return 10,20,30
- 返回函数时，常见应用是闭包
1.返回元组
def f():
return 10,20,30
print(f()) # (10, 20, 30)
2.Python中的闭包
def counter():
count=0
def inc():
nonlocal count
count +=1
return count
return inc
c = counter()
print(c()) #1
print(c()) #2
传参
Python 中传参数全部都是"传对象引用"，没有传值，但这并不意味着在函数内修改int等类型数据，会导致函数外的变量值被修改：
- 可变对象（list、dict、set）：参数与外部 NAME 指向同一个对象，对该对象进行原地修改，调用者可以看到变化。
- 不可变对象（int、str、tuple）：对象不能被原地修改；函数内对参数的重新赋值只是让局部 NAME 绑定到新的对象，不影响外部 NAME
a, b = [], 5
def modify(lst, num):
lst.append(1) # 调用者的 list 会变
num += 1 # 调用者的 int 不变
modify(a, b)
print(a, b) # [1] 5
2.函数内修改全局变量，需要使用global
a=10
def modify():
global a
a+=1
modify()
print(a)
可变序列的默认参数陷阱(面试高频)
def add_item(item, lst=[]): !函数每次执行不会都创建一个空lst： lst=[]
lst.append(item)
return lst
add_item(1) # [1]
add_item(2) # [1, 2]
python中，函数是一个对象，在执行 def 语句定义函数时，python会创建add_item函数对象，此时函数中的默认参数表达式（ lst = [] ）会求值一次，并将结果添加到函数对象中( 可以通过add_item.__defaults__ 查看这些默认参数 )
后续调用该函数，如果未明确指定 lst，则函数默认使用在定义时已经创建好的 lst，而且它不会在每次调用函数时重新求值，因此上述两次add_item() 都将向 add_item 函数定义时创建的 lst 列表中添加元素
正确写法为：
def add_item(item, lst=None):
if lst is None:
lst = []
lst.append(item)
return lst
或者，每次调用函数都显式传入一个空列表：
add_item(1,[])
add_item(2,[])
参数收集
python中可以使用以下符号收集多余的参数：
- *：可以收集所有未被其他参数接收的位置参数，这些参数会被收集到一个元组tuple中，常用变量名为*args
- **：可以收集所有未被其他参数接收的关键字参数，这些参数会被收集到一个字典dict中，常用变量名为**kwargs
def f(a,b,*args,**kwargs):
print(f"普通参数有：{a,b}") #由于a,b间逗号的关系，输出为元组(1,2)
print(f"多余的位置参数：{args}") #元组(3, 4)
print(f"多余的关键字参数:{kwargs}") #字典{'x': 10, 'y': 20}
f(1, 2, 3, 4, x=10, y=20)解包传参
函数定义时，Python会通过 *和** 收集多个参数，将它们集中到一个参数中；函数调用时，Python会解包已有的序列或映射，将其中的元素作为多个参数传入
args = [1, 2, 3]
d = {"a": 1, "b": 2, "c": 3}
def add(a, b, c):
return a + b + c
add(*args) # 解包，等价 add(1, 2, 3)；类似 JS 的 ...args
add(**d) # 解包，等价 add(a=1, b=2, c=3)仅关键字参数与仅位置参数
Python 在定义函数时，可以使用 / 和 * 限制参数的传入方式
- /之前的参数只能通过位置传参，称为仅限位置参数（positional-only），如：def f(a, b, /)，正确传参数f(10, 20)，错误传参： f(a=10, b=20)，
- *之后的参数只能通过关键字传参，可以提升可读性，防参数顺序错，称为仅限关键字参数（keyword-only）如：def f(*, kw, kw2=10)，正确传参：f(kw=1)或f(kw=1, kw2=20)，错误传参：f(1)
- 位于二者之间的参数既可以通过位置，也可以通过关键字传参，称为位置或关键字参数（positional-or-keyword）
def f(pos1, pos2, /, mix, *, kw1, kw2=10):
pass
f(1, 2, 3, kw1=5) # 正确传参数
f(1, 2, mix=3, kw1=5) # 正确传参数
f(1, 2, 3, 5) # TypeError: kw1 必须用关键字
# pos1, pos2: 只能用位置传
# mix: 位置或关键字都行
# kw1, kw2: 只能用关键字传
Lambda 表达式
lambda 参数: 表达式：用于创建匿名函数，Lambda 的函数体只能是一个表达式，不能直接包含普通语句，也不能写类型注解，因此 Lambda 适合简单的一次性函数，不适合复杂逻辑
lambda x: x * 2
等价于：
def func(x):
return x * 2
# 也常用于为排序指定逻辑
numbers.sort(key=lambda x: -x)
装饰器
装饰器（decorator）是一种用于包装函数或类、为其增加功能的机制，它允许开发者在不修改原函数代码的情况下，为函数或类增加额外功能。
装饰器本质上是一个返回函数的高阶函数，它利用了Python 中函数是一等对象的特性，接收一个函数作为参数，并返回一个新的函数（通常是包装后的原函数）
def my_decorator(func):
def wrapper(*args, **kwargs):
print("原函数执行前的额外功能")
func(*args, **kwargs)
print("原函数执行后的额外功能")
return wrapper
@my_decorator
def f(a, b):
print(f"原函数结果：{a+b}")
# 添加装饰器后，执行原函数f会带来my_decorator中添加的额外
print(f(3, 5))保留原函数元信息
使用装饰器后，原函数的__name__、__doc__ 等元信息会丢失，需要通过 functools.wraps 保留
import functools
def my_decorator(func):
@functools.wraps(func) # 关键：保留原函数信息
def wrapper(*args, **kwargs):
"""wrapper 的文档"""
return func(*args, **kwargs)
return wrapper
@my_decorator
def greet():
"""问候函数"""
pass
print(greet.__name__) # greet（没有 wraps 会输出 wrapper）
print(greet.__doc__) # 问候函数带参数的装饰器
装饰器本身也可以接受参数，需要再嵌套一层函数
import functools
def repeat(times):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for _ in range(times):
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@repeat(3)
def say(msg):
print(msg)
say("Hi") #输出三次Hi
生成器函数(yield)
生成器（Generator）是一种惰性求值的迭代器，它不会一次性把所有数据生成并存入内存，而是用的时候才生成，从而节省内存，它可以由两种方式生成：
- 生成器函数
- 生成器表达式，参考表达式章节
Python中只需要在函数中使用yield关键字，函数就会变成生成器函数（而不需要像JS一样需要通过特殊定义方式 function * 定义），此时调用函数不会执行函数体，而是返回一个生成器对象：
- 生成器对象可被迭代，可以通过 for 遍历，被 list() 等方法提取元素
- 执行 yield 会暂停函数执行，保存当前执行状态，返回 yield 后面的值
- 执行next()会从暂停处继续执行
- 生成器只能遍历一次，遍历结束时如果继续执行 next() 则会抛出 StopIteration 异常
- 生成器函数中可以用 return 提前结束，返回值会作为 StopIteration 的参数
def my_gen():
print("开始")
yield 1
print("继续")
yield 2
yield 3
g = my_gen() # 首次执行my_gen()并不真的执行函数(不执行print("开始"))，只返回生成器对象
print(next(g)) # 输出: 开始 \n 1
print(next(g)) # 输出: 继续 \n 2
print(next(g)) # 输出: 3
# print(next(g)) # StopIterationsend()方法
send() 可以向生成器内部传值，同时驱动它向下执行
def echo():
while True:
received = yield
print(f"收到: {received}")
g = echo()
next(g) # 预激活:第一次必须用 next() 或 send(None) 预激活
g.send("hello") # 收到: hello
g.send("world") # 收到: worldyield from委托生成器
yield from 会自动处理子生成器的迭代
def sub_gen():
yield 1
yield 2
def main_gen():
yield from sub_gen() # 委托给子生成器
yield 3
print(list(main_gen())) # [1, 2, 3]经典应用
def read_large_file(path):
with open(path, 'r', encoding='utf-8') as f:
for line in f:
yield line.strip()
for line in read_large_file("big.txt"):
process(line)def producer():
for i in range(10):
yield i
def square(nums):
for n in nums:
yield n * n
def even(nums):
for n in nums:
if n % 2 == 0:
yield n
pipeline = even(square(producer()))
print(list(pipeline)) # [0, 4, 16, 36, 64]输入输出函数
input()
input("提示语句")从标准输入读取一行，返回字符串类型，参数可选，用于添加提示
name = input("请输入姓名: ")
age = int(input("请输入年龄: ")) # 转换为整数
height = float(input("请输入身高: ")) # 转换为浮点数print()
print(*objects, sep=' ', end='\n', file=sys.stdout, flush=False)
- *objects：要输出的对象，可多个
- sep：多个对象之间的分隔符，默认空格
- end：结尾字符，默认换行 \n
- file：输出目标，默认 sys.stdout
- flush：是否立即刷新缓冲区，默认 False
print("Hello")
print("a", "b", "c") # a b c
print("a", "b", sep="-") # a-b
print("Hello", end="") # 不换行
print("内容", file=open("f.txt","w")) # 输出到文件
print("立即输出", flush=True)内置函数
常用
- enumerate(iterable, start=0)：用于遍历可迭代对象时同时获取索引和值，start可选，用于指定索引的起始值，它返回的是迭代器，不是列表，因此只能遍历一次
1.遍历的同时获得索引
fruits = ['apple', 'banana', 'cherry']
for index, fruit in enumerate(fruits):
print(index, fruit)
2. 指定索引从 1 开始计数
for index, fruit in enumerate(fruits, start=1):
print(f"{index}. {fruit}")
3.转换为字典
print(dict(enumerate(fruits)))
# {0: 'apple', 1: 'banana', 2: 'cherry'}
- zip(*iterables, strict=False)：按位置配对多个可迭代对象的元素并返回元组，如果可迭代对象长度不一致，以最短的为准，其他对象中多出的元素会被忽略。strict参数可选，strict = True 时，长度不一致时抛出 ValueError。zip() 返回的是迭代器，只能遍历一次，需要多次使用要转成 list
1. 按位置匹配元素
names = ['Alice', 'Bob', 'Cindy']
scores = [85, 92, 78]
for name, score in zip(names, scores):
print(name, score)
2.转为字典
print(dict(zip(names,scores)))
# {'Alice': 85, 'Bob': 92, 'Cindy': 78}
3. 长度不等时以短的为准
a = [1, 2, 3, 4]
b = ['x', 'y']
print(list(zip(a, b)))
# [(1, 'x'), (2, 'y')] ← 多余的 3、4 被丢弃
4.遍历多个对象
names = ['Alice', 'Bob']
ages = [25, 30]
cities = ['北京', '上海']
for name, age, city in zip(names, ages, cities):
print(f"{name} 今年 {age} 岁，住在 {city}")
5. 也可搭配解包，拆分对象
pairs = [(1, 'a'), (2, 'b'), (3, 'c')]
nums, letters = zip(*pairs)
print(nums) # (1, 2, 3)
print(letters) # ('a', 'b', 'c')
- range(start, stop, step)：创建一个可迭代的 range 对象，它是一个生成器，因此不会立即创建大量数据
- sorted(iterable, *, key=None, reverse=False)：用于对可迭代对象进行排序，返回一个新的已排序列表，不改变原对象：
- key：指定排序依据的函数，默认为 None（按元素本身比较），常用的有：
- key = abs：按绝对值排序
- key = len：按对象元素长度排序
- key=str.lower：忽略大小写排序
- 给key一个 lambda 表达式，根据表达式逻辑排序
- reverse：False 升序（默认），True 降序
- key：指定排序依据的函数，默认为 None（按元素本身比较），常用的有：
1.排序
nums = [3, 1, 4, 1, 5, 9, 2]
print(sorted(nums)) # [1, 1, 2, 3, 4, 5, 9]
print(sorted(nums, reverse=True)) # [9, 5, 4, 3, 2, 1, 1]
print(nums) # [3, 1, 4, 1, 5, 9, 2] 原列表不变
2.用 key 参数指定排序方式
按字典的 age 键排序
students = [
{'name': 'Tom', 'age': 20},
{'name': 'Alice', 'age': 18},
{'name': 'Bob', 'age': 22},
]
result = sorted(students, key=lambda s: s['age'])
# [{'name': 'Alice', 'age': 18}, {'name': 'Tom', 'age': 20}, {'name': 'Bob', 'age': 22}]
3. 按多个字段排序
data = [('Tom', 20), ('Alice', 20), ('Bob', 18)]
# 先按年龄升序，年龄相同再按名字升序
result = sorted(data, key=lambda x: (x[1], x[0]))
# [('Bob', 18), ('Alice', 20), ('Tom', 20)]
- reversed()：返回一个反向迭代器(只能遍历一次)，可以倒序访问可迭代对象中的元素，但不修改原对象，支持的对象有列表、元组、字符串、range（支持 __reversed__() 或 __len__() + __getitem__() 的对象）
nums = [1, 2, 3, 4]
print(list(reversed(nums)))- all(iterable)：用于对可迭代对象中的元素做布尔判断，所有元素都为真，才返回 True，其中False, None, 0, 0.0, "", [], (), {}, set()会被视为假。具有短路求值特性，遇到第一个假值就返回，后面的元素不再计算
- any(iterable)：用于对可迭代对象中的元素做布尔判断，只要有任意一个元素为真，就返回 True。具有短路求值特性，遇到第一个真值就返回
nums = [2, 4, 6, 8]
# 判断是否全为偶数
print(all(n % 2 == 0 for n in nums))
# 是否存在大于 5 的数
print(any(n > 5 for n in nums))iter(iterable)：这是python中处理迭代器的核心内置函数，该函数也可以用来从可迭代对象获取迭代器，高级用法为 iter(callable, sentinel)调用 callable 直到返回 sentinel
next(iterator)：从迭代器中取出下一个元素，高级用法：next(iterator, default)迭代结束时返回 default，不抛StopIteration异常
不断读取输入，直到输入哨兵值 "quit" for line in iter(input, "quit"): print(f"输入了：{line}")map(function, iterable, ...)：对可迭代对象中的每个元素执行相同的操作，返回处理后的结果(一个迭代器)，但大多数情况下，列表推导式可以替代map的功能，因此它适用于已经定义好一个函数，直接套用的情况。
1.对每个元素求平方
nums = [1, 2, 3, 4]
def square(x):
return x ** 2
result = list(map(square, nums))
print(result) # [1, 4, 9, 16]
2.输入多个可迭代对象
a = [1, 2, 3]
b = [10, 20, 30]
result = list(map(lambda x, y: x + y, a, b))
print(result) # [11, 22, 33]
- filter(function, iterable)：根据函数的返回值（True/False）筛选可迭代对象中的元素，保留返回 True 的元素，最终返回一个迭代器，通常也更推荐使用列表推导式
1.保留偶数
def is_even(x):
return x % 2 == 0
nums = [1, 2, 3, 4, 5, 6]
result = list(filter(is_even, nums))
print(result) # [2, 4, 6]
2. 从字典中筛选保留满足指定条件的元素
users = [{'name': 'Tom', 'age': 20}, {'name': 'Jerry', 'age': 15}]
adults = list(filter(lambda u: u['age'] >= 18, users))
数学函数
- abs(x)：取绝对值或复数的模，如：abs(3+4j) 返回 5.0
- divmod(a, b)：返回 (a//b, a%b) 元组，如：divmod(5,3) 返回(1, 2)
- pow(base, exp)：冥运算，pow(base,exp,mod) 可在计算大数时指定模
- round(number, ndigits=None)：四舍六入，但在 .5 时舍入到最近偶数(银行家舍入)，ndigits可选，用于指定保留的小数位round(0.5) # 0 （不是 1） round(1.5) # 2 round(2.5) # 2 （不是 3） round(3.5) # 4 round(4.5) # 4 （不是 5）
- sum(iterable, start=0)：求和，start可选，用于指定起始值
- min(iterable)：返回可迭代对象中的最小元素，用法完全同于max()
- max(iterable)：返回可迭代对象中的最大元素，如：max([10,20,30])；更为强大的用法为max(iterable, *, key=None, default=...)，default用于指定可迭代对象为空时的默认返回值，而不会抛出ValueError，key 参数用于指定 max 比较的内容：
类型构造/类型转换
它们是python中用于构造各类数据的构造器，也用作类型转换：
bool()、int()、float()、complex()、str()、bytes()、bytearray()、list()、tuple()、set()、frozenset()、dict()、memoryview()、object()，具体参考各数据类型的创建。
类
Python中的类：
- 使用class关键字定义类，类名首字母大写
- 实例化类不需要new关键字
- 创建实例由特殊方法__new__()负责，该方法接收类对象，负责创建并返回一个新的实例对象，通常不需要自定义__new__()方法，Python 会使用从父类继承的实现
- 实例创建完成后，Python 会调用__init__()方法对这个实例进行初始化
- __init__()方法中第一个参数通常写作self，表示当前正在被初始化的实例对象
- 可以通过isinstance(obj, Class)判断 实例 是不是某个类（或其子类）的实例，是则返回True，可以使用元组同时指定多个类class Person: def __init__(self, name, age): self.name = name self.age = age person = Person("Tom", 20)初始化过程中，self将指向刚刚创建的 person 实例
属性与方法
class Person:
pass
A = Person # 将类赋值给变量
p = A() #通过变量实例化
2.由于类也是一个对象，因此类中定义的属性和方法，实际上都是类对象身上的属性，实际上可以通过类名直接调用(如果方法需要self，调用时需要传入一个实例)
class Person:
def say(self):
print("hello")
say() 函数将成为 Person 的一个属性：Person.say
通过实例访问该方法时：p = Person(); p.say(); 会通过self将p绑定为say方法的第一个参数，最终执行 Person.say(p)，因此可以直接使用该语句执行方法
3.实例和类都可以拥有各自的属性，通过实例访问某个属性/方法时，如果实例自身没有，则会从类中找。如果实例和类存在同名属性，实例属性会优先于普通的类属性
class Person:
species = "human" #类属性
def __init__(self, name):
self.name = name #实例的属性
self.age = age #实例的属性
访问 p.species 时，如果实例自身没有 species，Python 会继续从类中查找，最终找到 Person.species
4. Python中类的属性是动态的，实例通常可以在运行时直接增加、修改或删除属性，不需要像 Java 一样预先在类中声明实例字段
p.age = 20
p.age = 21
del p.age
5. Python 不提供 Java 式的 private 访问控制，对于不希望外部访问的属性或方法，通常约定在属性、方法名前添加单个_，调用者也约定不手动访问这些属性或方法：
+ 在属性/方法名前添加单个_：此时外部依旧可以访问该属性，如
class Person:
def __init__(self):
self._age = 20
p = Person()
p._age # 外部依旧可以访问到
+ 在属性/方法名前添加两个_：此时Python 会对类定义中的两个前导下划线名称进行名称改写（name mangling），它主要用于避免子类或外部代码意外使用、覆盖同名属性，此时无法通过该名称直接访问
class Person:
def __init__(self):
self.__age = 20 # 会被改写为self._Person__age
p = Person()
p.__age # 只能通过p._Person__age访问
__slots__属性
- 默认情况下，类和实例允许自由添加属性，实例属性默认以字典形式存到 __dict__属性中(内存占用大)
- 可以指定__slots__(可以写为列表、元组等形式，推荐元组)来限制实例只能添加这些属性，同时去掉实例的__dict__，以此实现限制可添加的属性，并节省内存
- 如果父类指定了__slots__，子类不会被限制添加属性，默认会重新生成__dict__，如果也要限制子类的属性，需要依次向所有子类都独立添加该属性
class Person:
__slots__=("name","age")
p = Person()
p.name="Jay" # 允许添加 name
p.scores=10 # 不允许添加该属性
print(p.scores) #AttributeError实例方法、类方法、静态方法
类中的函数并不一定都是实例方法，它们分为：
class Person:
def say(self):
print(self)
2.类方法：第一个参数接收当前类，通常命名为 cls，常用于提供与类本身相关的操作或替代构造方式
@classmethod：
class Person:
@classmethod
def create(cls):
return cls()
Person.create() # 调用
3.静态方法：不自动接收实例或类
@staticmethod：
class Person:
@staticmethod
def add(a, b):
return a + b
Person.add(1, 2)
继承
- Python 使用 class 子类(父类1,父类2..):继承父类，子类可以直接使用父类定义的属性和方法，也可以重写（override）父类方法
- Python 中所有类最终都继承自object
- 可以使用issubclass(子类，父类)检查某个类是否为某个父类的子类，实则返回True
- super()用于在子类中调用父类的方法，如果有多个父类，则选中 MRO 中当前类的下一个类(参考多继承)
class Father:
def __init__(self,name):
self.name=name
class Son(Father):
def init(self, name, score):
super().init(name)
self.score=score
f = Father("f")
s = Son("s",100)
print(issubclass(Son,Father)) #True
多继承
多继承的菱形继承问题
A
/ \
B C
\ /
D- 如果多个父类都定义了同名方法，调用该方法时，Python根据MRO（Method Resolution Order，方法解析顺序）决定从哪个父类继承，MRO使用 C3 线性化算法 计算，核心规则为：
- 子类永远在父类之前
- 多个父类按声明顺序保留
- 每个类只出现一次
- 保持所有父类的局部优先级顺序
- 可以通过__mro__或mro()查看当前MRO
- 子类、子类实例查找一个属性/方法 或执行 super() 时，首先在子类中查找，如果该属性/方法不存在，则沿 MRO 列表从左到右查找，找到第一个定义该方法（或属性）的类即停止，都没找到则抛 AttributeError
# MRO 冲突示例：C 要求 A 在 B 前，D 要求 B 在 A 前，因此产生冲突
class A: pass
class B: pass
class C(A, B): pass
class D(B, A): pass
class E(C, D): pass # TypeError: Cannot create a consistent MRO
# MRO 执行顺序示例
class B(A):
def f(self):
print("B.f")
super().f()
class C(A):
def f(self):
print("C.f")
super().f()
class D(B, C):
def f(self):
print("D.f")
super().f()
print(D.__mro__)
#输出(<class '__main__.D'>, <class '__main__.B'>, <class '__main__.C'>, <class '__main__.A'>, <class 'object'>)
D().f() # 输出：D.f B.f C.f A.f
多态(鸭子类型)
多态可以使同一接口对应不同对象的具体实现，从而让调用方不需要区分对象的具体类型。Python 中主要通过鸭子类型（Duck Typing）实现多态：不关心对象属于什么类型，只关心对象是否具有所需的属性或方法（如果它走起来像鸭子，叫起来像鸭子，那它就是鸭子）
class Duck:
def quack(self):
print("嘎嘎嘎")
class Person:
def quack(self):
print("咕咕嘎嘎")
# 不检查类型，不需要写为let_quack(Duck xx)，只要求有 quack 方法
def let_quack(thing):
thing.quack()
let_quack(Duck()) # 嘎嘎嘎
let_quack(Person()) # 咕咕嘎嘎
魔术方法
魔术方法（Magic Methods）又叫双下划线方法（dunder methods）（double underscore methods），是 Python 中以双下划线__xxx__命名的特殊方法，用于让自定义类参与 Python 的内置语法和操作，可以让用户自定义的类能像内置类型一样"自然地"参与运算、迭代、打印等。它们通常不直接调用，而是由 Python 在执行相应操作时自动调用。
以下是自定义类可能用到的魔术方法，以及它们代表的python操作：
| 方法 | python操作 | 用途 |
|---|---|---|
| __new__(cls, ...) | 创建实例时最先调用 | 返回实例，极少重写 |
| __init__(self, ...) | 实例创建后初始化 | 最常用 |
| __del__(self) | 对象被垃圾回收前 | 不保证及时执行，慎用 |
| __str__ | str(obj) / print(obj) | 字符串表示 |
| __repr__ | repr(obj) | 面向开发者的表示 |
| __eq__ | obj == x | 相等比较 |
| __lt__ | obj < x | 大小比较 |
| __len__ | len(obj) | 长度 |
| __getitem__ | obj[key] | 索引、取元素 |
| __setitem__ | obj[key] = x | 设置元素 |
| __delitem__ | del obj[key] | 删除元素 |
| __contains__ | x in obj | 成员测试 |
| __iter__ | for x in obj | 迭代 |
| __next__ | next(obj) | 获取下一个元素 |
| __add__ | a + b | 加法 |
| __sub__ | a - b | 减法 |
| __mul__ | a * b | 乘法 |
| __bool__ | bool(obj) | 真值判断 |
| __hash__ | hash(obj) | 哈希值 |
| __enter__ / __exit__ | with | 用于上下文管理 |
| __call__ | obj() | 将对象转为可调用对象 |
| __getattr__ | 属性不存在时的查找 | |
| __setattr__ | 属性设置 | |
| __delattr__ | 属性删除 |
异常处理
try..except语句
程序运行过程中发生错误时，Python 会产生（raise）一个异常对象
- except 语句：
- 可以有多个except语句，针对不同的异常做不同的处理
- 也一次捕获多个异常，如：except (ValueError, TypeError)
- Python所有的错误都是从BaseException类派生的，包含 SystemExit、KeyboardInterrupt 等不应通常被业务代码捕获的异常，参考：python内置异常
- Exception是大多数普通运行时异常的基类，不写异常类型时可以通过它来捕获大多数异常：except Exception，但通常不推荐，因为容易掩盖真正的程序错误
- finally 中的代码无论是否发生异常，通常都会执行，常用于资源清理，例如关闭文件、释放锁等
def divide_and_check(a, b):
"""
异常处理流程
"""
try:
# 可能触发异常的语句
result = a / b
# 手动触发 ValueError（结果不能为负数）
if result < 0:
raise ValueError(f"结果 {result} 为负数，不允许")
print(f" try 块执行完毕，result = {result}")
except ZeroDivisionError as zde:
# 捕获特定类型异常
print(f" [ZeroDivisionError] 除数不能为 0: {zde}")
except (ValueError, TypeError) as vte:
# 多个异常合并捕获，通过 as 赋值给变量
print(f" [ValueError/TypeError] 参数或类型有问题: {vte}")
except Exception as e:
# 兜底：捕获当前正在处理的异常，并 raise 主动再触发
print(f" [兜底捕获] 未知异常: {type(e).__name__}: {e}")
# 主动重新抛出当前异常，交给上层处理
raise
else:
# 没有异常时执行
print(f" [else] 一切正常，最终结果 = {result}")
return result
finally:
# 无论是否发生异常都会执行
print(" [finally] 清理资源 / 收尾工作")
raise
raise 用于主动产生异常，语法为：raise 异常类型，当它已经位于某个异常处理块中时，可以省略异常类型来再次触发该异常以向上汇报(参考上文代码块)
异常链
在处理一个异常时，如果又产生了另一个异常，Python 会记录两者之间的因果关系，可以使用raise ... from xxx显式指定原因
try:
x = int("abc")
except ValueError as e:
raise RuntimeError("转换失败") from e
常用异常类型
| 异常 | 声明 |
|---|---|
| Exception | 所有非系统退出异常的基类 |
| BaseException | 所有异常的基类 |
| ValueError | 值不合适 |
| TypeError | 类型不匹配 |
| KeyError | 字典键不存在 |
| IndexError | 索引越界 |
| AttributeError | 属性不存在 |
| NameError | 变量名未定义 |
| FileNotFoundError | 文件不存在 |
| IOError / OSError | I/O 错误 |
| ZeroDivisionError | 除以零 |
| ImportError | 导入失败 |
| StopIteration | 迭代器结束 |
| KeyboardInterrupt | 用户中断（Ctrl+C） |
| SystemExit | sys.exit() 触发 |
并发编程
Python中常见的并发方式主要有线程（thread）、进程（process）和异步 I/O（asyncio）：
- 线程（Thread）：同一进程内执行，共享内存，受 GIL 限制，适合 I/O 密集型任务
- 进程（Process）：独立内存空间，不受 GIL 限制，适合 CPU 密集型任务
- 异步任务（Asyncio）通过协程（Coroutine）与事件循环（Event Loop）实现并发，多个协程在单线程事件循环中交替执行，当协程函数执行到 await 时，如果 await 的对象需要等待网络、文件等 I/O ，主动让出执行权，由事件循环调度其他协程继续执行，从而在不创建大量线程的情况下实现高并发 I/O，适合大量 I/O 密集型任务
GIL全局解释器锁
GIL全局解释器锁（Global Interpreter Lock）是 CPython 解释器中的一把全局互斥锁，用于确保同一时刻只有一个线程执行 Python 字节码
CPython 的内存管理、引用计数等设计和实现不是线程安全的，为了保护 Python 对象不被多线程同时修改，又不增加解释器实现的复杂度，Cpython 用一把大锁简化实现，保证解释器同一时间只允许一个线程执行 Python 字节码，由此带来 CPython 的特点：
- 多线程并发：由于GIL的存在，多个线程不能同时执行 Python 字节码，因此 CPU 密集型任务通常无法通过增加线程获得多核并行加速。但当线程执行网络请求、文件 I/O、数据库操作等需要等待外部资源的操作时，CPython 通常会释放 GIL，让其他线程获得执行机会，因此多线程不适合 CPU 密集型任务，但依旧适合I/O 密集型任务
- 多进程：每个进程拥有自己独立的解释器和 GIL，不同进程之间的 GIL 互不影响。因此，多个进程可以分别在不同的 CPU 核心上执行 Python 字节码，实现真正的多核并行
- GIL 是 CPython 的实现细节，不是 Python 语言规范的要求。不同解释器可能采用不同的实现方式，例如 Jython 不使用 GIL。此外，CPython 从 Python 3.13 开始提供可选的 free-threaded（自由线程、无 GIL）构建，允许多个线程并行执行 Python 字节码。不过，无 GIL 构建仍需考虑线程安全、扩展模块兼容性及实际运行开销，不能简单地认为所有多线程程序都会因此获得性能提升
import threading
# CPU 密集型：多线程几乎无加速
def count():
i = 0
for _ in range(10_000_000):
i += 1
# 两个线程串行执行，因为 GIL 互斥
t1 = threading.Thread(target=count)
t2 = threading.Thread(target=count)
t1.start(); t2.start()
t1.join(); t2.join()threading线程
threading 是 Python 标准库中的线程模块，只导入Thread类：
from threading import Thread如果要使用模块函数等其他内容，应该完整导入：
import threading创建线程
threading.Thread(target=None..., args=(), kwargs={}..) 创建线程，而后通过 .start() 启动线程，通过 .join() 等待线程结束
- target：线程执行的可调用对象
- args：位置参数元组
- kwargs：关键字参数字典
- name：线程名称
- daemon：是否设置为守护线程，守护线程适合后台辅助任务
from threading import Thread
def func():
for i in range(50):
print("func",i)
if __name__=="__main__":
t = Thread(target=func)
t.start()
for i in range(50):
print("main",i)
#主线程与t线程交替执行，导致输出为：
main 0func 0
main 1
func 1
main 2func
main2
3func
常用属性
- t.name线程名，可读写
- t.daemon是否为守护线程（主线程退出时自动结束）
- t.setDaemon(bool)设置守护线程需在 start() 前，也可以 t.daemon = True
- t.ident线程标识符（整数）
- t.native_id操作系统原生线程 ID
常用方法
- t.start()启动线程，自动调用 run()
- t.run()线程实际执行体，可重写，如果手动调用该方法，则直接在当前线程执行任务
- t.join(timeout=None)等待线程结束，可设等待的超时时间
- t.is_alive()判断线程是否存活
模块函数
需要完整导入模块import threading
- threading.active_count()返回当前存活线程数
- threading.current_thread()返回当前线程对象
- threading.enumerate()返回所有存活线程列表
- threading.main_thread()返回主线程对象
- threading.get_ident()返回当前线程标识符
- threading.get_native_id()返回当前线程原生 ID
multiprocessing进程
python 中，每个进程拥有独立的 Python 解释器，因此可以利用多个 CPU 核心并行执行 Python 代码，python的进程并发能力由multiprocessing模块提供
import multiprocessing创建进程
multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None) 创建进程，而后通过 .start() 启动进程，通过.join() 等待进程结束，API与线程很接近
- target：进程执行的可调用对象
- args：位置参数元组
- kwargs：关键字参数字典
- name：进程名称
- daemon：是否设置为守护进程
if __name__ == "__main__":
...
这是因为，Windows中，子进程需要重新导入主模块，如果创建进程的代码位于模块顶层，可能导致递归创建进程
from multiprocessing import Process
def work():
print("work")
if __name__ == "__main__":
p = Process(target=work)
p.start()
p.join()
常用属性
- p.name进程名，可读写
- p.daemon是否为守护进程
- p.pid 进程PID
常用方法
- p.start()启动进程，自动调用 run()
- p.run()进程执行体，可重写
- p.join(timeout=None)等待进程结束，可设等待的超时时间
- p.is_alive()判断进程是否存活
- terminate()发送 SIGTERM 终止进程
- kill()发送 SIGKILL 终止进程（Python 3.7+）
- close()关闭进程对象，释放资源
模块函数
- multiprocessing.current_process()返回当前进程对象
- multiprocessing.parent_process()返回父进程对象
- multiprocessing.active_children()返回活动子进程列表
- multiprocessing.cpu_count()返回 CPU 核心数
- multiprocessing.freeze_support()Windows 打包 exe 时需调用
异步
Python 的异步编程主要通过协程（coroutine）和事件循环（event loop）实现，执行逻辑类似于JS中的事件循环与异步队列处理，核心思想为：任务等待 I/O 时，不阻塞整个程序，而是将执行权交给事件循环，继续其他已经就绪的任务，主要处理大量 I/O 密集型、网络请求等任务
协程
协程（Coroutine）是一种用户态的轻量级执行单元，可以在执行过程中主动暂停，并在之后恢复执行。Python 的协程通常运行在单个线程、单个进程的事件循环中。它不会创建多个线程，而是当协程函数 await 的对象需要等待网络、文件等各类资源时，协程自己主动让出执行权，由事件循环协作调度其他就绪的协程，其模型类似于：
进程
└── 线程
└── EventLoop
├── 协程 A
├── 协程 B
└── 协程 C
代码示例：
import asyncio
async def f1():
print("f1")
await asyncio.sleep(1) # 模拟异步 I/O
print("f1 Done")
async def f2():
print("f2")
await asyncio.sleep(1) # 模拟异步 I/O
print("f2 Done")
async def main():
await asyncio.gather(f1(),f2())
if __name__=="__main__":
print('1')
asyncio.run(main())
print('2')
以上模块：
+首先执行 print('1')
+执行 main() 函数时，两个协程f1、f2执行流程如下：
当f1遇到需要等待的事件 asyncio.sleep(1)，f1暂停执行，主动让出执行权，执行权交还事件循环
开始执行f2，f2在执行完print("f2")后同样需要等待事件，f2暂停执行，执行权交还事件循环
main() 函数内事件循环只有f1,f2，因此再次调度被暂停的f1执行，然后执行f2
+main()执行完毕返回，然后执行 print('2')
+最外层的语句 print('1') ; asyncio.run(main()) ; print('2') 属于同步代码，asyncio.run(main())调用执行完毕并返回后才会执行print('2')，print('2')不会因为main()中的协程都让出执行权就被先执行，它们分属不同作用域的内的事件循环
#输出
1
f1
f2
f1 Done
f2 Done
2
同步与异步执行比较
import time
def task1():
print("任务1开始执行")
time.sleep(5)
print("任务1执行完毕")
return 10
def task2():
print("任务2开始执行")
time.sleep(3)
print("任务2执行完毕")
return 20
def main():
t1 = task1()
t2 = task2()
print("t1结果：",t1)
print("t2结果：",t2)
if __name__ =="__main__":
start = time.time()
main()
print("本次共用时：",time.time()-start)
任务1开始执行
任务1执行完毕
任务2开始执行
任务2执行完毕
t1结果： 10
t2结果： 20
本次共用时： 8.002832174301147
import asyncio
import time
#通过 async 将 task1 交给事件循环管理调度
async def task1():
print("任务1开始执行")
await asyncio.sleep(5) #通知事件循环该位置可以挂起
print("任务1执行完毕")
return 10
async def task2():
print("任务2开始执行")
await asyncio.sleep(3)
print("任务2执行完毕")
return 20
async def main():
result = await asyncio.gather(task1(),task2())
print(result)
if __name__ =="__main__":
start = time.time()
asyncio.run(main())
print("本次共用时：",time.time()-start)
任务1开始执行
任务2开始执行
任务2执行完毕
任务1执行完毕
[10, 20]
本次共用时： 5.01055383682251
asyncio库
asyncio是 Python 3.4 引入的标准库，用于创建协程，编写单线程并发程序，为异步 I/O 提供支持，其核心是事件循环（Event Loop），事件循环负责管理和调度协程、Task 以及 I/O 等待事件，其核心语句包括：
- async def：将一个函数定义为coroutine类型，调用协程函数时，不会立即执行函数体，而是返回一个协程对象，使用 await 或 asyncio.run() 运行
- await：用于等待一个可等待对象（awaitable） ，最常见的是另一个协程，如：await fetch()，await 只能出现在 async def 定义的协程函数中
- asyncio.run()创建事件循环并运行协程，是程序主入口
import asyncio
async def hello():
print("Hello")
await asyncio.sleep(1) # 模拟异步 I/O
print("World")
asyncio.run(hello()) # 运行协程函数
async
async关键字主要用于定义异步函数/协程函数(coroutine function)，它提供了暂停和恢复执行的机制，使程序在等待网络响应、异步 I/O 等操作时，可以将执行权交还给事件循环，让其他就绪的任务运行，注意：
- 直接调用协程函数，并不会立即执行函数体，而是返回一个协程对象
- 协程对象由事件循环调度执行，通常使用使用 await 或 asyncio.run() 运行
import asyncio
async def f():
print("函数执行")
def main():
#直接执行协程函数，并不会立即执行函数体
coro = f()
#f()返回的是协程对象：<coroutine object f at 0x00000120A6A61F00>，此时函数体尚未开始执行
print(coro)
#协程需要由事件循环调度执行
#创建事件循环，并将 coro 协程对象注册到事件循环中，由事件循环调度运行
asyncio.run(coro) #可直接写为 asyncio.run(f())，传入的需要是调用 f() 所产生的协程对象，而不是函数 f 本身
if __name__ == "__main__":
main()await
await关键字主要用于 async 定义的协程函数内部，用于等待一个可等待对象（awaitable）完成，并获取其执行结果，可等待对象主要包括协程对象、asyncio.Task、asyncio.Future，通常不能是普通函数，但如果普通函数返回的是一个可等待对象，那么等待该返回值是合法的，注意：
- await 并不一定引起任务切换
- await 协程函数时：会继续执行函数，事件循环并不会直接切换到其他任务中执行(除非协程函数内部有挂起操作)，因此await 协程函数不会自动创建新的 Task，也不保证发生任务切换
- await Future 对象时：如果 Future 尚未完成，当前 Task 通常会挂起，事件循环便可以调度其他就绪任务；如果 Future 已经完成，则可能直接取得结果，不发生任务切换
import asyncio
async def f2():
print("f2执行")
# 由于sub_f通过await调用，因此需要将其声明为async
async def sub_f():
print("子函数执行")
return 100
async def f1():
print("f1执行")
#并不会切换到执行f2，而是继续执行 sub_f()，并完整执行完f1再执行f2
subF_result = await sub_f()
print(subF_result)
async def main():
await asyncio.gather(f1(),f2())
if __name__ == "__main__":
asyncio.run(main())
#输出结果为:
#f1执行
#子函数执行
#100
# f2执行
asyncio.Future
asyncio.Future是一种底层可等待对象（awaitable），用于表示异步操作未来才能确定的结果，它可以保存操作的返回值、异常或取消状态，并允许协程通过 await 等待结果，Future 本身不负责执行协程，主要用于连接基于回调的异步操作与 async/await 协程机制，比如影响执行await时是否应该切换任务，通常也不需要在业务代码中手动创建 Future，以下例子用于展示事件循环
- asyncio.gather(task1(), task2()) 并发安排两个协程执行。
task1() 开始运行，调用并等待 sub_task()。sub_task() 创建 Future，并提交 thread_task() 到另一个线程中执行（由线程池创建维护该独立的线程） - 执行到 result = await future 时，Future 尚未完成。
sub_task() 对应的 Task 暂停推进协程，等待 Future 完成。由于 task1() 正在等待 sub_task()，它也不能继续执行后续代码 - task2() 开始执行，运行到 await asyncio.sleep(1) 后挂起，大约 1 秒后，task2() 恢复，打印 task2执行完毕 并返回 200
- 大约 3 秒后，工作线程调用 loop.call_soon_threadsafe()，通知事件循环执行 future.set_result(100)。
Future 转为完成状态，并安排等待它的 Task 恢复。事件循环再次推进 sub_task()，使 await future 返回 100
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor
def thread_task(future):
time.sleep(3)
future.set_result(100) #模拟3s后future完成并获得结果
async def sub_task():
print("sub_task开始")
#创建 future 对象
event_loop=asyncio.get_event_loop()
future=event_loop.create_future()
# 通过另一个线程执行任务
event_loop.run_in_executor(ThreadPoolExecutor(),thread_task,future)
#执行到该语句时，事件循环将调度执行task2
result = await future
print("sub_task执行结束")
return result
async def task1():
print("task1执行")
result = await sub_task()
print("task1结束")
return result
async def task2():
print("task2执行")
await asyncio.sleep(1)
print("task2执行完毕")
return 200
async def main():
result = await asyncio.gather(task1(),task2())
print(result)
if __name__ == "__main__":
asyncio.run(main())
# 输出为：
#task1执行
#sub_task开始
#task2执行
#task2执行完毕
#sub_task执行结束
#task1结束
#[100, 200]asyncio.Task
asyncio.Task继承自 Future，用于将协程包装为一个可由事件循环调度的任务，负责推进协程执行，并在协程挂起后等待相应操作完成，再恢复协程。如前所述，调用协程函数只会得到协程对象，不会自动将其安排为独立的并发任务，可以使用 asyncio.create_task(coro) 将协程对象创建为 Task，也可以将协程对象传入 asyncio.gather()、asyncio.as_completed() 等接口，由它们负责将协程包装为任务并交由事件循环调度，Task在一个事件循环中的典型执行流程如下：
- 通过 asyncio.create_task(coro)、asyncio.gather等方法将协程包装为 Task，并安排执行
- 事件循环调度 Task，Task 推进协程，直到协程返回、抛出异常或发生挂起
- 如果推进协程过程中遇到 await，如果等待对象能够立即完成，协程可能直接继续执行而不切换任务；如果需要等待尚未完成的 Future，Task 通常会挂起，事件循环可以调度其他就绪 Task、处理 I/O 事件或执行定时器
- 恢复执行：等待的 Future 完成后，相关回调安排 Task 恢复，Task 继续推进原协程
- 协程正常返回时，Task 保存返回值；协程异常结束时，Task 保存异常；如果任务被取消，则进入取消完成状态
注意：
- 创建 Task 后通常需要手动指定 await 来等待 Task 完成的结果，如：使用 await task，或者使用 await asyncio.gather(...) 等待一组任务
- asyncio.create_task() 需要在运行中的事件循环内调用；普通脚本通常通过 asyncio.run(main()) 启动事件循环。
- 取消 Task 是发出取消请求，并不保证协程立刻停止。协程应正确处理 CancelledError，并在必要时完成资源清理
创建任务
- asyncio.create_task(coro)：将一个协程对象包装成 Task 并开始参与事件循环的执行调度
- asyncio.wait(tasks,timeout)等待任务完成，返回done(已完成的任务)、pending(未完成的任务)，如：done, pending = await asyncio.wait(tasks)，支持参数：timeout：指定超时时间，如：await asyncio.wait(tasks,timeout=3)
import asyncio
async def task(a,num):
return ord(a)+num
async def main():
t = asyncio.create_task(task("A", 1))
result = await t
print(result)
if __name__ =="__main__":
asyncio.run(main())
常用方法
- task.result()获取任务结果
- task.done()判断任务是否已经完成
- task.cancel()取消任务
- task.cancelled()判断任务是否被取消
- asyncio.current_task()获取当前事件循环中正在执行的 Task
- asyncio.all_tasks()获取指定事件循环中尚未完成的 Task 集合
事件循环
- asyncio.run()创建事件循环，也是推荐的创建循环方式，它会自动创建事件循环、运行协程、最后关闭循环而不需要手动维护，通常用于异步任务的程序主入口
- asyncio.get_running_loop()获取当前正在运行的事件循环
- 手动维护事件循环：
import asyncio
async def task(name, delay):
print(f"{name} 开始")
await asyncio.sleep(delay)
print(f"{name} 结束")
return name
#创建事件循环
loop = asyncio.new_event_loop()
#设置为当前线程的事件循环
asyncio.set_event_loop(loop)
try:
# 运行协程直到完成
result = loop.run_until_complete(task("A", 1))
print("返回值:", result)
# 可以继续在同一个循环中运行其他任务
loop.run_until_complete(task("B", 0.5))
finally:
# 关闭循环
loop.close()
并发执行
asyncio.gather
asyncio.gather(对象1,对象2..,return_exceptions=False)：并发运行多个操作，返回一个聚合的Future，按输入顺序汇总结果，其特点为：
- 接收参数为多个可等待对象(协程、Task、Future 等)，如果是协程，gather 会自动用 ensure_future 包装成 Task 参与调度
- 如果所有可等待对象都成功完成，结果为一个由所有返回值聚合而成的列表，结果顺序按输入顺序排序
- 等待所有操作完成：默认情况下，成功时会等待所有可等待对象完成后再返回结果，因此总等待时间通常受最慢的操作影响。各操作是并发执行的，不是依次等待执行
- return_exceptions=False（默认）：如果某个操作抛出异常，gather() 会立即将该异常传播给调用方，但其他其它任务不会被取消，通常仍会继续执行。异常操作的返回值无法取得；其他操作的结果也不会由这次 gather() 正常汇总返回
- return_exceptions=True：操作抛出的异常会作为对应位置的结果返回，而不是立即向调用方传播。gather() 会等待所有操作完成，返回值列表中既包含正常结果，也可能包含异常对象
import asyncio
async def task(name, sec):
await asyncio.sleep(sec)
return f"{name} done"
async def main():
#C 最快完成，但结果顺序仍是 A, B, C
results = await asyncio.gather(
task("A", 3),
task("B", 2),
task("C", 1),
)
# 等待最长任务 3s 后才会获得结果
print(results) # ['A done', 'B done', 'C done']
if __name__=="__main__":
asyncio.run(main())asyncio.as_completed
asyncio.as_completed(可迭代对象,timeout=None)：并发地运行可迭代对象中的可等待对象，特点为
- 接收参数为可等待对象组成的可迭代对象，返回的对象也可以被迭代，以获取可等待对象结束时的结果
- 谁先完成先处理：按照操作的完成先后顺序处理结果，适用于希望尽早处理已完成请求的场景
- timeout：指定等待完成操作的最长时间。超时后，迭代会引发 TimeoutError。超时本身不会自动取消所有尚未完成的操作；如果需要停止它们，应显式取消并处理相关任务
for coro in asyncio.as_completed([task("A",1), task("B",2)]):
print(await coro)超时控制与sleep
- asyncio.sleep(seconds)异步等待时间，相较于普通的time.sleep(1)，它不会阻塞线程
- asyncio.timeout(seconds)Python 3.11+ 提供的更现代的超时控制方法，超时时会抛出TimeoutError，推荐使用with管理
- asyncio.wait_for()：旧版本用的超时取消任务方法
异步同步控制
- asyncio.Lock()异步锁，在协程之间使用，不会阻塞线程
sem = asyncio.Semaphore(3) # 最多允许 3 个协程同时进入
async def worker(i):
async with sem: # 获取信号量，超出则等待
print(f"task {i} start")
await asyncio.sleep(1)
print(f"task {i} done")
async def main():
await asyncio.gather(*(worker(i) for i in range(10)))
asyncio.run(main())
- asyncio.Queue()异步队列，用于生产者-消费者模型queue = asyncio.Queue() await queue.put(item) item = await queue.get()
模板
# 发起网络请求/爬虫
async def fetch(url):
...
async def main():
tasks = [
asyncio.create_task(fetch(url))
for url in urls
]
results = await asyncio.gather(*tasks)
asyncio.run(main())资源管理
资源管理主要解决文件、网络连接、锁、数据库连接等资源使用完后，或者使用过程中发生异常时的可靠释放问题
Python 使用with语句实现资源的自动管理。它建立在上下文管理器（context manager）机制之上
with与上下文管理器
Python 提供了上下文管理器协议，通过with语句自动管理资源的释放，如：with open("data.txt", "r") as f: 在离开 with 块时，无论是否异常，f 都会自动关闭
能够用于 with 的对象称为上下文管理器，它需要提供__enter__()和__exit__()两个魔术方法
class MyResource:
def __enter__(self):
# 获取资源，返回值赋给 as 后的变量
print("获取资源")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
# 释放资源，无论是否异常都会执行
print("释放资源")
# 返回 True 表示异常已被处理，不再向外传播
# 返回 False 或 None 则异常继续传播
return False
# 使用资源
with MyResource() as r:
print("使用资源")
通常不需要自己编写 __enter__()和__exit__()，标准库 contextlib 提供了更方便的工具，@contextmanager可以把一个生成器函数转换为上下文管理器
from contextlib import contextmanager
@contextmanager
def resource():
print("获取资源")
try:
yield # yield 的值赋给 as 后的变量
finally:
print("释放资源") # 确保异常时也能释放
with resource():
...内置上下文管理器
python已经内置一些常用的上下文管理器
- 文件：with open(...) as f:
- 线程锁：with lock:
- 数据库连接：with conn: / with conn.cursor() as cur:
- 临时目录：with tempfile.TemporaryDirectory() as d:
- 网络连接：with socket.socket() as s:
文件读写
open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True, opener=None) 打开文件，返回一个 file object
- file：文件路径（字符串或路径对象），该参数必须，其他参数可选
- mode：打开模式，默认为'r'(只读)，参考下文
- buffering：指定缓冲策略， 0 表示关闭缓冲（仅允许在二进制模式下使用）， 1 表示选择行缓冲（仅在文本模式下使用），整数 > 1 表示固定大小块缓冲区的字节大小
- encoding：编码方式，默认使用系统默认编码方式，国内系统可能为GBK，文件模式建议显式指定，如：'utf-8'
- errors：指定编码、解码错误时的处理方式，值有：'strict'抛出错误，'ignore'忽略错误，'replace'会在数据出现错误的地方插入替换标记（如 “？”），其他值参考文档
- newline：控制换行符处理（None、''、'\n'(Unix默认)、'\r'、'\r\n'(Windows默认)）
- closefd：如果该值为 False，且给出的是文件描述符而非文件名，则在文件关闭时，底层文件描述符将保持打开。如果给定文件名，closefd 必须为 True（默认）;否则将出现错误
文件打开模式：
- 对已有内容的文件，任何情况下谨慎使用 w
| mode | 含义 | 说明 |
|---|---|---|
| 'r' | 只读(默认) | 打开读取，指针在文件开头，文件必须存在 |
| 'w' | 只写 | 打开写入，会先清空文件！，指针在文件开头，不存在则创建 |
| 'a' | 追加 | 打开追加，指针在末尾，不清空，不存在则创建 |
| 'x' | 独占创建 | 文件已存在则报错 FileExistsError |
| 't' | 打开为文本文件(默认) | 如 'rt'等价于'r' |
| 'b' | 打开为二进制文件 | 与上面组合，如 'rb'、'wb' |
| '+' | 读写 | 与上面组合，在原有模式上增加另一种能力：r+(不清空文件内容，从文件头开始覆盖写入)，w+(清空文件，重新写入)，a+(从原文件末尾追加写入) |
文件上下文管理器
- 推荐使用 with open('data.txt',encoding='utf-8') as f:进行上下文管理，优点是：在with块结束后，即便触发异常也会正常关闭文件，建议永远显式指定 encoding='utf-8'
- 手动通过 f.close() 关闭文件，需要通过 try-finally 手动维护
with open("text.txt",encoding="utf-8") as f:
for line in f:
print(line.rstrip()) # 去掉行尾换行（print也会输出换行，输出的行之间会有空行）
读取文件
- read(size=-1)：读取全部或指定字符数（文本模式为字符，二进制为字节），如：f.read()
- readline(size=-1)：读取一行（含行尾 \n）
- readlines(hint=-1)：读取所有行，返回列表
- 迭代读取(推荐)for line in f:逐行读取、处理，内存友好
with open('a.txt', encoding='utf-8') as f:
for line in f:
print(line,end='') # 去掉行尾换行写文件
- write(s)：写入字符串（文本模式）或字节（二进制模式），返回写入的字符/字节数，，如：f.write('hello\n')
- writelines(lines)：写入一个字符串序列，不会自动添加换行符， 如：f.writelines(['a\n', 'b\n', 'c\n'])
- print(..., file=f)：如果print写入，如：print('hello', file=f)
其他操作/属性
- f.tell()：返回当前指针位置（文本模式为不透明数字，二进制为字节偏移）
- f.seek(offset, whence=0)：移动指针位置，whence=0：从开头（默认），whence=1：从当前位置，whence=2：从末尾。文本模式只允许 seek(0)、seek(0, 2) 或 tell() 返回的位置
- f.flush()：强制将缓冲区内容写入磁盘
- f.readable() / f.writable() / f.seekable()：是否可读/写，返回bool值
- f.closed：是否已关闭
- f.name：文件名
- f.mode：打开模式
- f.encoding：编码
文件操作模板
实现像sed、awk一样修改某个文本行中的内容，建议：
- 大文件：r 读取原文件 + 写入临时文件（或新文件）+ 替换原文件
- 小文件：r 读取全部，再 w 写回
- 如果必须原地修改，推荐 fileinput 库
- 尽量不要依赖 r+、w+ 对原文件进行修改
#大文件，写入临时文件+替换
from pathlib import Path
import re
src = Path("config.txt")
tmp = Path("config.txt.tmp")
with src.open("r", encoding="utf-8") as fin, \
tmp.open("w", encoding="utf-8") as fout:
for line in fin:
line = re.sub(r"old_value", "new_value", line)
fout.write(line)
tmp.replace(src)
#小文件，完整读取
from pathlib import Path
import re
path = Path("config.txt")
text = path.read_text(encoding="utf-8")
text = re.sub(
r"port=\d+",
"port=8080",
text
)
path.write_text(text, encoding="utf-8")
模块/包/库
- 模块：一个.py文件就是一个模块
- 包（Package）：包含多个模块的目录，目录中必须有__init__.py文件（Python 3.3+ 可省略）
- 库（Library）：一组相关包的集合
模块
import模块导入
- import 模块名：导入整个模块
- import 模块名 as xxx：导入模块并起别名
- from 模块名 import xx,xx：从模块中导入指定内容，它们可以是函数、变量等
- from 模块名 import xx,xx as xxx：导入特定内容并起别名
模块
模块的查找路径：
- 导入模块时，Python 按以下顺序查找模块：
- 内置模块，如 sys、math
- sys.path 中的路径：
- 当前脚本所在目录
- PYTHONPATH 环境变量指定的目录
- Python 安装目录的标准库
- 第三方库目录（site-packages）
- 可以通过以下语句在脚本文件中临时添加模块查找路径(它们需要先 import sys 导入 sys 模块后才能用)：
- sys.path.append('路径')
- sys.path.insert(0, '路径')：添加到查找路径开头，优先级最高，以解决同名模块冲突
import sys
import os
# 把上级目录加入搜索路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from parent_package import some_module
导入模块后的行为：
- 导入模块后，可以通过模块名.模块内资源的方式调用这些资源
- 第一次导入模块时，Python 会执行模块中的顶层代码(文件中声明的变量、函数等)，这些被导入的模块对象会被存放在sys.modules这个字典中；之后再次导入同一个模块时，通常直接使用已经加载的模块，不会重新执行模块的顶层代码
- 每个模块都有一个__name__ 属性：
- 直接运行该文件时，__name__ == "__main__"
- 被导入时，__name__ 为模块名
print("模块被加载了！") # 顶层代码，首次导入模块该输出语句会被执行
x = 100 # 顶层代码
def foo(): # 函数内的print不会被执行
print("foo 被调用")
class Bar: # 定义类（顶层代码）
pass
编写模块
- 可以通过__author__指定模块作者
- 模块中，不希望用户调用的函数名、变量等内容，应该添加前缀_。python没有提供语言层面的private函数和变量控制，只能人为约定
- 脚本文件末尾可以通过 __name__=='__main__' 来决定，该脚本文件被主动执行时，它能正常执行代码；该脚本文件被当作模块导入时，则不能主动执行
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
' 模块说明 '
__author__ = 'shiwivi'
import sys
def _xxx:
.....
# 模块功能实现
if __name__=='__main__':
main()
包
可以将多个模块文件放到同一个目录下，形成一个包(Package)，包中通常包含一个__init__.py 文件，多个模块文件，以及一些子包，目录类似于：
myproject/
├── main.py
└── mypackage/ # 包目录
├── __init__.py # 包的初始化文件
├── module_a.py # 子模块
├── module_b.py # 子模块
└── subpackage/ # 子包
├── __init__.py
└── module_c.py__init__.py文件
__init__.py文件通常包含以下内容：
- __version__：包版本
- __author__：包作者信息
- 简化导入：提前导入子模块/子包，方便用户使用
- __all__：控制from package import *会导入哪些内容，语法为：__all__ = ["函数名", "类名"]
- 初始化代码：导入包时自动执行，可做初始化
# mypackage/__init__.py
# 1. 元信息
__version__ = "1.0.0"
__author__ = "Your Name"
# 2. 简化导入：让用户能直接 from mypackage import func_a
from .module_a import func_a, ClassA
from .module_b import func_b
# 3. 控制 from mypackage import * 的行为
__all__ = ["func_a", "ClassA", "func_b"]
# 4. 初始化代码(如日志、配置)
import logging
logging.getLogger(__name__).addHandler(logging.NullHandler())导入包
- import 包.模块：导入整个模块，可以使用 as 语句重命名
- from 包.模块 import *：从模块导入所有(受 __all__ 限制)
相对导入：只能在包内部使用，不能用于顶层脚本，下文可以替换使用(..)代表上级目录
- from . import 模块名：从同级目录导入模块，如：from . import module_b
- from .模块名 import xxx：从同级模块导入某个资源，如：from .module_b import func_b