【Python】一些PEP提案（四）：scandir、类型约束，异步asyncawait

PEP 471 – os.scandir () function，遍历目录

os.scandir() 返回一个迭代器，每次产生一个 os.DirEntry 对象，该对象同时包含文件名和文件属性（如大小、修改时间、是否为目录等）。传统的os.listdir()仅会返回文件的名字。

比如我们想遍历目录下的所有文件和子目录（Linux里目录也看做文件），os.listdir()的实现如下：

import osfor name in os.listdir('.'):path = os.path.join('.', name)print(name, os.path.isfile(path))  # 需要额外调用 os.path.isfile()

然而文件名和是否为文件是可以作为对象的一个属性的，如果我们查找到的文件会包装成对象的话，所以用os.scandir() 可以这么写：

import oswith os.scandir('.') as entries:for entry in entries:print(entry.name, entry.is_file())  # 直接通过 entry.is_file() 判断

再比如获取文件属性（如大小、修改时间），os.listdir()的实现如下：

import os
from datetime import datetimefor name in os.listdir('.'):path = os.path.join('.', name)if os.path.isfile(path):stat = os.stat(path)  # 额外的系统调用print(f"{name}: {stat.st_size} bytes, modified at {datetime.fromtimestamp(stat.st_mtime)}")

os.stat意味着一次额外的系统调用，但是事实上可以将stat作为类的一个属性，比如os.scandir() 的写法：

import os
from datetime import datetimewith os.scandir('.') as entries:for entry in entries:if entry.is_file():stat = entry.stat()  # 无需额外系统调用（属性已缓存）print(f"{entry.name}: {stat.st_size} bytes, modified at {datetime.fromtimestamp(stat.st_mtime)}")

PEP 484 – Type Hints，类型约束

python的类型是动态确定的，因此早期python的变量不会标注类型。但这对IDE带来了不小的挑战，设想一种场景，你设计某个变量logger，其类型为logging的Logger，然后你去写代码logger.info，然后发现IDE并不能自动补全.info的部分，因为IDE并不知道变量的类型。

即使IDE在多数场景都会自动推导变量的类型，上面的情况也是很常见的。更何况大部分情况下阅读代码的场景并不在IDE里——当团队评审代码的时候，比如在gerrit上——我知道程序员信奉“好的代码不需要注释”，但适当的类型提示确实有助于降低代码的阅读成本。

python的类型约束使用后置类型，这也是大多数拥有自动推导类型的编程语言的做法（除了C++这种一开始不支持自动推导类型的语言）：

def greeting(name: str) -> str:return "Hello, " + name# 参数 `name` 预期为 str 类型，返回值预期为 str 类型

这样做的好处是可以和不带类型约束的代码完美兼容。另外，本文无意讨论前置和后置类型的优劣。

python也支持类似于C++中using和typedef的用法，即给类型起一个别名：

Vector = list[float]  # 定义类型别名def scale(scalar: float, vector: Vector) -> Vector:return [scalar * num for num in vector]# 使用别名进行类型提示

python的函数并不限制返回值的类型是唯一的（或者说返回值的“不唯一类型”可以用下面的“唯一”来抽象）：
使用Optional可以声明可选的None返回值：

from typing import Optional, Uniondef get_name() -> Optional[str]:  # 可能返回 str 或 None...

使用Union可以声明可能的多个返回值类型：

def process_value(value: Union[int, str]) -> None:  # 接受 int 或 str...

Literal用来指定变量的具体字面值，如果你用VSCode的Pylance插件开启类型提示，你会发现插件对于函数只读变量的返回值，默认分配的类型就是Literal

from typing import Literaldef move(direction: Literal["up", "down", "left", "right"]) -> None:...

更多的类型提示可以开启你IDE（Pycharm自带，VSCode需要安装Pylance）类型提示，它提示你的一般是对的。当然也有不对的时候，比如我的Pylance至今无法识别负责带yield的函数的可迭代。

最后需要指出的一点是，通常来说，类型约束不是强制的，不会改变编译结果。比如下面的代码：

def add(a: int, b: int) -> int:return a + bresult = add("hello", "world") 

类型标注都是错的，但不耽误代码运行。

但是部分框架可能会内置了强制的类型检查，或者针对类型标注进行了特殊处理，需要结合具体场景来分析。

PEP 492 – Coroutines with async and await syntax，async/await 语法

仅仅靠yield对协程的支持已经不太够了，async/await的提出正是为了提升python对协程的支持。不过我这里也只是粗略地介绍下，异步编程需要大量的实践提升代码感知。

协程是一种特殊的函数，可在执行过程中暂停并恢复，允许其他代码在暂停期间运行。在 Python 中，协程使用 async def 定义：

async def task(num):print(f"start task {num}")await io_func()print(f"end task {num}")

而await 只能在 async def 定义的协程函数中使用，用于暂停当前协程的执行，等待另一个异步操作（如另一个协程或 Future 对象）完成。比如上述代码就是在等待io_func完成（假设io_func是个io密集型操作）。

而协程的意义就在这里。

对于普通的同步函数，io_func会阻塞整个进程。此时CPU完全处于空闲的状态，压力都在io操作上，浪费了很多CPU资源。但是在异步环境下，此时外层事件循环就可以暂停函数的执行，将控制权交给其他异步函数。当io_func执行完，外层事件循环再将控制权交回，恢复task函数的执行。

打个比方就是，task就是烧水，前面的从水龙头接水等操作完成后，io_func就是加热并等待水开，此时你可以做点其他的事，等水开了再回来，然后把水倒进暖壶里。

不过不用担心，事件循环有现成的库asyncio，不需要你来完成上面的调度。

async def io_func():await asyncio.sleep(2)async def task(num):print(f"start task {num}")await io_func()print(f"end task {num}")async def tasks():await asyncio.gather(task(1), task(2))asyncio.run(tasks())

上述代码应该输出：

start task 1
start task 2
end task 1
end task 2

给人一种并行的假象，其实是并发。代码并非使用了多个线程，而是单线程交替运行。

所以协程不适合的第一个场景就是，从头到尾只有一个任务在运行。好比就算你可以在等水开的时候干点其他事，但是如果你没事可干，那这段时间本质上也干不了别的。此时异步和同步就没区别了。

另外一个就是协程只适用于io密集型任务，如果你是CPU密集型任务，那就不适合了。CPU压力本来就大，协程的调度更加大了CPU的压力。此时更适合使用多线程，甚至多进程来优化。

其他的用法比如，在协程中使用 yield 生成数据，通过 async for 消费：

async def generate_data():for i in range(3):await asyncio.sleep(1)yield iasync def main():async for data in generate_data():print(data)

通过 async with 使用异步资源，确保正确的获取和释放：

class AsyncDatabase:async def __aenter__(self):await self.connect()  # 异步连接数据库return selfasync def __aexit__(self, exc_type, exc, tb):await self.disconnect()  # 异步关闭连接async def main():async with AsyncDatabase() as db:await db.query("SELECT * FROM users")

需要注意的是，在异步函数中尽量全部使用异步方法，同步方法会阻塞整个事件循环。比如如果你之前使用的time.sleep()，需要换成asyncio.sleep()