Python asyncio 实战:从入门到踩坑

Python 3.5 引入 asyncio 之后,异步编程逐渐成为 Python 工程师的必备技能。但从"跑通 demo"到"在生产项目里用好它",中间有一段不短的距离。 event loop 的生命周期 很多人第一次写 asyncio 代码是这样的: import asyncio async def main(): await asyncio.sleep(1) print("done") asyncio.run(main()) asyncio.run() 是 Python 3.7 引入的,它做了三件事: 创建一个新的 event loop 运行传入的协程直到完成 关闭 event loop 并清理资源 注意:不要在已经运行的 event loop 里调用 asyncio.run(),这会抛出 RuntimeError。 常见坑:忘记 await async def fetch_data(): return await some_async_operation() # 错误写法——fetch_data() 返回的是协程对象,不是结果 result = fetch_data() # 正确写法 result = await fetch_data() 协程对象不会自动执行,必须被 await、asyncio.create_task() 或 asyncio.gather() 驱动。 并发执行多个任务 import asyncio async def fetch(url): # 模拟 IO 操作 await asyncio.sleep(0.1) return f"result from {url}" async def main(): urls = ["url1", "url2", "url3"] # 并发执行,总耗时约 0.1s 而不是 0.3s results = await asyncio.gather(*[fetch(url) for url in urls]) print(results) asyncio.run(main()) asyncio.gather() 并发调度多个协程,是最常用的并发原语。 ...

2022-09-15 · 1 min · Kada Liao

Python 虚拟环境管理:从 venv 到 uv

Python 的包管理历史是一部混乱史。这篇文章梳理一下各种工具的演变,以及我现在的实践选择。 为什么需要虚拟环境 不同项目依赖不同版本的库。不用虚拟环境,所有包都装在系统 Python 里,版本冲突是迟早的事: 项目 A:需要 Django 3.2 项目 B:需要 Django 4.2 → 只能装一个,两个项目不能同时开发 虚拟环境为每个项目创建独立的 Python 环境,依赖互不干扰。 各种工具的演变 venv(Python 3.3+ 内置) python -m venv .venv source .venv/bin/activate # macOS/Linux .venv\Scripts\activate # Windows pip install requests pip freeze > requirements.txt 最基础,不需要额外安装,但功能简单,requirements.txt 不区分开发依赖和生产依赖。 virtualenv + pip-tools pip install pip-tools # requirements.in(只写直接依赖) requests>=2.28 flask>=2.0 # 生成锁定版本的 requirements.txt pip-compile requirements.in # 安装 pip-sync requirements.txt pip-compile 解决了依赖锁定的问题,是很长一段时间内的最佳实践。 Poetry poetry new my-project poetry add requests poetry add pytest --group dev poetry install # pyproject.toml 管理所有配置 Poetry 把依赖管理、打包、发布整合在一起,pyproject.toml 是单一配置文件。问题是速度慢,解析依赖时间长。 ...

2021-10-05 · 1 min · Kada Liao

Python 生成器与迭代器:惰性求值的力量

处理大文件或数据流时,把所有数据加载进内存是不现实的。生成器提供了一种"按需生成"的方式,解决了这个问题。 迭代器协议 Python 的 for 循环本质上是在调用迭代器协议: # for 循环等价于这段代码 it = iter(my_list) # 调用 __iter__,返回迭代器 while True: try: item = next(it) # 调用 __next__,取下一个元素 # 执行循环体 except StopIteration: break 实现了 __iter__ 和 __next__ 的对象就是迭代器。 生成器函数 用 yield 的函数就是生成器函数,调用它返回一个生成器对象: def count_up(start, end): current = start while current <= end: yield current # 暂停,返回 current,等待下次调用 current += 1 for n in count_up(1, 5): print(n) # 1 2 3 4 5 关键在于暂停和恢复:每次 next() 调用,函数从上次 yield 的地方继续执行,直到下一个 yield。 实际用处:处理大文件 # 不好:一次性加载所有行到内存 def read_all_lines(filename): with open(filename) as f: return f.readlines() # 10GB 文件直接 OOM # 好:逐行生成,内存始终只有一行 def read_lines(filename): with open(filename) as f: for line in f: yield line.strip() # 使用 for line in read_lines("huge_file.csv"): process(line) 生成器表达式 列表推导式的"惰性版本": ...

2021-06-30 · 2 min · Kada Liao

Python 类型注解实战:让代码更可维护

Python 是动态语言,但这不意味着不需要类型。类型注解让 IDE 提示更准确、代码意图更清晰、重构更安全。 基础语法 # 变量注解 name: str = "Kada" age: int = 30 scores: list[float] = [9.5, 8.8, 9.2] # Python 3.9+ # 函数注解 def greet(name: str, times: int = 1) -> str: return f"Hello, {name}! " * times # 返回 None def log(message: str) -> None: print(message) 常用类型 from typing import Optional, Union, List, Dict, Tuple, Any # Optional:可以是 None def find_user(user_id: int) -> Optional[str]: ... # Union:多种类型之一(Python 3.10 可以用 X | Y) def process(data: Union[str, bytes]) -> str: ... # Python 3.10+ def process(data: str | bytes) -> str: ... # 字典和列表(Python 3.9+ 可以直接用小写) def get_config() -> dict[str, Any]: ... # Tuple:固定长度和类型 def get_coordinates() -> tuple[float, float]: ... TypedDict:给字典加类型 JSON 数据经常用字典传递,TypedDict 让字典有类型检查: from typing import TypedDict class UserInfo(TypedDict): id: int name: str email: str age: int | None # 可选字段 def create_user(info: UserInfo) -> None: print(info["name"]) # IDE 有提示,且知道是 str 类型 dataclass:更结构化的数据类 比 TypedDict 更进一步,dataclass 是带类型的数据容器: ...

2021-04-08 · 2 min · Kada Liao

Python with 语句与上下文管理器

with 语句在 Python 里随处可见,但很多人只会用,不了解它的工作原理。这篇文章把上下文管理器讲清楚。 with 解决了什么问题 最典型的场景是文件操作: # 没有 with:如果中间报错,文件不会被关闭 f = open("data.txt") data = f.read() f.close() # 有 with:无论是否报错,退出 with 块时一定会关闭文件 with open("data.txt") as f: data = f.read() with 保证了"进入时执行某些操作,退出时一定执行另一些操作",无论退出是正常还是异常。 底层原理:enter 和 exit 实现了 __enter__ 和 __exit__ 的对象就是上下文管理器: class ManagedFile: def __init__(self, filename): self.filename = filename def __enter__(self): self.file = open(self.filename) return self.file # as 后面的变量就是这个返回值 def __exit__(self, exc_type, exc_val, exc_tb): self.file.close() # 返回 True 表示吞掉异常,False 或 None 表示继续传播异常 return False with ManagedFile("data.txt") as f: data = f.read() __exit__ 接收三个参数:异常类型、异常值、traceback。如果没有异常,三者都是 None。 更简单的写法:contextmanager 装饰器 用类实现太繁琐,contextlib.contextmanager 让你用生成器函数实现上下文管理器: from contextlib import contextmanager @contextmanager def managed_file(filename): f = open(filename) try: yield f # yield 之前是 __enter__,yield 的值是 as 后面的变量 finally: f.close() # yield 之后是 __exit__,finally 保证一定执行 with managed_file("data.txt") as f: data = f.read() yield 把函数分成两半:之前是进入逻辑,之后是退出逻辑。 ...

2021-01-22 · 2 min · Kada Liao

Python 装饰器深入理解:从语法糖到元编程

装饰器是 Python 里我最喜欢的特性之一——它让你能在不修改函数本身的情况下,给它增加行为。这篇文章从头梳理一下装饰器的原理和用法。 从闭包说起 装饰器本质上是闭包。理解闭包是理解装饰器的前提: def outer(x): def inner(): print(x) # inner 捕获了外层的 x return inner f = outer(42) f() # 输出 42,即使 outer 已经返回 inner 函数"记住"了它定义时所在的环境(x = 42),这就是闭包。 装饰器的基本形态 装饰器就是一个接受函数、返回函数的函数: def my_decorator(func): def wrapper(*args, **kwargs): print("函数执行前") result = func(*args, **kwargs) print("函数执行后") return result return wrapper @my_decorator def say_hello(name): print(f"Hello, {name}!") # 等价于:say_hello = my_decorator(say_hello) say_hello("Kada") @my_decorator 只是语法糖,等价于 say_hello = my_decorator(say_hello)。 functools.wraps:保留函数元信息 用装饰器之后,函数的 __name__、__doc__ 会被替换成 wrapper 的: print(say_hello.__name__) # 输出 "wrapper",而不是 "say_hello" 用 functools.wraps 解决: import functools def my_decorator(func): @functools.wraps(func) # 把原函数的元信息复制过来 def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper 这是写装饰器的最佳实践,几乎任何时候都应该加上。 ...

2020-03-12 · 2 min · Kada Liao

Python 2 到 Python 3 迁移实战:我们是怎么做的

Python 2 的 EOL(End of Life)定在 2020 年 1 月 1 日。我们团队从 2019 年中开始规划迁移,历时约半年完成。这里记录一下整个过程中遇到的问题和解决方案。 为什么不能拖 很多团队的心态是"能跑就行,等依赖库停止支持再说"。但现实是: 安全补丁不再提供,线上系统面临风险 新的第三方库越来越多只支持 Python 3 招进来的新人都不熟 Python 2,维护成本越来越高 迁移前的准备 用 2to3 扫描 Python 官方提供了 2to3 工具,可以扫描出大多数需要改动的地方: 2to3 -l # 列出所有可用的修复器 2to3 -f all your_project/ --no-diffs # 只扫描不修改,看有多少问题 用 pylint 或 pyupgrade 批量处理 pip install pyupgrade # 批量转换为 Python 3.6+ 语法 find . -name "*.py" | xargs pyupgrade --py36-plus 最常见的改动点 print 语句 # Python 2 print "hello" print "hello", "world" # Python 3 print("hello") print("hello", "world") integer 除法 # Python 2:整数除整数 = 整数 5 / 2 # 结果是 2 # Python 3:真除法 5 / 2 # 结果是 2.5 5 // 2 # 整除,结果是 2 这个坑最隐蔽,2to3 不会自动处理,需要手动审查所有除法运算。 ...

2019-11-20 · 2 min · Kada Liao