Python 的修炼之道

import copy

# 深浅拷贝的坑
a = [[1, 2], [3, 4]]
b = copy.copy(a)        # 浅拷贝：外层新，内层还是同一个！
b[0][0] = 99            # a[0][0] 也变成 99 了（踩坑）
c = copy.deepcopy(a)    # 深拷贝：全新建（安全）

# 装饰器：日志/计时/鉴权（第一本元编程老朋友）
def timer(fn):
    def wrapper(*args, **kwargs):
        t0 = time.time(); r = fn(*args, **kwargs)
        print(f"{fn.__name__} 耗时 {time.time()-t0:.3f}s")
        return r
    return wrapper

@timer
def slow_query(): ...          # 一行装饰，处处计时

# 生成器：惰性求值，大文件不爆内存
def read_big(path):
    with open(path) as f:
        for line in f:
            yield line.strip()   # 一次一行，不全部加载

# 类型注解：让 Python 可静态检查
def calc_discount(price: float, rate: float = 0.9) -> float:
    return price * rate

【Python vs Java（底层差异）】
  Java：静态编译，运行时快，启动慢
  Python：解释执行，开发快，运行时慢
  Python 的「底层懂」= 懂解释器机制（GIL/内存模型），不是懂字节码

from collections import defaultdict, Counter, deque

# ❌ 反例：用 list 做「查重/查询」——O(n) 慢到爆
seen = []
if x not in seen: seen.append(x)      # 10 万条就卡

# ✅ 正解：set 做成员判断（O(1)）、dict 做查询（O(1)）
seen = set()
if x not in seen: seen.add(x)

# defaultdict：自动默认值（省 if 判断）
words = defaultdict(int)
for w in text.split(): words[w] += 1  # 不用先判断存在

# Counter：一行计数
cnt = Counter(text.split())           # {'hello': 3, 'world': 2}

# deque：队列场景（list 头部插入 O(n)，deque O(1)）
queue = deque(maxlen=100)             # 固定长度滑动窗口

# 内存优化：生成器 vs 列表
nums = [i*i for i in range(10_000_000)]   # 400MB！爆内存
nums = (i*i for i in range(10_000_000))   # 生成器：几乎不占内存

【复杂度速记（面试）】
  list 查找 O(n)  vs  dict/set 查找 O(1)
  list 尾插 O(1)  vs  list 头插 O(n)（用 deque）
  dict 依赖哈希——key 要可哈希（tuple 可以，list 不行）

# ① 协程 asyncio：IO 密集高并发（爬虫/API）
import asyncio, aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            return await r.text()

async def main():
    tasks = [fetch(f"https://example.com/p{i}") for i in range(1000)]
    results = await asyncio.gather(*tasks)   # 1000 个请求并发飞

asyncio.run(main())     # 单线程扛千级并发！

# ② 多进程：CPU 密集真并行
from concurrent.futures import ProcessPoolExecutor

def heavy_calc(n):       # 纯计算（不涉及 IO）
    return sum(i*i for i in range(n))

with ProcessPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(heavy_calc, [10**7]*8))   # 8 核全用上

# ③ 线程池：简单 IO 密集
from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=20) as pool:
    pool.map(download_one, urls)

【选型速查】
  场景                  选择
  网络/文件 IO 高并发    asyncio（协程）
  简单 IO 并发           ThreadPoolExecutor
  CPU 计算              ProcessPoolExecutor
  GIL 影响              CPU 密集绕不开，必须多进程

【对比 Java（第十四本）】
  Java：多线程真并行（无 GIL）——但复杂
  Python：协程极简 + 多进程补 CPU——开发效率高

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="账小灵 API")

class Order(BaseModel):
    id: int
    amount: float
    status: str = "pending"        # Pydantic 自动校验类型！

@app.get("/orders/{order_id}")
def get_order(order_id: int):
    if order_id not in orders:
        raise HTTPException(404, "订单不存在")     # 标准错误
    return orders[order_id]

@app.post("/orders")
def create_order(order: Order):    # 请求体自动校验
    orders[order.id] = order
    return {"ok": True, "id": order.id}

# 自动特性：
#   · 启动后 /docs 自动生成 Swagger 交互文档（白嫖文档！）
#   · 类型注解即校验（Pydantic）——省一半校验代码
#   · 支持 async def → 异步接口（第3站协程直接上）

【三大框架选型】
  Flask   ：轻量/小服务/灵活拼装
  Django  ：全家桶/管理后台/内容站
  FastAPI ：异步/API 优先/自动文档（新项目首选）

【WSGI vs ASGI】
  WSGI：同步标准（Flask/Django）——老而稳
  ASGI：异步标准（FastAPI）——支持 WebSocket/高并发

【RESTful 规范（快速版）】
  GET    /orders       查列表
  POST   /orders       创建
  GET    /orders/{id}  查详情
  PUT    /orders/{id}  全量更新
  DELETE /orders/{id}  删除
  → 资源用名词复数，动作交给 HTTP 方法

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class Product(Base):                       # ORM：类 = 表
    __tablename__ = "products"
    id    = Column(Integer, primary_key=True)
    name  = Column(String(100))
    price = Column(Integer)                # 单位：分

engine = create_engine("mysql+pymysql://user:pass@db/products",
                       pool_size=20)       # 连接池
Session = sessionmaker(bind=engine)

# 增删改查全用 Python 对象（不用拼 SQL）
s = Session()
p = Product(name="无线耳机", price=29900)
s.add(p); s.commit()
found = s.query(Product).filter_by(name="无线耳机").first()

# Redis 缓存：热点查询先查缓存
import redis
r = redis.Redis(host="redis", decode_responses=True)

def get_product(product_id: int):
    cached = r.get(f"product:{product_id}")
    if cached: return cached                 # 缓存命中
    p = s.query(Product).get(product_id)
    r.setex(f"product:{product_id}", 300, p.name)   # 5 分钟过期
    return p

# Alembic 迁移（表结构像 git 一样管）
#   alembic revision -m "add stock column"
#   alembic upgrade head        # 应用迁移

【事务与锁（第十四本老朋友）】
  事务：session.begin() ... commit()/rollback()
  乐观锁：version 字段；悲观锁：with_for_update()

【数据库最佳实践】
  N+1 查询：ORM 循环查库 → 用 joinedload/selectinload 预加载
  慢查询：EXPLAIN + 索引（第八本老朋友）
  连接池参数：pool_size + pool_timeout——耗尽=库扛不住

myproject/
├── pyproject.toml          # 依赖与项目配置（poetry 风格）
├── src/
│   └── mypkg/
│       ├── __init__.py
│       ├── main.py         # 入口
│       ├── models.py       # ORM 模型
│       ├── services/       # 业务逻辑
│       └── api/            # 路由
├── tests/                  # 测试（第7站）
├── alembic/                # 数据库迁移
└── .pre-commit-config.yaml # 提交前检查

【pyproject.toml（核心片段）】
[project]
name = "mypkg"
dependencies = [
    "fastapi>=0.110",
    "sqlalchemy>=2.0",
]
[tool.black] line-length = 100        # 格式化
[tool.ruff]  line-length = 100        # lint
[tool.mypy]  strict = true            # 类型检查

【工程化流程】
  写代码 → black 格式化 → ruff 检查 → mypy 类型检查
  → pytest 测试 → pre-commit 卡关 → 提交 → CI/CD（第10站）

【依赖管理对比】
  pip + requirements.txt：能用但裸（版本不锁定）
  poetry / uv：现代方案——锁定 + 可复现 + 打包一条龙
  生产部署：poetry export → 锁定版本安装（可复现！）

import pytest
from mypkg.services import calc_discount

# fixture：准备数据 + 自动清理
@pytest.fixture
def sample_order():
    return {"id": 1, "amount": 100.0, "status": "pending"}

# 参数化：一组数据测多组用例
@pytest.mark.parametrize("price,rate,expected", [
    (100, 0.9, 90.0),      # 正常折扣
    (0, 0.9, 0.0),         # 边界：零元
    (100, 1.5, 150.0),     # 反向折扣（超价）
])
def test_discount(price, rate, expected):
    assert calc_discount(price, rate) == expected

# mock：不碰真网络
def test_fetch_user(mocker):
    mocker.patch("mypkg.services.requests.get",
                 return_value=Mock(status_code=200, json=lambda: {"id": 1}))
    user = fetch_user(1)             # 不走真实网络！
    assert user["id"] == 1

# 覆盖率：pytest --cov=mypkg --cov-report=html
# → 打开 htmlcov 看哪行没测到

【调试三板斧】
  ① breakpoint()：插断点，进 pdb 交互
  ② 日志：log.info/error——生产排障的主武器
  ③ cProfile：性能分析——慢在哪一函数（第8站）

【测试金字塔（第二本老朋友）】
  单测（函数级，多）→ 集成测（接口级，中）→ E2E（少）

# ① 先测量：cProfile 找热点
# python -m cProfile -s cumulative app.py
# → 发现：90% 时间花在 discount_chain() 的重复计算上

# ② lru_cache：相同输入只算一次
from functools import lru_cache

@lru_cache(maxsize=1024)
def discount_chain(level: int) -> float:
    return 0.9 ** level          # 重复调用直接命中缓存

# ③ 循环优化：提升常量 + 局部变量
def total_price(items, rate):
    local_round = round          # 局部化内置函数（快 20%）
    total = 0.0
    for it in items:
        total += it.price * rate
    return local_round(total, 2)

# ④ numba：数值计算 JIT 加速
from numba import jit
@jit(nopython=True)
def heavy_loop(n):               # 接近 C 的速度
    s = 0
    for i in range(n):
        s += i * i
    return s

# ⑤ 终极：换数据结构/算法（复杂度降级）
#   list 查找 O(n) → set/dict O(1)（第2站）

【优化优先级（从便宜到贵）】
  1. 算法/数据结构（最赚，先做！）
  2. 内置函数 + 局部变量 + 缓存（lru_cache）
  3. 多进程（CPU 密集）
  4. numba / Cython / C 扩展（最后手段）
  → 别一上来就上 C 扩展！先看复杂度

# ① 输入校验（Pydantic 自动）
class OrderIn(BaseModel):
    amount: float = Field(gt=0, lt=1_000_000)   # 金额范围自动校验
    note: str = Field(max_length=200)

# ② 依赖审计（CI 里必跑）
# pip-audit → 扫出有 CVE 的依赖 → 红牌不许发布

# ③ 配置管理：环境变量 + 校验
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
    database_url: str
    redis_url: str
    api_key: str            # 从环境变量读，绝不打进代码
    model_config = SettingsConfigDict(env_file=".env")   # .env 不进 Git！
settings = Settings()

# ④ 日志：分级 + 结构化（JSON 给 ELK 吃）
import logging, json
logger = logging.getLogger("order")
logger.info("创建订单", extra={"order_id": 1, "amount": 99.9})
# → 输出 JSON 日志 → Filebeat 采集 → ELK（第十二本闭环）

# ⑤ 限流（slowapi）
from slowapi import Limiter
limiter = Limiter(key_func=lambda: "global")
@app.get("/api/orders")
@limiter.limit("100/minute")        # 每分钟 100 次
def list_orders(request: Request): ...

【安全红线（第七本老朋友在 Python 的落地）】
  SQL 注入：永远用 ORM 参数化，不拼字符串
  密钥管理：.env + 环境变量 + 密钥管理系统（绝不入 Git）
  依赖漏洞：pip-audit 每周扫
  日志脱敏：不打完整手机号/密码（第八本脱敏）

【上线前 Checklist（第2本老朋友）】
  □ 输入全校验（Pydantic）   □ 依赖已审计（pip-audit）
  □ 密钥在环境变量（不在代码） □ 日志分级且脱敏
  □ 限流已开启               □ 错误处理统一（不裸奔堆栈）

【Dockerfile（多阶段构建）】
FROM python:3.12-slim AS base
WORKDIR /app
COPY pyproject.toml poetry.lock .
RUN pip install --no-cache-dir -r requirements.txt   # 锁定版本
COPY src/ src/
EXPOSE 8000
CMD ["uvicorn", "mypkg.main:app", "--host", "0.0.0.0",
     "--port", "8000", "--workers", "4"]

【进程模型】
  Uvicorn --workers 4：4 个进程分担（配合多核）
  Gunicorn -w 4：Flask/Django 用（同步世界）
  → 单进程扛不住？先加 workers，再考虑异步/缓存

【CI/CD 流水线（GitHub Actions 简版）】
  git push
   → pytest（测试全绿？）
   → ruff + mypy（规范/类型过了？）
   → pip-audit（依赖没漏洞？）
   → docker build & push（构建镜像）
   → kubectl rollout restart（部署到 K8s，第十三本）

【Serverless 方向（了解）】
  FastAPI → AWS Lambda / 腾讯云函数 / Google Cloud Run
  按调用计费、自动伸缩——轻量 API 的现代归宿

【学习路线（3 个月进阶）】
  第1月：语言进阶 + 数据结构 + 并发（本册 1-3 站）
  第2月：FastAPI + 数据库 + 工程化（4-6 站）
  第3月：测试 + 性能 + 安全 + 部署（7-10 站）+ 一个完整项目
  → 项目推荐：订单系统 / 爬虫平台 / AI 服务（账小灵！）