Python开发者的9个"神器"库深度测评

from fastapi import FastAPI                                                 from pydantic import BaseModel                                             app = FastAPI()                                                             class User(BaseModel):                                                     id: int     name: str     email: str  @app.post("/users") async def create_user(user: User):     """创建新用户"""     return {"message": f"用户 {user.name} 创建成功", "user_id": user.id}  @app.get("/users/{user_id}") async def get_user(user_id: int):     """获取用户信息"""     return {"user_id": user_id, "name": "示例用户"}

from databases import Database  database = Database("postgresql://user:password@localhost/dbname")  @app.on_event("startup") async def startup():     await database.connect()  @app.on_event("shutdown") async def shutdown():     await database.disconnect()  @app.get("/products/{product_id}") async def get_product(product_id: int):     query = "SELECT * FROM products WHERE id = :id"     return await database.fetch_one(query, {"id": product_id})

import pandas as pd  # 读取CSV文件，自动检测编码 df = pd.read_csv('large_data.csv', encoding='utf-8')  # 数据清洗 - 处理缺失值 df['price'].fillna(df['price'].mean(), inplace=True) df.drop_duplicates(subset=['user_id'], keep='first', inplace=True)  # 数据转换 df['date'] = pd.to_datetime(df['date']) df['year'] = df['date'].dt.year  # 分组聚合 result = df.groupby('category')['value'].agg(['mean', 'sum', 'count'])

from multiprocessing import Pool  def process_chunk(chunk):     """处理数据块的函数"""     return chunk.groupby('category')['value'].sum()  # 分块读取并并行处理 chunk_size = 100000 chunks = pd.read_csv('very_large_file.csv', chunksize=chunk_size)  with Pool(processes=4) as pool:     results = pool.map(process_chunk, chunks)  # 合并结果 final_result = pd.concat(results)

# 1. 选择合适的数据类型减少内存 df['user_id'] = df['user_id'].astype('int32')  # 而非int64 df['status'] = df['status'].astype('category')  # 分类数据  # 2. 使用query方法提升查询性能 filtered = df.query('price > 100 and category in ["A", "B"]')  # 3. 向量化操作替代循环 df['total'] = df['price'] * df['quantity']  # 向量化操作

import torch import torch.nn as nn import torch.optim as optim  # 定义模型 class SimpleNN(nn.Module):     def __init__(self):         super().__init__()         self.fc1 = nn.Linear(784, 128)         self.fc2 = nn.Linear(128, 10)      def forward(self, x):         x = torch.relu(self.fc1(x))         x = self.fc2(x)         return x  model = SimpleNN() criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.001)

# 使用torch.compile进行编译加速（PyTorch 2.5新特性） compiled_model = torch.compile(model)  # 训练循环 for epoch in range(10):     for batch_x, batch_y in dataloader:         optimizer.zero_grad()         outputs = compiled_model(batch_x)         loss = criterion(outputs, batch_y)         loss.backward()         optimizer.step()

import torch.distributed as dist  # 初始化分布式环境 dist.init_process_group("nccl")  # 使用DistributedDataParallel进行分布式训练 model = model.to(rank) ddp_model = torch.nn.parallel.DistributedDataParallel(model, device_ids=[rank])

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.orm import sessionmaker, declarative_base  Base = declarative_base()  class User(Base):     __tablename__ = 'users'     id = Column(Integer, primary_key=True)     name = Column(String(50))     email = Column(String(100))  # 创建引擎和会话 engine = create_engine('sqlite:///example.db') Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session()  # 添加数据 new_user = User(name='Alice', email='alice@example.com') session.add(new_user) session.commit()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import selectinload  # 创建异步引擎 async_engine = create_async_engine(     "postgresql+asyncpg://user:pass@localhost/db",     echo=True )  async_session = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)  async def get_users_with_posts():     async with async_session() as session:         # 使用Select进行查询         stmt = select(User).options(selectinload(User.posts))         result = await session.execute(stmt)         users = result.scalars().all()         return users

# 1. 使用新的Select API from sqlalchemy import select  stmt = select(User).where(User.name == 'Alice')  # 2. 关系加载优化 stmt = select(User).options(selectinload(User.posts))  # 3. 结果集处理 result = await session.execute(stmt) users = result.scalars().all()

from pydantic import BaseModel, EmailStr, Field, field_validator  class User(BaseModel):     id: int     name: str = Field(min_length=2, max_length=50)     email: EmailStr     age: int = Field(ge=0, le=120)      @field_validator('name')     @classmethod     def name_must_contain_space(cls, v):         if ' ' not in v:             raise ValueError('must contain a space')         return v  # 使用示例 user = User(id=1, name="Alice Smith", email="alice@example.com", age=25) print(user.model_dump())

from typing import List, Optional from pydantic import BaseModel, HttpUrl, field_validator  class Product(BaseModel):     name: str     price: float = Field(gt=0)     tags: List[str] = []     url: Optional[HttpUrl] = None     stock: int = Field(ge=0)      @field_validator('tags')     @classmethod     def tags_must_not_be_empty(cls, v):         if len(v) > 10:             raise ValueError('tags cannot exceed 10 items')         return v      def calculate_discount(self, discount_rate: float) -> float:         """计算折扣价格"""         return self.price * (1 - discount_rate)

# 1. 使用Pydantic运行时编译 from pydantic import TypeAdapter  # 创建类型适配器，性能更好 user_adapter = TypeAdapter(List[User]) users_data = [{'id': i, 'name': f'User{i}', 'email': f'user{i}@example.com'} for i in range(100)] users = user_adapter.validate_python(users_data)  # 2. JSON序列化优化 json_data = user_adapter.dump_json(users)  # 更快的JSON序列化

from rich.console import Console from rich.text import Text  console = Console()  # 彩色文本 text = Text.assemble(     "Hello, ",     ("World", "bold red"),     "!", ) console.print(text)  # 进度条 from rich.progress import track import time  for i in track(range(100), description="处理中..."):     time.sleep(0.01)

from rich.console import Console from rich.table import Table from rich.layout import Layout from rich.panel import Panel  console = Console()  # 创建表格 table = Table(title="库性能对比") table.add_column("库", style="cyan", width=20) table.add_column("速度", justify="right", style="green") table.add_column("内存", justify="right", style="red")  table.add_row("Polars", "⚡⚡⚡", "🟢低") table.add_row("Pandas", "⚡", "🔴高") console.print(table)

from rich.syntax import Syntax from rich.console import Console  console = Console()  code = ''' def hello_world():     """打印Hello World"""     print("Hello, World!") '''  syntax = Syntax(code, "python", theme="monokai", line_numbers=True) console.print(syntax)

import httpx  # GET请求 response = httpx.get("https://api.github.com/users/octocat") print(response.status_code) print(response.json())  # POST请求 data = {"name": "Alice", "email": "alice@example.com"} response = httpx.post("https://httpbin.org/post", json=data) print(response.text)

import httpx import asyncio  async def fetch_multiple_urls():     urls = [         "https://api.github.com/users/octocat",         "https://api.github.com/users/torvalds",         "https://api.github.com/users/gvanrossum"     ]      async with httpx.AsyncClient() as client:         tasks = [client.get(url) for url in urls]         responses = await asyncio.gather(*tasks)          for response in responses:             print(f"Status: {response.status_code}, URL: {response.url}")  asyncio.run(fetch_multiple_urls())

# 使用会话提高性能 with httpx.Client() as client:     # 复用连接     for i in range(10):         response = client.get("https://httpbin.org/get")         print(f"Request {i}: {response.status_code}")  # 设置超时和重试 timeout = httpx.Timeout(10.0, connect=5.0) limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)  client = httpx.Client(timeout=timeout, limits=limits)

# 启用HTTP/2 client = httpx.Client(http2=True)  # 异步HTTP/2 async_client = httpx.AsyncClient(http2=True)

import asyncio  async def fetch_data(url):     """模拟异步获取数据"""     await asyncio.sleep(1)  # 模拟IO操作     return f"Data from {url}"  async def main():     # 创建任务     task1 = asyncio.create_task(fetch_data("https://api.example.com/1"))     task2 = asyncio.create_task(fetch_data("https://api.example.com/2"))      # 等待任务完成     result1 = await task1     result2 = await task2      print(result1, result2)  asyncio.run(main())

import asyncio  async def process_item(item):     """处理单个项目"""     await asyncio.sleep(0.1)     return f"Processed {item}"  async def process_items_concurrently(items, max_concurrent=5):     """并发处理项目，控制并发数量"""     semaphore = asyncio.Semaphore(max_concurrent)      async def limited_process(item):         async with semaphore:             return await process_item(item)      tasks = [limited_process(item) for item in items]     results = await asyncio.gather(*tasks)     return results  # 使用示例 items = range(20) results = asyncio.run(process_items_concurrently(items))

import asyncio  async def fetch_with_timeout():     try:         # 设置超时         result = await asyncio.wait_for(             fetch_data("https://api.example.com"),             timeout=2.0         )         return result     except asyncio.TimeoutError:         print("请求超时")         return None  async def main_with_exception_handling():     try:         tasks = [fetch_data(url) for url in urls]         results = await asyncio.gather(*tasks, return_exceptions=True)          for result in results:             if isinstance(result, Exception):                 print(f"Error: {result}")             else:                 print(f"Success: {result}")     except Exception as e:         print(f"Main error: {e}")

# 创建新项目 poetry new my_project cd my_project  # 添加依赖 poetry add requests poetry add pytest --group dev  # 安装依赖 poetry install  # 运行脚本 poetry run python main.py

# pyproject.toml配置示例 [tool.poetry] name = "my-project" version = "0.1.0" description = "My awesome project" authors = ["Your Name"]  [tool.poetry.dependencies] python = "^3.9" fastapi = "^2.0.0" uvicorn = {extras = ["standard"], version = "^0.24.0"}  [tool.poetry.group.dev.dependencies] pytest = "^7.4.0" black = "^23.0.0"  [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api"

# 构建项目 poetry build                                                              # 发布到PyPI poetry publish                                                     # 使用私有仓库 poetry config repositories.myrepo https://pypi.myrepo.com/simple/ poetry publish --repository myrepo

# 依赖管理 poetry show               # 显示已安装的包 poetry show tree          # 显示依赖树 poetry update             # 更新依赖 poetry lock               # 更新lock文件  # 脚本管理 poetry run pytest         # 运行测试 poetry run black .        # 代码格式化  # 环境管理 poetry env list           # 列出虚拟环境 poetry env remove         # 删除虚拟环境 poetry shell              # 激活虚拟环境