Python 项目实践

学完 Python 基础后，下一步不是继续堆语法，而是学会把代码组织成一个“别人能看懂、能运行、能测试、能部署、出问题能排查”的项目。

很多初学者会把所有代码写在一个 .py 文件里。练习阶段可以，商业项目不行。功能一多，单文件会把配置、参数校验、业务规则、数据库访问、日志、异常处理、启动逻辑全部混在一起，最后变成“能跑但没人敢改”的代码。

学习目标

学完本页你应该能回答：

一个 Python 项目为什么要分层。
api、service、repository、model、core 分别放什么。
为什么配置不能写死在代码里。
requirements.txt 和 pyproject.toml 解决什么问题。
日志、异常、测试、脚本、Docker 在项目中怎么配合。
商业项目从开发到上线要经过哪些步骤。
线上出问题时，如何通过日志、健康检查、配置和测试定位问题。

项目实践到底练什么

项目实践不是“写一个看起来像项目的目录”，而是训练工程能力。

能力	解决什么问题
目录结构	让代码职责清晰，不同能力放在不同位置
配置管理	开发、测试、生产环境用不同配置，不泄露密钥
依赖管理	别人拉代码后能复现同样环境
分层设计	业务逻辑可复用、可测试、可维护
日志和异常	出问题能定位，不把堆栈直接暴露给用户
测试	修改代码后能自动验证核心行为
脚本和命令	初始化、导入、清洗、巡检可以重复执行
部署	程序能稳定运行在服务器或容器中

一个项目从需求到上线的完整过程

零基础最容易犯的错是：一上来就写代码。商业项目不是“代码写完就结束”，而是要经过需求澄清、建模、接口设计、实现、测试、部署、监控和迭代。

mermaid

flowchart TD
    A["明确业务目标"] --> B["拆分核心流程"]
    B --> C["设计数据模型"]
    C --> D["设计 API 和脚本入口"]
    D --> E["确定配置和依赖"]
    E --> F["实现 Service Repository Client"]
    F --> G["补单元测试和接口测试"]
    G --> H["本地运行和联调"]
    H --> I["构建镜像或部署包"]
    I --> J["测试环境验证"]
    J --> K["生产发布"]
    K --> L["观察日志 指标 告警"]
    L --> M["根据线上反馈迭代"]

以“医疗数据资产导入”为例，业务目标不是“写一个导入接口”，而是：

用户能提交 CSV 导入任务。
系统能校验字段、记录错误行。
成功数据能形成资产目录。
失败后能定位、重试、审计。
大文件不能拖垮 HTTP 请求。
敏感字段不能进入日志。

所以项目结构、事务、日志、测试、部署都要围绕这些目标设计。

从单文件到项目的演进

最开始你可能写一个单文件脚本：

python

import csv

with open("assets.csv", encoding="utf-8") as file:
    for row in csv.DictReader(file):
        print(row["asset_code"], row["asset_name"])

这个脚本能跑，但无法支撑真实项目：

问题	为什么会出事
文件名写死	换文件要改代码
没有校验	脏数据直接进入后续流程
没有日志	失败不知道哪一行
没有测试	改一处可能破坏导入
没有分层	Web、CLI、定时任务无法复用

演进路径：

mermaid

flowchart TD
    A["单文件脚本"] --> B["抽函数"]
    B --> C["拆模块"]
    C --> D["增加配置和日志"]
    D --> E["增加测试"]
    E --> F["封装 CLI 或 Web API"]
    F --> G["接入数据库和部署"]

不要为了“架构感”一开始就过度设计，但当代码开始承担业务责任时，就必须把输入、业务、数据访问、异常、日志和测试拆开。

目录	放什么	不建议放什么
`api`	HTTP 路由、请求参数、响应模型	大量业务规则、SQL
`schemas`	Pydantic 入参和出参模型	数据库连接
`services`	业务规则、流程编排、事务边界	HTTP Header 解析
`repositories`	数据库读写、SQL/ORM 查询	复杂业务判断
`models`	ORM 实体或领域对象	接口返回格式
`clients`	调用外部 HTTP、AI、第三方系统	主业务数据持久化
`core`	配置、异常、日志、安全等基础能力	具体业务流程
`tests`	单元测试、接口测试、集成测试	生产代码
`scripts`	初始化、导入、巡检、一次性任务	Web 请求处理

文件划分为什么这么分

文件划分不是为了“目录多显得专业”，而是为了让变化范围可控。

变化	应该主要改哪里
HTTP 路径改了	`api`
请求字段改了	`schemas`
业务规则改了	`services`
SQL 或 ORM 查询改了	`repositories`
数据库表结构改了	`models` 和 `migrations`
外部系统地址或调用方式改了	`clients`
日志、配置、异常格式改了	`core`
导入脚本参数改了	`scripts` 或 CLI

如果一个需求导致你同时改很多不相关目录，通常说明分层边界不清晰。比如“资产来源系统增加 PACS”应该主要改业务校验或枚举，不应该顺手改数据库连接、日志格式和 Dockerfile。

import 路径和运行方式

项目拆成包后，运行方式会影响导入路径。推荐从项目根目录运行模块：

bash

python -m app.main

而不是进入 app/ 目录后运行：

bash

python main.py

后者容易导致 ModuleNotFoundError 或相对导入混乱。

FastAPI 启动：

bash

uvicorn app.main:app --reload

app.main:app 的含义是：导入 app/main.py 里的 app 对象。理解这一点后，很多“为什么启动找不到模块”的问题就能定位到工作目录、包名或解释器配置。

分层调用流程

mermaid

flowchart TD
    A["客户端请求"] --> B["API 层"]
    B --> C["Schema 校验"]
    C --> D["Service 业务编排"]
    D --> E["Repository 数据访问"]
    D --> F["Client 外部服务"]
    E --> G["数据库"]
    F --> H["AI 或第三方系统"]
    D --> I["返回业务结果"]
    I --> B

为什么 Service 不应该直接处理 HTTP？因为业务规则可能被多个入口复用：HTTP 接口、命令行脚本、定时任务、消息消费者都可能调用同一段业务逻辑。如果业务写死在 API 层，其他入口只能复制。

不分层会怎样

反例：

python

def create_asset(request):
    name = request.json["name"]
    source = request.json["source_system"]
    if source not in ["HIS", "LIS"]:
        return {"code": "ERROR", "message": "来源系统不支持"}
    sql = f"INSERT INTO assets(name, source_system) VALUES('{name}', '{source}')"
    db.execute(sql)
    ai_result = requests.post("http://ai/analyze", json={"name": name})
    print("创建资产", name, ai_result.text)
    return {"code": "SUCCESS"}

问题：

SQL 拼接有注入风险。
业务校验、数据库、AI 调用混在一起。
没有事务，写库成功但 AI 调用失败时状态不清晰。
没有超时，AI 接口慢会拖住请求。
不能单独测试“来源系统校验”。
日志用 print，线上不好查。

分层后，同一个业务会变成：

python

class AssetService:
    def __init__(self, repository, ai_client):
        self.repository = repository
        self.ai_client = ai_client

    def create_asset(self, request):
        self._check_source_system(request.source_system)
        asset = self.repository.save(request)
        self.ai_client.submit_asset_analyze(asset.id, timeout_seconds=3)
        return asset

    def _check_source_system(self, source_system: str):
        if source_system not in {"HIS", "LIS", "PACS", "EMR"}:
            raise ValueError("来源系统不支持")

这样 Service 可以被 API 调用，也可以被测试直接调用。

配置管理

配置不要写死在代码里。

错误做法：

python

DATABASE_URL = "mysql://root:123456@localhost:3306/app"
SECRET_KEY = "abc123"

问题：

密码会进入 Git。
开发、测试、生产环境无法区分。
改配置必须改代码再重新发版。

配置加载顺序和校验

商业项目建议把配置加载做成明确流程：

mermaid

flowchart TD
    A["读取默认配置"] --> B["读取环境变量"]
    B --> C["读取 .env 或配置中心"]
    C --> D["转换类型"]
    D --> E{"必填配置是否存在"}
    E -->|否| F["启动失败并提示缺失项"]
    E -->|是| G["应用启动"]

为什么启动时就要失败？因为缺少数据库连接、密钥、模型 Key 这类关键配置时，继续运行只会把错误延后到用户请求或定时任务中，排查成本更高。

使用 Pydantic Settings 的思路：

python

from pydantic import Field
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_env: str = "dev"
    database_url: str = Field(min_length=1)
    secret_key: str = Field(min_length=16)
    log_level: str = "INFO"
    ai_api_key: str | None = None


settings = Settings()

配置分类：

配置	示例	建议
普通配置	`LOG_LEVEL`、`APP_ENV`	可放 `.env.example`
敏感配置	`SECRET_KEY`、`AI_API_KEY`	环境变量或密钥系统
连接配置	`DATABASE_URL`、`REDIS_URL`	启动时校验
业务阈值	`IMPORT_BATCH_SIZE`	有默认值，允许调整

.env.example 只放字段名和示例，不放真实密钥。

依赖管理

小项目可以用 requirements.txt：

text

fastapi==0.115.0
uvicorn==0.30.0
pytest==8.2.0
sqlalchemy==2.0.31

安装：

bash

pip install -r requirements.txt

新项目更推荐逐步学习 pyproject.toml，它可以统一描述项目元信息、依赖、测试配置、格式化配置。

示例：

toml

[project]
name = "asset-api"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
  "fastapi==0.115.0",
  "uvicorn==0.30.0",
  "sqlalchemy==2.0.31",
]

[project.optional-dependencies]
dev = [
  "pytest==8.2.0",
  "pytest-cov==5.0.0",
]

为什么要锁版本？因为不锁版本时，今天安装的是 A 版本，明天可能变成 B 版本。线上故障里很常见的一类就是依赖升级导致行为变化。

开发依赖、生产依赖和锁文件

依赖不是都要进生产环境。

类型	例子	是否生产需要
运行依赖	`fastapi`、`sqlalchemy`、`pymysql`	需要
开发依赖	`pytest`、`ruff`、`mypy`	通常不需要
构建依赖	打包工具、生成工具	构建阶段需要

如果用 requirements.txt，常见拆分：

text

requirements.txt
requirements-dev.txt

requirements-dev.txt：

text

-r requirements.txt
pytest==8.2.0
ruff==0.5.0
mypy==1.10.0

新项目用 pyproject.toml 时，可以把开发工具放到 optional dependencies。这样本地开发装全量，生产镜像只装运行依赖。

依赖管理的目标是：

本地、测试、生产版本一致。
安全漏洞能定位到具体包。
新人拉代码能快速跑起来。
CI 能复现构建。

代码质量工具

项目里常见工具：

工具	作用
`ruff`	代码风格和常见错误检查
`black`	自动格式化
`mypy` / `pyright`	类型检查
`pytest`	自动测试
`coverage`	测试覆盖率

pyproject.toml 示例：

toml

[tool.ruff]
line-length = 100

[tool.pytest.ini_options]
testpaths = ["tests"]

[tool.mypy]
python_version = "3.11"
warn_return_any = true
disallow_untyped_defs = false

不要把工具当成形式主义。它们的价值是把低级错误尽量提前到提交前、CI 阶段发现，而不是让线上用户帮你发现。

命令行入口和脚本

项目里经常需要脚本，例如初始化数据库、导入文件、修复历史数据、巡检任务。脚本必须可重复执行、可记录日志、失败可定位。

脚本入口：

python

def main():
    print("开始导入资产")


if __name__ == "__main__":
    main()

为什么要加 if __name__ == "__main__"？因为文件被其他模块导入时不应该自动执行任务，否则测试或复用时会产生副作用。

导入脚本示例：

python

import csv
import logging

logger = logging.getLogger(__name__)


def import_assets(csv_path: str, service):
    success = 0
    failed = 0
    with open(csv_path, encoding="utf-8") as file:
        reader = csv.DictReader(file)
        for row in reader:
            try:
                service.create_asset(row)
                success += 1
            except Exception:
                failed += 1
                logger.exception("导入资产失败：%s", row.get("name"))
    logger.info("导入完成 success=%s failed=%s", success, failed)

商业脚本不要只追求跑完，还要记录成功数、失败数、失败原因和可重试依据。

日志和 TraceId

正式项目少用 print()，用 logging。Web API 推荐每个请求生成或透传 request_id，所有日志都带上它。

mermaid

flowchart TD
    A["请求进入"] --> B["生成 request_id"]
    B --> C["API 日志"]
    C --> D["Service 日志"]
    D --> E["Repository SQL 日志"]
    D --> F["外部接口日志"]
    E --> G["按 request_id 串联排查"]
    F --> G

日志应该记录：

字段	作用
时间	判断问题发生时间
级别	区分正常、警告、错误
request_id	串联一次请求
user_id/task_id	定位用户或任务
path/function	定位代码入口
cost_ms	判断耗时
error_code	聚合错误类型

不要记录：

密码。
Token。
身份证号、手机号明文。
API Key。
完整病历、患者敏感信息。

异常设计

项目中不要到处 raise Exception("失败")，也不要捕获异常后什么都不做。

推荐定义业务异常：

python

class BizError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
        super().__init__(message)


class NotFoundError(BizError):
    pass

Service 使用：

python

def get_asset(asset_id: int, repository):
    asset = repository.find_by_id(asset_id)
    if asset is None:
        raise NotFoundError("ASSET_NOT_FOUND", "资产不存在")
    return asset

API 层统一转换成响应。这样前端看到的是稳定错误码，日志里保留详细堆栈。

全局异常处理链路

项目里异常要分层处理，不要每个接口都写一堆 try except。

mermaid

flowchart TD
    A["Repository 抛数据库或查询异常"] --> B["Service 转换或继续抛出"]
    B --> C["API 全局异常处理器"]
    C --> D{"异常类型"}
    D -->|业务异常| E["返回稳定错误码和提示"]
    D -->|参数异常| F["返回 400 或 422"]
    D -->|未知异常| G["记录堆栈 返回系统异常"]

示例：

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()


class BizError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
        super().__init__(message)


@app.exception_handler(BizError)
async def handle_biz_error(request: Request, exc: BizError):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message, "data": None},
    )


@app.exception_handler(Exception)
async def handle_unknown_error(request: Request, exc: Exception):
    # 真实项目这里用 logger.exception，并带 request_id
    return JSONResponse(
        status_code=500,
        content={"code": "INTERNAL_ERROR", "message": "系统异常", "data": None},
    )

注意：未知异常要记录完整堆栈，但不要把堆栈返回给前端。前端只需要稳定错误码和友好提示，开发排查需要日志里的 traceback、request_id、用户和输入摘要。

错误码设计

错误码不要随便写中文字符串，建议稳定、可搜索、可统计。

错误码	含义
`ASSET_NOT_FOUND`	资产不存在
`SOURCE_SYSTEM_INVALID`	来源系统非法
`IMPORT_FILE_INVALID`	导入文件不合法
`PERMISSION_DENIED`	无权限
`EXTERNAL_SERVICE_TIMEOUT`	外部服务超时

错误码的价值：

前端可以按错误码做不同提示。
日志平台可以统计错误类型。
面试或排查时能说清楚错误边界。
国际化时不依赖中文文案。

测试策略

项目测试分层：

mermaid

flowchart TD
    A["测试"] --> B["单元测试"]
    A --> C["接口测试"]
    A --> D["集成测试"]
    B --> E["Service 业务规则"]
    C --> F["FastAPI TestClient"]
    D --> G["真实数据库或测试容器"]

优先测试 Service，因为业务规则最容易出 bug，也最值得保护。

Service 测试示例：

python

import pytest


class FakeRepository:
    def save(self, request):
        return {"id": 1, "name": request["name"]}


class FakeAiClient:
    def submit_asset_analyze(self, asset_id: int, timeout_seconds: int):
        return {"submitted": True}


def test_create_asset_success():
    service = AssetService(FakeRepository(), FakeAiClient())
    asset = service.create_asset({"name": "检验报告", "source_system": "LIS"})
    assert asset["id"] == 1


def test_create_asset_invalid_source():
    service = AssetService(FakeRepository(), FakeAiClient())
    with pytest.raises(ValueError):
        service.create_asset({"name": "未知数据", "source_system": "UNKNOWN"})

接口测试示例：

python

from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "UP"

测试金字塔和测试边界

不是所有测试都要启动完整服务。越底层的测试越快，越上层的测试越接近真实环境但成本更高。

mermaid

flowchart TD
    A["少量端到端测试"] --> B["接口测试"]
    B --> C["Service 单元测试"]
    C --> D["纯函数和工具函数测试"]

测试建议：

层级	测什么	特点
工具函数	字段清洗、金额计算、解析函数	快、稳定
Service	业务规则、事务边界、异常路径	最值得重点测
Repository	SQL、ORM 查询、约束	需要测试数据库
API	参数校验、状态码、响应格式	使用 TestClient
E2E	从入口到数据库完整链路	少量关键路径

测试命名要表达业务：

python

def test_import_asset_should_record_error_when_asset_code_is_empty():
    ...

比下面这种更容易维护：

python

def test_case_1():
    ...

Mock、Fake 和真实依赖怎么选

方式	适合	注意
Fake	内存 Repository、假 Client	适合 Service 单测
Mock	验证是否调用某个外部方法	不要过度 mock 内部细节
真实依赖	数据库、Redis、HTTP 服务	适合集成测试

外部 AI、短信、支付、医院接口不要在单元测试里真实调用。应使用 Fake Client 或 Mock，避免测试慢、贵、不稳定。

Dockerfile 示例

简单 Dockerfile：

dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app ./app

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

注意：

先复制依赖文件再安装依赖，可以利用 Docker 缓存。
不要把 .env、日志、缓存、虚拟环境复制进镜像。
镜像里不要写死生产密钥。
容器启动命令要可配置 worker、端口和环境变量。

Docker 构建为什么这样写

Dockerfile 中先复制依赖文件再安装依赖，是为了利用构建缓存。

mermaid

flowchart TD
    A["COPY requirements.txt"] --> B["pip install"]
    B --> C["COPY app 代码"]
    C --> D["启动应用"]

如果只改业务代码，没有改 requirements.txt，Docker 可以复用已经安装好的依赖层，构建更快。

生产镜像注意：

不要复制 .venv，容器里重新安装依赖。
不要复制 .env，密钥运行时注入。
不要复制测试缓存、日志、上传文件。
固定基础镜像版本，避免构建结果漂移。
安装依赖用 --no-cache-dir 减少镜像体积。

.dockerignore 示例：

text

.venv
.env
__pycache__
.pytest_cache
.mypy_cache
*.log
docs
tests

如果生产镜像需要运行迁移，要明确迁移由谁执行：应用启动前执行、CI/CD 执行，还是人工审批后执行。不要让每个应用实例同时自动迁移数据库，否则多副本启动时可能互相冲突。

健康检查

健康检查分两类：

接口	作用	是否检查依赖
`/health`	进程是否活着	通常不查数据库
`/ready`	是否准备好接流量	可以检查数据库、Redis、配置

示例：

python

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


@app.get("/ready")
def ready(repository=Depends(get_asset_repository)):
    repository.ping()
    return {"status": "READY"}

不要让健康检查执行很重的 SQL 或调用慢外部接口，否则健康检查本身会成为压力源。

从开发到上线流程

mermaid

flowchart TD
    A["创建虚拟环境"] --> B["安装依赖"]
    B --> C["编写配置模板"]
    C --> D["实现 API Service Repository"]
    D --> E["补单元测试和接口测试"]
    E --> F["本地运行和联调"]
    F --> G["构建 Docker 镜像"]
    G --> H["测试环境部署"]
    H --> I["执行数据库迁移"]
    I --> J["生产发布"]
    J --> K["观察日志 指标 错误率"]

上线前检查：

依赖版本是否锁定。
.env 是否没有提交到 Git。
数据库迁移是否已在测试环境执行。
核心测试是否通过。
日志是否包含 request_id 和错误堆栈。
外部接口是否设置超时。
健康检查是否可用。
敏感信息是否脱敏。

线上问题排查

mermaid

flowchart TD
    A["线上报错或变慢"] --> B["确认影响范围"]
    B --> C["查日志 request_id"]
    C --> D{"是否有异常堆栈"}
    D -- "有" --> E["定位代码和输入"]
    D -- "没有" --> F["看耗时和依赖状态"]
    F --> G["数据库 连接池 外部接口 CPU 内存"]
    E --> H["本地或测试环境复现"]
    G --> H
    H --> I["补测试 防止复发"]

不要只修现象。比如接口慢，不能只把超时时间调大；数据库连接耗尽，不能只把连接池调大。要找真实原因：慢 SQL、连接未释放、外部接口无超时、任务并发过高、日志爆量、依赖不可用。

商业场景：数据资产导入服务

需求：医院不同系统导出 CSV，平台导入后形成资产目录。

项目落地设计：

模块	设计
API	提交导入任务、查询任务状态
Service	校验文件、解析字段、创建任务、处理状态
Repository	保存资产、字段、批次、审计日志
Script	支持离线批量导入历史文件
Log	记录 batch_id、request_id、成功数、失败数
Test	测试字段校验、重复导入、异常回滚
Deploy	Docker 部署，环境变量注入数据库和密钥

关键原则：

HTTP 接口只负责提交任务，不要等待大文件全部处理完。
导入过程要记录批次，失败后能重试。
数据库写入要考虑事务边界，不能出现半成品数据。
日志不要记录患者隐私明文。
测试要覆盖重复导入、非法字段、数据库异常。

商业 Demo：最小可落地的数据资产导入项目

下面给出一个能落地的简化骨架。它不追求功能多，而是把分层、配置、异常、日志、测试入口串起来。

text

asset_api/
  app/
    main.py
    api/asset_api.py
    core/config.py
    core/errors.py
    schemas/asset_schema.py
    services/asset_service.py
    repositories/asset_repository.py
  tests/test_asset_service.py

core/errors.py：

python

class BizError(Exception):
    def __init__(self, code: str, message: str):
        self.code = code
        self.message = message
        super().__init__(message)

schemas/asset_schema.py：

python

from pydantic import BaseModel, Field


class AssetCreate(BaseModel):
    asset_code: str = Field(min_length=1, max_length=50)
    asset_name: str = Field(min_length=1, max_length=100)
    source_system: str = Field(min_length=1, max_length=50)


class AssetVO(BaseModel):
    id: int
    asset_code: str
    asset_name: str
    source_system: str

repositories/asset_repository.py：

python

class AssetRepository:
    def __init__(self):
        self._rows: list[dict] = []
        self._next_id = 1

    def exists_by_code(self, asset_code: str) -> bool:
        return any(row["asset_code"] == asset_code for row in self._rows)

    def save(self, data) -> dict:
        row = {
            "id": self._next_id,
            "asset_code": data.asset_code,
            "asset_name": data.asset_name,
            "source_system": data.source_system,
        }
        self._next_id += 1
        self._rows.append(row)
        return row

services/asset_service.py：

python

from app.core.errors import BizError
from app.repositories.asset_repository import AssetRepository


class AssetService:
    def __init__(self, repository: AssetRepository):
        self.repository = repository

    def create_asset(self, request):
        if request.source_system not in {"HIS", "LIS", "PACS", "EMR"}:
            raise BizError("SOURCE_SYSTEM_INVALID", "来源系统不支持")
        if self.repository.exists_by_code(request.asset_code):
            raise BizError("ASSET_CODE_DUPLICATED", "资产编号已存在")
        return self.repository.save(request)

api/asset_api.py：

python

from fastapi import APIRouter, Depends

from app.repositories.asset_repository import AssetRepository
from app.schemas.asset_schema import AssetCreate, AssetVO
from app.services.asset_service import AssetService

router = APIRouter(prefix="/assets", tags=["assets"])
repository = AssetRepository()


def get_asset_service() -> AssetService:
    return AssetService(repository)


@router.post("", response_model=AssetVO)
def create_asset(
    request: AssetCreate,
    service: AssetService = Depends(get_asset_service),
):
    return service.create_asset(request)

main.py：

python

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from app.api.asset_api import router as asset_router
from app.core.errors import BizError

app = FastAPI(title="asset-api")
app.include_router(asset_router)


@app.exception_handler(BizError)
async def handle_biz_error(request: Request, exc: BizError):
    return JSONResponse(
        status_code=400,
        content={"code": exc.code, "message": exc.message, "data": None},
    )


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

tests/test_asset_service.py：

python

import pytest

from app.core.errors import BizError
from app.repositories.asset_repository import AssetRepository
from app.schemas.asset_schema import AssetCreate
from app.services.asset_service import AssetService


def test_create_asset_success():
    service = AssetService(AssetRepository())
    request = AssetCreate(
        asset_code="A001",
        asset_name="心电监护仪",
        source_system="HIS",
    )

    asset = service.create_asset(request)

    assert asset["id"] == 1
    assert asset["asset_code"] == "A001"


def test_create_asset_invalid_source_system():
    service = AssetService(AssetRepository())
    request = AssetCreate(
        asset_code="A001",
        asset_name="心电监护仪",
        source_system="UNKNOWN",
    )

    with pytest.raises(BizError) as exc:
        service.create_asset(request)

    assert exc.value.code == "SOURCE_SYSTEM_INVALID"

这个 Demo 虽然用内存 Repository，但已经体现了生产项目的核心结构：

API 只处理 HTTP。
Schema 只处理入参结构。
Service 处理业务规则。
Repository 隔离数据访问。
BizError 统一表达业务失败。
测试绕过 HTTP，直接保护核心业务。

后续把内存 Repository 换成 SQLAlchemy，不需要大改 API 和 Service 的调用方式。

调用链：

mermaid

flowchart TD
    A["POST /assets"] --> B["Pydantic 校验 AssetCreate"]
    B --> C["API 获取 AssetService"]
    C --> D["Service 校验来源系统和重复编号"]
    D --> E["Repository 保存资产"]
    E --> F["返回 AssetVO"]
    D --> G["业务异常 BizError"]
    G --> H["全局异常处理返回错误码"]

常见坑

问题	后果	正确做法
所有代码写一个文件	难维护、难测试	按 API/Service/Repository 分层
配置写死	密钥泄露、环境混乱	`.env.example` + 环境变量
不锁依赖版本	环境不可复现	固定版本或使用锁文件
业务异常直接返回堆栈	不安全、不友好	全局异常处理和错误码
用 `print` 打日志	线上难查	`logging` + request_id
没有测试	改代码容易回归	核心 Service 和 API 写测试
脚本无日志无统计	失败不知道多少	记录成功、失败、原因
生产密钥进镜像	安全风险	运行时注入环境变量

README 应该写什么

README 是项目入口，至少包含：

markdown

# asset-api

## 项目用途

医疗数据资产 API，提供资产登记、导入任务、查询和审计能力。

## 环境要求

- Python 3.11
- MySQL 8

## 安装依赖

pip install -r requirements.txt

## 配置

复制 .env.example 为 .env，并填写 DATABASE_URL、SECRET_KEY。

## 启动

uvicorn app.main:app --reload

## 测试

pytest

## 目录说明

api 处理 HTTP，services 处理业务，repositories 处理数据库。

面试标准回答

Python 项目怎么组织目录？
我会按入口、接口、业务、数据访问、配置、测试分层。FastAPI 项目中通常有 api、schemas、services、repositories、models、core、tests。API 层处理 HTTP，Service 层写业务规则，Repository 层访问数据库，配置、异常、日志放到 core。

为什么不能把所有代码写在一个文件里？
单文件短期能跑，但业务、数据库、配置、异常、日志混在一起，后续难测试、难复用、难排查。分层的目的是控制复杂度，让每一层职责明确，改一个点不会影响所有代码。

配置为什么不能写死？
写死配置会导致密钥泄露、环境无法区分、修改配置必须重新发版。生产项目应该用环境变量、配置文件或配置中心，提交 .env.example 说明需要哪些配置，真实 .env 不提交。

上线前要检查什么？
检查依赖版本、配置注入、数据库迁移、核心测试、日志、异常处理、超时、健康检查、敏感信息脱敏和回滚方案。上线后观察错误率、接口耗时、CPU、内存、数据库连接和日志。

Service、Repository、API 分别负责什么？
API 负责 HTTP 请求响应、状态码和依赖注入；Service 负责业务规则、流程编排和事务边界；Repository 负责数据库访问和查询封装。这样同一段业务逻辑可以被 Web、CLI、定时任务复用，也方便单元测试。

为什么要有全局异常处理？
全局异常处理可以把业务异常、参数异常、未知异常转换成统一响应格式，前端拿到稳定错误码，后端日志保留 traceback。否则每个接口手写 try except 容易重复、遗漏，也可能把内部堆栈暴露给用户。

为什么核心业务要优先写 Service 测试？
Service 承载业务规则，变化频繁且最容易出错；直接测试 Service 不需要启动完整 Web 服务，速度快、定位准。API 测试用于补充参数校验和响应格式，端到端测试只覆盖少量关键链路。

为什么 Docker 镜像里不要放 .env？
.env 往往包含数据库密码、Token、密钥。放进镜像会导致密钥随镜像传播，测试和生产也难以区分。正确做法是镜像只包含代码和依赖，配置通过环境变量、Secret 或配置中心在运行时注入。

关联知识点

本章小结

Python 项目实践的核心是工程化：结构清晰、配置可控、依赖可复现、日志能排查、测试能回归、部署能落地。零基础学习时不要只满足于“能运行”，要逐步养成“能维护、能解释、能定位、能上线”的思维。

数据处理

Python 项目实践

学习目标

项目实践到底练什么

一个项目从需求到上线的完整过程

从单文件到项目的演进

推荐项目结构

文件划分为什么这么分

import 路径和运行方式

分层调用流程

不分层会怎样

配置管理

配置加载顺序和校验

依赖管理

开发依赖、生产依赖和锁文件

代码质量工具

命令行入口和脚本

日志和 TraceId

异常设计

全局异常处理链路

错误码设计

测试策略

测试金字塔和测试边界

Mock、Fake 和真实依赖怎么选

Dockerfile 示例

Docker 构建为什么这样写

健康检查

从开发到上线流程

线上问题排查

商业场景：数据资产导入服务

商业 Demo：最小可落地的数据资产导入项目

常见坑

README 应该写什么

面试标准回答

关联知识点

本章小结

Python 项目实践 ​

学习目标 ​

项目实践到底练什么 ​

一个项目从需求到上线的完整过程 ​

从单文件到项目的演进 ​

推荐项目结构 ​

文件划分为什么这么分 ​

import 路径和运行方式 ​

分层调用流程 ​

不分层会怎样 ​

配置管理 ​

配置加载顺序和校验 ​

依赖管理 ​

开发依赖、生产依赖和锁文件 ​

代码质量工具 ​

命令行入口和脚本 ​

日志和 TraceId ​

异常设计 ​

全局异常处理链路 ​

错误码设计 ​

测试策略 ​

测试金字塔和测试边界 ​

Mock、Fake 和真实依赖怎么选 ​

Dockerfile 示例 ​

Docker 构建为什么这样写 ​

健康检查 ​

从开发到上线流程 ​

线上问题排查 ​

商业场景：数据资产导入服务 ​

商业 Demo：最小可落地的数据资产导入项目 ​

常见坑 ​

README 应该写什么 ​

面试标准回答 ​

关联知识点 ​

本章小结 ​

Python 项目实践

学习目标

项目实践到底练什么

一个项目从需求到上线的完整过程

从单文件到项目的演进

推荐项目结构

文件划分为什么这么分

import 路径和运行方式

分层调用流程

不分层会怎样

配置管理

配置加载顺序和校验

依赖管理

开发依赖、生产依赖和锁文件

代码质量工具

命令行入口和脚本

日志和 TraceId

异常设计

全局异常处理链路

错误码设计

测试策略

测试金字塔和测试边界

Mock、Fake 和真实依赖怎么选

Dockerfile 示例

Docker 构建为什么这样写

健康检查

从开发到上线流程

线上问题排查

商业场景：数据资产导入服务

商业 Demo：最小可落地的数据资产导入项目

常见坑

README 应该写什么

面试标准回答

关联知识点

本章小结