编程学习网 > 编程语言 > Python > 强烈推荐的 7 个 神级 Python 库!
2026
07-24

强烈推荐的 7 个 神级 Python 库!

一个不到两百行的导入脚本,CSV 还没处理完,内存先顶满了。

另一个同步数据的任务更离谱,第三方接口抖了一下,整个批次直接退出,前面跑的半小时全白费。

这种代码我一般不急着优化循环,也不先上多线程。先看库是不是选错了。Python 自带工具够全,但真放到业务里跑,有些活还是得交给更专业的库。

下面这 7 个,是我现在写接口、清洗数据和做运维脚本时经常会翻出来的。

Pydantic:别再手写一堆字段判断

接口参数、Excel 导入、MQ 消息,只要数据来自外部,我都不太信。

以前的代码经常是这种味道:

if not row.get("order_no"):
    raise ValueError("订单号不能为空")
if float(row["amount"]) <= 0:
    raise ValueError("金额错误")

字段一多,判断散得到处都是。换成 Pydantic,校验规则跟数据结构放在一起:

from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator

class PaymentRow(BaseModel):
    order_no: str = Field(min_length=6)
    amount: Decimal = Field(gt=0)
    paid_at: datetime
    @field_validator("order_no", mode="before")
    @classmethod
    def clean_order_no(cls, value: object) -> str:
        return str(value).strip().upper()

row = PaymentRow.model_validate({
    "order_no": " od202601 ",
    "amount": "39.90",
    "paid_at": "2026-07-10 09:30:00",
})

类型转换、字段约束、错误定位一起处理。尤其是导入数据,报错能直接落到具体字段,比一句数据格式不正确强多了。

HTTPX:请求接口时,超时要写明白

requests 我还在用,但新写异步服务或者批量调用接口,我更愿意上 HTTPX

它同时提供同步和异步 API,并且默认就有网络超时控制。这个细节很重要,线上最怕的不是接口报错,而是连接一直挂着不回来。

import httpx

async def fetch_stock(sku: str) -> int:
    timeout = httpx.Timeout(connect=2, read=5, write=5, pool=1)
    limits = httpx.Limits(
        max_connections=20,
        max_keepalive_connections=8,
    )
    async with httpx.AsyncClient(
        base_url="https://inventory.internal",
        timeout=timeout,
        limits=limits,
    ) as client:
        response = await client.get("/stock", params={"sku": sku})
        response.raise_for_status()
        return int(response.json()["available"])

这里我一般不会只写一个 timeout=5。连接、读取、连接池等待,最好拆开,不然后面排查超时时,只能对着一条模糊日志猜。

Polars:大 CSV 别一上来全塞进内存

碰到几十万行以上的 CSV,我第一眼先看是不是还在用普通循环逐行处理。

Polars 的懒执行很适合这类任务。scan_csv 不会立刻把整个文件读进来,过滤和列裁剪还能提前下推。

import polars as pl

report = (
    pl.scan_csv("data/orders-*.csv")
    .filter(pl.col("status") == "PAID")
    .select("shop_id", "amount")
    .group_by("shop_id")
    .agg(pl.col("amount").sum().alias("paid_amount"))
    .sort("paid_amount", descending=True)
    .collect()
)
report.write_csv("output/shop-payment.csv")

这段代码真正有用的地方,不是比 Pandas 少写几行,而是没必要先把无关状态、无关字段全部读完,再慢慢扔掉。

Rich:脚本跑到哪了,别让人猜

批处理脚本最烦的一种状态是:终端没报错,也没输出,不知道它在跑,还是已经卡死。

Rich 能做进度条、表格和日志美化。它不是纯粹为了好看,长任务有明确进度,排查时能少开一个终端查进程。

from pathlib import Path
from rich.progress import track

files = list(Path("incoming").glob("*.json"))
for file_path in track(files, description="校验订单文件"):
    raw = file_path.read_bytes()
    check_order_file(file_path.name, raw)

正式任务里,我还会把当前文件名、成功数和失败数放进去。只显示一个花里胡哨的进度条,出了问题还是没法定位。

Typer:把临时脚本做成正经命令

很多内部工具最后都会变成这样:

python clean.py data.csv true 30

过两个月,连作者自己都忘了 true  30 分别是什么。

Typer 可以直接利用类型标注生成命令参数和帮助信息,适合做导入、清理、巡检这类内部 CLI

from pathlib import Path
from typing import Annotated
import typer

app = typer.Typer()

@app.command()
def clean(
    source: Annotated[Path, typer.Argument(exists=True)],
    dry_run: Annotated[
        bool, typer.Option("--dry-run", help="只检查,不写入结果")
    ] = False,
) -> None:
    clean_expired_records(source, dry_run)

if __name__ == "__main__":
    app()

至少执行 python tool.py --help 时,别人能看懂怎么用,不用再去翻源码。

Tenacity:重试不是套一个死循环

接口超时之后直接重试没问题,问题是很多代码会毫无间隔地重试十几次,把本来就不稳定的下游再锤一遍。

Tenacity 可以限制异常类型、重试次数和等待策略。

import httpx
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_random_exponential,
)

@retry(
    retry=retry_if_exception_type(
        (httpx.ConnectError, httpx.ReadTimeout)
    ),
    stop=stop_after_attempt(4),
    wait=wait_random_exponential(multiplier=0.5, max=8),
    reraise=True,
)
def push_invoice(payload: dict) -> None:
    response = httpx.post(
        "https://billing.internal/invoices",
        json=payload,
        timeout=4,
    )
    response.raise_for_status()

注意,我不会把所有异常都重试。参数错误、权限错误,重试多少次都没用,只是在制造垃圾请求。

orjsonJSON 很多时,别只盯业务代码

接口响应大、日志事件多、缓存里频繁存 JSON 时,序列化也可能变成耗时点。

orjson 支持常见 Python 类型,dumps 返回的是字节数据,写文件和网络响应时反而省了一次编码。

from pathlib import Path
import orjson

EVENT_FILE = Path("logs/order-events.jsonl")

def append_event(event: dict) -> None:
    payload = orjson.dumps(
        event,
        option=orjson.OPT_APPEND_NEWLINE,
    )
    with EVENT_FILE.open("ab") as file:
        file.write(payload)

这里有个容易踩的点:它返回 bytes,不是 str。旧代码直接替换标准库 json.dumps,很可能在拼接字符串或者写文本文件时出错。

7 个库没必要一次全装。

接口多,先看 PydanticHTTPX Tenacity;数据文件多,先上 Polars;内部脚本准备长期使用,再补 Typer Rich。至于 orjson,先看 profilingJSON 根本不在耗时链路上,就别为了性能优化硬换。

库选对了,代码通常不是变得更炫,而是少写几层补丁,出问题时也知道该从哪一层查。

以上就是“强烈推荐的 7 个 神级 Python 库!的详细内容,想要了解更多Python教程欢迎持续关注编程学习网。 

扫码二维码 获取免费视频学习资料

Python编程学习

查 看2022高级编程视频教程免费获取