【Python】一些PEP提案（四）：scandir、类型约束，异步async&await
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 osfrom 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 osfrom 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 1start task 2end task 1end 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 self async 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()
