Python 商业场景训练营

Python 真正常见的商业价值不是“语法简单”，而是能快速把文件处理、接口调用、数据清洗、自动化任务、Web API、AI 调用串成稳定可维护的工具或服务。

一句话主线：

Python 项目从脚本走向生产，靠的是环境隔离、清晰分层、异常处理、日志、类型标注、测试、配置管理和可观测排查。

训练目标

学完本页，你要能做到：

写一个可维护的批处理脚本，而不是一次性脚本。
写一个 FastAPI 服务，知道请求、校验、业务层、数据层怎么分。
解释 list、dict、引用、可变对象、GIL、线程、进程、asyncio 的原理边界。
处理文件、CSV、JSON、数据库和外部 HTTP 接口。
加日志、测试、配置和错误码。
在面试里讲清楚 Python 为什么适合脚本、Web、数据和 AI 工程。

商业场景一：医疗数据 CSV 入库前清洗

假设每天从医院系统导出就诊 CSV，需要清洗字段、校验必填项、脱敏患者 ID，然后输出合格数据和错误报告。

mermaid

flowchart TD
    A["读取 CSV 文件"] --> B["逐行解析"]
    B --> C["字段标准化"]
    C --> D{"必填和格式是否通过"}
    D -- "通过" --> E["患者 ID 脱敏"]
    E --> F["写入 clean.csv"]
    D -- "不通过" --> G["写入 error.csv"]
    F --> H["输出处理统计"]
    G --> H

可运行 Demo：

python

import csv
import hashlib
from pathlib import Path


def hash_patient_id(patient_id: str) -> str:
    return hashlib.sha256(patient_id.encode("utf-8")).hexdigest()


def normalize_row(row: dict[str, str]) -> dict[str, str]:
    return {
        "patient_id_hash": hash_patient_id(row["patient_id"].strip()),
        "visit_time": row["visit_time"].strip(),
        "dept_code": row["dept_code"].strip().upper(),
        "diagnosis": row.get("diagnosis", "").strip(),
    }


def validate(row: dict[str, str]) -> list[str]:
    errors: list[str] = []
    if not row.get("patient_id"):
        errors.append("patient_id required")
    if not row.get("visit_time"):
        errors.append("visit_time required")
    if not row.get("dept_code"):
        errors.append("dept_code required")
    return errors


def clean_csv(input_file: Path, clean_file: Path, error_file: Path) -> None:
    with input_file.open("r", encoding="utf-8", newline="") as source, \
            clean_file.open("w", encoding="utf-8", newline="") as clean, \
            error_file.open("w", encoding="utf-8", newline="") as error:

        reader = csv.DictReader(source)
        clean_writer = csv.DictWriter(clean, fieldnames=[
            "patient_id_hash", "visit_time", "dept_code", "diagnosis"
        ])
        error_writer = csv.DictWriter(error, fieldnames=[
            *(reader.fieldnames or []), "errors"
        ])

        clean_writer.writeheader()
        error_writer.writeheader()

        for row in reader:
            errors = validate(row)
            if errors:
                row["errors"] = ";".join(errors)
                error_writer.writerow(row)
                continue
            clean_writer.writerow(normalize_row(row))


if __name__ == "__main__":
    clean_csv(Path("visit.csv"), Path("clean.csv"), Path("error.csv"))

为什么这样写：

写法	原因
`Path`	比字符串路径更清晰，跨平台更好
`with`	自动关闭文件，避免资源泄露
`DictReader`	按字段名读，避免列顺序变化导致错误
单独 `validate`	校验逻辑可测试、可复用
单独 `normalize_row`	清洗逻辑和文件读写解耦

商业场景二：FastAPI 资产查询接口

Python Web 项目不要把所有逻辑写进路由函数。路由只负责接收请求和返回响应，业务规则放 service，数据访问放 repository。

mermaid

flowchart TD
    A["HTTP 请求"] --> B["FastAPI 路由"]
    B --> C["Pydantic 参数校验"]
    C --> D["Service 处理业务规则"]
    D --> E["Repository 查询数据库"]
    E --> F["返回领域数据"]
    F --> G["组装响应"]

最小 Demo：

python

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()


class AssetResponse(BaseModel):
    asset_code: str
    name: str
    owner_dept: str


ASSETS = {
    "DATASET_001": {
        "asset_code": "DATASET_001",
        "name": "门诊就诊明细",
        "owner_dept": "信息科",
    }
}


def get_asset_by_code(asset_code: str) -> AssetResponse:
    data = ASSETS.get(asset_code)
    if data is None:
        raise HTTPException(status_code=404, detail="asset not found")
    return AssetResponse(**data)


@app.get("/assets/{asset_code}", response_model=AssetResponse)
def get_asset(asset_code: str):
    return get_asset_by_code(asset_code)

运行：

bash

pip install fastapi uvicorn
uvicorn main:app --reload --port 8000

访问：

bash

curl http://127.0.0.1:8000/assets/DATASET_001

Python 对象和引用原理

Python 变量不是盒子本身，而是指向对象的名字。

mermaid

flowchart TD
    A["变量 a"] --> C["list 对象 [1, 2]"]
    B["变量 b"] --> C
    C --> D["append 会修改同一个对象"]

示例：

python

a = [1, 2]
b = a
b.append(3)
print(a)  # [1, 2, 3]

这就是为什么可变对象要小心传递。函数里修改 list、dict 会影响外部对象。想避免副作用，可以复制：

python

def add_item(items: list[int]) -> list[int]:
    new_items = items.copy()
    new_items.append(100)
    return new_items

GIL、线程、进程和 asyncio

Python 初学者常问：为什么开了多线程 CPU 还是上不去？关键是 CPython 有 GIL。

GIL 可以粗略理解为：同一时刻一个进程里只有一个线程执行 Python 字节码。它简化了解释器内存管理，但限制了 CPU 密集型多线程并行。

mermaid

flowchart TD
    A["任务类型"] --> B{"主要耗时在哪里"}
    B -- "等待网络/磁盘/数据库" --> C["线程或 asyncio"]
    B -- "大量 CPU 计算" --> D["多进程或 C 扩展"]
    B -- "需要高并发 HTTP" --> E["asyncio + async Web 框架"]

选择建议：

场景	推荐
批量请求外部接口	`asyncio` 或线程池
读写大量文件	线程池或异步 IO
图片压缩、加密计算	多进程
FastAPI 高并发接口	`async def` + 异步数据库客户端
简单定时脚本	普通同步代码即可

异步 Demo：

python

import asyncio
import httpx


async def fetch(client: httpx.AsyncClient, url: str) -> int:
    response = await client.get(url, timeout=10)
    return response.status_code


async def main() -> None:
    urls = [
        "https://example.com",
        "https://example.com",
        "https://example.com",
    ]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*(fetch(client, url) for url in urls))
    print(results)


if __name__ == "__main__":
    asyncio.run(main())

asyncio 并不是让 CPU 变多，而是在等待网络返回时切换去处理其他任务。

日志和异常排查

生产脚本最怕只 print。日志要包含时间、级别、业务主键和异常堆栈。

python

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)

logger = logging.getLogger("asset_cleaner")


def handle_asset(asset_code: str) -> None:
    try:
        logger.info("start handle asset_code=%s", asset_code)
        # business logic
        logger.info("finish handle asset_code=%s", asset_code)
    except Exception:
        logger.exception("handle asset failed asset_code=%s", asset_code)
        raise

排查流程：

mermaid

flowchart TD
    A["Python 程序报错"] --> B["看异常类型"]
    B --> C["看文件名和行号"]
    C --> D["看调用栈上一层业务参数"]
    D --> E{"是否依赖外部服务"}
    E -- "是" --> F["检查超时、状态码、重试和降级"]
    E -- "否" --> G["检查输入数据、类型、空值"]
    F --> H["补日志和测试"]
    G --> H

测试 Demo

核心业务函数必须能脱离文件和网络单独测试。

python

from cleaner import normalize_row


def test_normalize_row_should_upper_dept_code():
    row = {
        "patient_id": "p001",
        "visit_time": "2026-07-06 10:00:00",
        "dept_code": "opd",
        "diagnosis": "cold",
    }

    result = normalize_row(row)

    assert result["dept_code"] == "OPD"
    assert result["patient_id_hash"] != "p001"

为什么要测试：

数据清洗规则经常改，没有测试很容易改坏旧逻辑。
测试让你敢重构。
面试和真实项目都会关注代码是否可维护，而不是只会跑。

常见坑

坑	后果	正确做法
全局安装依赖	项目互相污染	每个项目使用虚拟环境
默认参数写 `[]`	多次调用共享同一对象	用 `None` 再创建
捕获异常不记录堆栈	线上无法定位	用 `logger.exception`
所有代码写一个文件	难测试、难维护	拆分 route/service/repository
CPU 密集任务用线程	利用不了多核	用多进程或专用计算库
请求外部接口无超时	线程卡死、任务堆积	必须设置 timeout

面试标准回答

Python 为什么适合脚本和自动化？

Python 语法简洁，标准库和第三方库丰富，处理文件、JSON、CSV、HTTP、Excel、数据库都很方便。商业项目中常用它做批处理、数据清洗、接口巡检、自动化测试、AI 服务胶水层。但生产脚本也要有虚拟环境、日志、异常处理、配置和测试。

GIL 是什么？

GIL 是 CPython 的全局解释器锁，同一进程同一时刻通常只有一个线程执行 Python 字节码。它让解释器内存管理更简单，但限制 CPU 密集型多线程并行。IO 密集任务仍然可以用线程或 asyncio，因为等待 IO 时可以切换任务；CPU 密集任务更适合多进程。

Python 中 list 和 dict 为什么常用？

list 适合保存有序集合，dict 适合按 key 快速查找。dict 基于哈希表思想，平均查询接近 O(1)。但 key 必须可哈希，可变对象如 list 不能作为 key。

关联知识点

知识点	继续学习
Python 主线	Python 从零到生产级掌握
环境	环境与包管理
数据结构	数据结构
并发异步	并发与异步
Web API	Web API 开发
测试	测试
面试	Python 面试题

Python 商业场景训练营 ​

训练目标 ​

商业场景一：医疗数据 CSV 入库前清洗 ​

商业场景二：FastAPI 资产查询接口 ​

Python 对象和引用原理 ​

GIL、线程、进程和 asyncio ​

日志和异常排查 ​

测试 Demo ​

常见坑 ​

面试标准回答 ​

关联知识点 ​

Python 商业场景训练营

训练目标

商业场景一：医疗数据 CSV 入库前清洗

商业场景二：FastAPI 资产查询接口

Python 对象和引用原理

GIL、线程、进程和 asyncio

日志和异常排查

测试 Demo

常见坑

面试标准回答

关联知识点