---

前言

学Python，入门很简单。但很多人学了一两年，还是卡在同一个地方：能写出"能跑"的代码，但写不出"好的"代码。

具体表现是：

• 看别人代码里的 @ 符号，知道叫装饰器，但自己不会写
• 知道有 yield 这个关键字，但不知道什么时候用
• 听说异步能提升性能，写出来却比同步还慢
• 选框架靠"感觉"，不知道为什么选 FastAPI 还是 Django

这四个问题，是 Python 开发者从"入门"到"进阶"最常见的拦路虎。这篇文章用实际代码一次讲清楚。

---

第一道坎：装饰器，到底在干什么

本质：函数也是对象

import functools
import time

def timer(func):
    """计时装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        print(f"{func.__name__} 执行耗时: {end - start:.4f}s")
        return result
    return wrapper

@timer
def slow_function(n):
    return sum(range(n))

result = slow_function(10_000_000)
# slow_function 执行耗时: 0.3241s

带参数的装饰器

def retry(max_times=3, delay=1.0):
    """失败自动重试装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_times + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_times:
                        raise
                    print(f"第{attempt}次失败: {e}，{delay}s后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_times=3, delay=0.5)
def call_api(url: str) -> dict:
    import httpx
    response = httpx.get(url, timeout=5)
    response.raise_for_status()
    return response.json()

一句话记住：装饰器就是用一个函数包裹另一个函数，不改原代码的情况下给它加功能。

---

第二道坎：生成器，懒才是美德

为什么需要生成器

# 危险：100万行全装进内存
def read_logs_bad(filename):
    with open(filename) as f:
        return f.readlines()

# 正确：生成器按需读取，内存恒定
def read_logs_good(filename):
    with open(filename) as f:
        for line in f:
            yield line.strip()

for line in read_logs_good("huge.log"):
    process(line)

生成器表达式

numbers = range(1_000_000)

# 列表推导式：占 ~8MB
squares_list = [x**2 for x in numbers]

# 生成器表达式：占 ~200字节
squares_gen = (x**2 for x in numbers)

total = sum(squares_gen)  # 用法一样，内存差几万倍

数据流水线

def read_csv(filename):
    with open(filename) as f:
        next(f)  # 跳过表头
        for line in f:
            yield line.strip().split(',')

def filter_active(rows):
    for row in rows:
        if row[2] == 'active':
            yield row

def format_output(rows):
    for row in rows:
        yield {'name': row[0], 'email': row[1]}

# 三个生成器串联，每次只处理一行
pipeline = format_output(filter_active(read_csv("users.csv")))
for user in pipeline:
    send_email(user)

一句话记住：生成器是会暂停的函数，用 yield 一次产出一个值——懒加载，省内存。

---

第三道坎：异步编程，协程不是多线程

最常见的误解

很多人以为"异步 = 多线程 = 更快"，然后用 asyncio 跑 CPU 密集任务，结果比同步还慢。

记住一句话：asyncio 解决的是 I/O 等待问题，不是 CPU 计算问题。

CPU密集型（大量计算）→ multiprocessing（多进程）
I/O密集型（网络/文件）→ asyncio（协程）

异步的真实威力

import asyncio
import httpx

# 同步：顺序请求，总时间 = 所有请求时间之和（约10秒）
def sync_fetch(urls):
    with httpx.Client() as client:
        return [client.get(url).status_code for url in urls]

# 异步：并发请求，总时间 ≈ 最慢那一个（约1秒）
async def async_fetch(urls):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
    return [r.status_code for r in responses]

urls = ["https://httpbin.org/delay/1"] * 10
asyncio.run(async_fetch(urls))

核心用法速查

import asyncio

# 并发运行多个协程
async def main():
    await asyncio.gather(
        say("Hello", delay=1),
        say("World", delay=2),
        say("Python", delay=0.5),
    )
    # 总耗时约 2 秒，不是 3.5 秒

# 带超时控制（Python 3.11+）
async def fetch_safe(url):
    try:
        async with asyncio.timeout(5.0):
            async with httpx.AsyncClient() as client:
                return await client.get(url)
    except TimeoutError:
        return None

asyncio.run(main())

一句话记住：asyncio 是单线程的高效时间管理——主动让出等待时间，同时处理多个 I/O。

---

第四道坎：框架选型，不要靠感觉

2026年三大框架的真实定位

框架	定位	性能	内置功能	适合场景
FastAPI	异步API优先	⭐⭐⭐⭐⭐	⭐⭐	AI服务、微服务、高并发API
Django	全功能MVC	⭐⭐⭐	⭐⭐⭐⭐⭐	全栈Web、内容管理、企业后台
Flask	极简微框架	⭐⭐⭐⭐	⭐	轻量服务、高度定制、快速原型

FastAPI 实战：给 AI 模型包一个接口

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
import httpx

class ChatRequest(BaseModel):
    message: str
    max_tokens: int = 512

class ChatResponse(BaseModel):
    reply: str
    model: str

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.client = httpx.AsyncClient(timeout=30.0)
    yield
    await app.state.client.aclose()

app = FastAPI(title="AI Chat API", lifespan=lifespan)

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await app.state.client.post(
            "https://api.your-model.com/chat",
            json={"message": request.message}
        )
        response.raise_for_status()
        data = response.json()
        return ChatResponse(reply=data["content"], model=data["model"])
    except httpx.HTTPStatusError as e:
        raise HTTPException(status_code=e.response.status_code, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "ok"}

# 启动，自动生成文档：http://localhost:8000/docs
uv run uvicorn main:app --reload

Django 适合什么：内置 Admin 零代码后台

from django.contrib import admin
from .models import Student

@admin.register(Student)
class StudentAdmin(admin.ModelAdmin):
    list_display = ['name', 'class_name', 'grade', 'status']
    list_filter = ['status', 'class_name']
    search_fields = ['name', 'student_id']
    # 5行配置 = 功能完整的后台管理界面

选型决策树

你的项目是什么类型？
│
├── 主要是 API（前后端分离 / AI 服务 / 微服务）→ FastAPI ✅
│
├── 全栈 Web（后台管理 / 内容管理 / 用户系统）→ Django ✅
│
└── 需要高度定制 / 极简 / 快速验证想法 → Flask ✅

附加判断：
- 有大量 LLM 调用？→ FastAPI（异步优势明显）
- 需要复杂权限体系？→ Django（内置最完善）
- 团队已有 Django 积累？→ 继续用，别因为"FastAPI更新"就换

---

把四个概念串起来：综合实战案例

import asyncio, csv, functools, time, httpx

# 装饰器：计时
def timer(func):
    @functools.wraps(func)
    async def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = await func(*args, **kwargs)
        print(f"{func.__name__} 耗时: {time.perf_counter() - start:.2f}s")
        return result
    return wrapper

# 生成器：懒加载 CSV
def read_csv_rows(filepath: str):
    with open(filepath, newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield row

# 异步：并发调用 AI API
async def analyze_row(client: httpx.AsyncClient, row: dict) -> dict:
    response = await client.post(
        "https://api.ai-service.com/analyze",
        json={"text": row["content"]},
        timeout=10.0
    )
    return {**row, "analysis": response.json()["result"]}

# 整合：流水线批量处理
@timer
async def process_file(filepath: str, batch_size: int = 10):
    results = []
    rows = list(read_csv_rows(filepath))

    async with httpx.AsyncClient() as client:
        for i in range(0, len(rows), batch_size):
            batch = rows[i:i + batch_size]
            batch_results = await asyncio.gather(
                *[analyze_row(client, row) for row in batch]
            )
            results.extend(batch_results)
            print(f"进度: {min(i + batch_size, len(rows))}/{len(rows)}")

    return results

asyncio.run(process_file("data.csv", batch_size=20))

---

总结

概念	核心思想	典型用途
装饰器	函数包裹函数，不改原代码加功能	日志、计时、鉴权、重试
生成器	yield 暂停，按需产出数据	大文件处理、数据流水线
异步	单线程主动让出等待，并发 I/O	网络请求、数据库查询
FastAPI	异步优先、类型安全、自动文档	API 服务、AI 接口
Django	全功能、内置 Admin、ORM 完善	全栈 Web、企业后台

真正掌握一个概念的标志，不是能背出定义，而是写代码时自然而然地想到用它。

---

💬 这四个里，你觉得哪个最难理解？评论区聊聊～ 👍 觉得有收获的话点个赞！

---