Python

这篇文章总结一些平时容易被忽略的 Python 知识 Primitives, Variables and Expressions print(f"{year:>3d} {principal:0.2f}") >3d 指至少 3 位十进制数，右对齐 0.2f 指精度为 2 位的浮点数 Arithmetic Operators round(x, [n]): 该函数采用 Banker’s Rounding 银行家舍入法，也叫 四舍六入五成双，当要舍弃的数字正好是 5 时 前一位是偶数 → 向下舍去（向偶数靠拢） 如果前一位是奇数 → 向上进位（向偶数靠拢） 这样做的目的是减少舍入误差的累积，在统计学和金融计算中更为公平。 # 常规四舍五入（Python实际行为是银行家舍入） print(round(1.5)) # 2 （1是奇数，5进位） print(round(2.5)) # 2 （2是偶数，5舍去） print(round(3.5)) # 4 （3是奇数，5进位） print(round(4.5)) # 4 （4是偶数，5舍去） # 更复杂的例子 print(round(1.25, 1)) # 1.2 （2是偶数，5舍去） print(round(1.35, 1)) # 1.4 （3是奇数，5进位） print(round(1.251, 1)) # 1.3 （因为后面还有1，不是正好5，正常进位） 银行家舍入法是 IEEE 754 标准推荐的方式，Python、R、NumPy 等都采用这种舍入方式，能有效减少大量数据计算时的统计偏差。 Python 二进制运算符会将整数视为 2’s complement binary representation 二进制补码，并且符号位会在左侧无限扩展。 此外，Python 不会截断二进制，也不会溢出。 ...

Working with blocking sockets socket 是在网络中读取和写入数据的一种方式。 可以将 socket 看成一个邮件，将信封放到里面后运送到接收者的地址。 下面使用 Python 的内置 socket 模块来创建一个简单的 server import socket server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 这里，给 socket 函数指定了两个参数，第一个是 socket.AF_INET，这个告诉我们要与什么类型的地址进行交互，在这个例子中是 hostname 和 phonenumber。 第二个是 socket.SO_REUSEADDR，这个参数是说我们使用 TCP 协议进行交互。 然后使用 socket.setsockopt 方法将 socket.SOL_SOCKET 标志设置为 1。这将允许在关闭和快速重启应用，避免 address already in use 这类错误，如果不这样做将会消耗操作系统一段时间来解除与 port 的绑定。 使用 socket.socket 创建 socket 后，并不能开始沟通，因为还没有将其绑定到任何地址上面。 在本例中，将使用电脑本地地址 127.0.0.1 和任意 port 8000 server_address = ('127.0.0.1', 8000) server_socket.bind(server_address) 这里将地址设置为 127.0.0.1:8000，这意味着 client 将能够使用该地址向服务器发送数据，如果要向 client 发送数据，也会看到该地址为来源地址。 接下来，在套接字上调用 listen 方法，主动监听来自客户端的连接请求。 随后，通过调用 accept 方法等待连接建立。 该方法会保持阻塞状态直至接收到连接请求，当连接成功时，将返回一个连接对象及客户端地址。 这个连接对象本质上是一个新的套接字，可以用于与客户端进行双向数据通信 server_socket.listen() connection, client_address = server_socket.accept() 有了这些组件，我们便掌握了创建基于套接字的服务器应用所需的所有基础模块。 该应用将等待连接，并在建立连接后打印提示信息。 ...

Tasks, coroutines, furtures, and awaitables Coroutines 和 tasks 都是 await 表达式，那他们的相同线程是哪个？ 下面介绍 future 也被称作 awaitable，理解 futures 是理解 asyncio 内部工作的重点。 Introducing futures Future 代表一个尚未完成的异步操作的最终结果。 from asyncio import Future my_future = Future() print(f"Is my_future done? {my_future.done()}") my_future.set_result(42) print(f"Is my_future done? {my_future.done()}") print(f"What is the result of my_future? {my_future.result()}") 输出为 Is my_future done? False Is my_future done? True What is the result of my_future? 42 使用构造器 Future 来创建 future，这时 future 没有值，因此调用 done 结果是 False。 然后使用 set_result 设置值，这将 future 标记为 done。 相似的，如果想要在 future 中设置异常，使用 set_exception 方法。 ...

Introducing coroutines 创建一个协程 coroutine 而不是创建一个函数类型，使用 async def 关键字，而不是 def: async def coroutine_add_one(number: int) -> int: return number + 1 def add_one(number: int) -> int: return number + 1 function_result = add_one(1) coroutine_result = coroutine_add_one(1) print(f"Function result is {function_result} and the type is {type(function_result)}") print(f"Coroutine result is {coroutine_result} and the type is {type(coroutine_result)}") 输出如下 Function result is 2 and the type is <class 'int'> Coroutine result is <coroutine object coroutine_add_one at 0x103000a00> and the type is <class 'coroutine'> 可以看到，协程返回的不是值，而是一个协程对象。 这里协程并没有执行，而是创建了一个协程对象可在之后运行，要运行一个协程则必须显式地在一个事件循环中运行它。 在 Python 3.7 之后的版本，必须创建事件循环来运行它。 asyncio 库添加了多个函数，抽象了事件循环的管理，例如 asyncio.run()，可以使用它来运行协程： ...

Python asyncio 基础篇 本篇包含 asyncio 是什么以及如何使用它 concurrency 并发、parallelism 并行、threads 线程和 processes 进程 GIL (global interpreter lock) 全局解释器锁和其带来的并发跳转 非阻塞 sockets 如何只通过一个线程实现并发 基于事件循环 (event-loop-based) 并发的基本原理 异步编程 (asynchronous programming) 意思是可以在主程序之外，额外运行一个特定的长时运行的任务。 一个 coroutine 协程是一种方法，协程是一种方法，当遇到可能长时间运行的任务时，它可以暂停执行，并在任务完成后恢复执行。 asyncio 这个库的名称可能让人人为其只适合编写 I/O 操作，但实际上该库可以和 threading 和 multiprocessing 库结合使用。 基于这种 interoperability 互操作性，可以使用 async/await 关键字让工作流更加容易理解。 这意味着，asyncio 不仅适合 I/O 的并发，也可以在 CPU 密集操作中使用。 所谓的 I/O-bound 和 CPU-bound 是指限制程序运行更快的主要因素，这意味着如果增加该方面的性能，程序就能够在更短的时间内完成。 下面是一些例子 I/O 密集操作：网络请求、文件读取 CPU 密集操作：循环遍历文件夹、计算 pi import requests response = requests.get('https://www.example.com') # 1 items = response.headers.items() headers = [f'{key}: {headers}' for key, header in items] # 2 formatted_headers = '\n'.join(headers) # 3 with open('headers.txt', 'w') as file: # 4 file.write(formatted_headers) I/O-bound 网络请求 CPU-bound 响应处理 CPU-bound 字符串拼接 I/O-bound 写入磁盘 Concurrency 并发 和 Parallelism 并行的区别这里就不多说了。 ...

Python logging 基础指南 实际项目中，print() 只能满足基本的输出要求，而 logging 模块提供了更灵活、分级别、可配置的日志系统。 核心概念 Logger 记录器 记录器是拿来写日志的东西 logger = logging.getLogger(__name__) logger.info("开始执行任务") 接收日志消息，按级别判断是否要输出，并交给 Handler Handler 处理器 决定日志“去哪里”，有下面常见 Handler StreamHandler: 输出到控制台 FileHandler: 写入文件 RotatingFileHandler: 自动滚动文件 SMTHandler: 发邮件 SocketHandler: 发送到日志服务器 一个 Logger 可以挂多个 Handler Formatter 格式器 负责日志的格式 '%(asctime)s - %(levelname)s - %(name)s - %(message)s' 所有 Handler 都可以设置自己的 Formatter，不同输出渠道可以呈现不同格式 LogRecord 日志记录对象 每次调用 logging.info("hello") 内部都会生成一个 LogRecord 对象 LogRecord 是日志系统的“消息载体”，包括全部的元数据，例如： 时间戳 模块名 文件名、行号 日志级别 写入消息 message 线程 ID、进程 ID Filter 过滤器 Filter 是更细粒度的筛选工具，可以控制某个模块的日志，阻止某些关键字，基于上下文附加标签等。 简单使用 import logging logging.basicConfig( # 输出 INFO 及以上几倍日志 level=logging.INFO, # 时间 - 模块名 - 级别 - 消息内容 format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logging.info("程序已启动") logging.warning("磁盘空间将不足") logging.error("读取文件失败") 使用 Logger 对象：在较大的项目中，不会使用基础配置，而是为每个模块创建自己的 logger ...

pathlib - Object-oriented filesystem paths 此模块提供表示文件系统路径的类，其语义适用于不同的操作系统。 路径类分为： 用于纯计算无 I/O 的 pure paths 继承 pure paths 但是有 I/O 操作的 concrete paths 基本使用 导入 Path from pathlib import Path p = Path('.') 列出所有子目录 [x for x in p.iterdir() if x.is_dir()] 列出所有 py 源码文件 list(p.glob('**/*.py')) 在目录树中移动 p = Path('/etc') q = p / 'init.d' / 'reboot' # .resolve() 方法会解析所有符号链接，返回文件绝对路径 # mac os 中的 /etc 实际上是一个符号链接，指向 /private/etc q.resolve() 查询文件路径 q.exists() # 文件是否存在 q.is_dir() # 是否为目录 打开一个文件 q = Path('.') / 'file.py' with q.open() as f: # 读取第一行内容 f.readline() Pure paths 纯路径 Pure path 对象提供路径处理操作，这些操作无需真的访问操作系统。 有三种方法来操作这些类，也被称为 flavours (风格)： class pathlib.PurePath(*pathsegemnts) 为一个通用的类，代表当前系统的路径风格 >>> PurePath('setup.py') PurePosixPath('setup.py') pathsegments 的每个元素即可以是代表一个路径的字符串，也可以是实现了 os.PathLike 接口的对象，其中 fspath() 方法返回一个字符串，例如另一个路径对象： ...

这篇文章深入介绍 uv 管理 Python 项目的使用 Features Python versions uv python install: 安装 Python 版本 uv python list: 查看可用的 Python 版本 uv python find: 查找安装的 Python 版本 uv python pin: 固定当前项目的 Python 版本 uv python uninstall: 卸载一个 Python 版本 Scripts uv run: 运行一个脚本 uv add --script: 为脚本添加一个依赖 uv remove --script: 移除一个依赖 Projects 使用 pyproject.toml 配置项目 uv init: 创建一个 Python 项目 uv add: 为项目添加依赖 uv remove: 删除项目依赖 uv sync: 同步环境下的依赖 uv lock: 为项目依赖创建一个锁文件 uv run: 在项目环境执行命令 uv tree: 查看项目依赖树 uv build: 将项目构建为分发归档文件 uv publish: 将项目发布到包索引 Tools 允许与安装工具 ...

Grabage Collection In Python 本篇文章介绍 Python 中的 Grabage Collection (GC) 机制介绍 What’s Python Object? Python 对象中有三样东西: 类型(Type)、值(value)和引用计数(reference count), 当给变量命名时, Python 会自动检测其类型, 值在定义对象时声明, 引用计数是指该对象名称的数量. 首先来看一个类 class Person: def __init__(self, name, unique_id, spouse): self.name = name self.unique_id = unique_id self.spouse = spouse def __del__(self): print( # !r: 调用 repr() 来获取该对象的字符串表达式 # !s: str() # !a: ascii() f"Object {self.unique_id!r} is about to be removed from memory. Goodbye!" ) 该 Person 类有以下3个属性: name: 人名 unique_id: 唯一性 id spouse: 将为 None 或者将存储另一个 Person 对象 有一个特殊方法 __del__(), 这个特殊方法有一定的误导性. 该方法并不像 __len__() 与 len() 或者 __iter__() 与 iter() 那样与 del 关键字相关联. __del__() 特殊方法并不定义当对对象引用时 del 会发送什么, 相反, __del__() 是一个终结器 finaliser: 它在对象被消毁之前从内存中移除之间被调用. 因此, __del__() 中 print() 调用的字符串仅在 Python 即将从内存中移除对象时显示. ...

python 中 asyncio 和 gevent 是两种协程(在一个线程内实现并发)的实现, 这篇文章对比介绍这两者实现. 下面先介绍一下基础概念: Coroutines 协程 在 Python 中, 协程是可以暂停和继续运行的函数, 使得其是否适合并发编程. 定义使用 async def 语法, 协程运行编写非阻塞的操作. 在协程内, await 关键字用于暂停执行, 直到给定的任务完成, 从而运行其他协程在此其间并发运行. Event Loop 事件循环 事件循环是一种控制结构, 它不断地处理一系列事件, 处理任务并管理程序的执行流程. 等待事件发生, 处理后再等待下一个事件. 这种机制确保程序能够以高效有序的方式响应事件, 例如用户输入、计时器或者消息. 下面是事件循环如何管理协程: 任务提交: 当向事件循环提交一个协程时, 其被封装在一个 Task 对象中, 然后任务被安排在事件循环上运行. 内部队列: 事件循环使用几个内部数据结构来管理和调度这些任务 就绪队列 (Ready Queue): 包含可以立即运行的任务. I/O 选择器 (I/O Selector): 监控文件描述符, 并根据 I/O 准备情况调度任务 计划回调 (Scheduled Callbacks): 管理计划在一定延迟后运行的任务. 调度: 事件循环不断检查这些队列和数据结构, 以确定哪些任务已准备好执行. 然后它运行这些任务, 在遇到 await 语句时, 根据需要暂停和恢复它们. 并发管理: 通过交错执行多个协程, 事件循环无需多个线程即可实现并发. 在任何时候, 只有一个任务会运行, 但如果一个任务是 I/O 密集型的, 它会切换到另一个任务, 给人一种并行的错觉. ...