09 文件、路径与数据交换

大约 3 分钟编程语言PythonpathlibJSONCSVSQLite

09 文件、路径与数据交换

文件操作的危险不只在语法，而在路径、编码、覆盖、权限、并发和大文件边界。

1. pathlib 基础

from pathlib import Path

root = Path("data")
root.mkdir(parents=True, exist_ok=True)

path = root / "service.txt"
path.write_text("api\n", encoding="utf-8")

print(path.resolve())
print(path.read_text(encoding="utf-8").strip())

常用能力：

/ 拼接路径。
.resolve() 获取绝对路径并解析符号链接。
.glob() 和 .rglob() 查找文件。
.suffix、.stem、.name、.parent 获取路径组成部分。
.is_file()、.is_dir()、.exists() 查询状态。

2. 编码

文本文件必须明确编码：

from pathlib import Path

path = Path("events.txt")
with path.open("r", encoding="utf-8", errors="strict") as handle:
    for line in handle:
        print(line.rstrip("\n"))

使用 errors="replace" 能避免程序崩溃，但会丢失原始内容信息。日志采集工具可以替换或转义，配置文件、审计记录和代码转换程序应优先暴露编码错误。

3. 大文件流式处理

from collections import Counter
from pathlib import Path

def count_levels(path: Path) -> Counter[str]:
    counts: Counter[str] = Counter()
    with path.open("r", encoding="utf-8") as handle:
        for line in handle:
            level = line.split(maxsplit=1)[0] if line.strip() else "EMPTY"
            counts[level] += 1
    return counts

不要对未知大小文件无条件调用 .read() 或 .read_text()。逐行、分块或流式解析能控制内存。

4. 原子替换

直接覆盖配置时，进程崩溃可能留下半个文件：

import os
import tempfile
from pathlib import Path

def atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=path.parent,
        delete=False,
        newline="\n",
    ) as handle:
        handle.write(content)
        handle.flush()
        os.fsync(handle.fileno())
        temporary = Path(handle.name)
    os.replace(temporary, path)

替换必须与目标文件位于同一文件系统。权限、所有者和 SELinux 上下文可能需要在 replace 后校正。

5. JSON

import json
from pathlib import Path

data = {"host": "db-01", "healthy": True, "ports": [5432, 6432]}
Path("health.json").write_text(
    json.dumps(data, ensure_ascii=False, indent=2) + "\n",
    encoding="utf-8",
)

loaded = json.loads(Path("health.json").read_text(encoding="utf-8"))
assert loaded["healthy"] is True

边界：

JSON 只支持对象、数组、字符串、数字、布尔和 null。
键始终是字符串。
浮点、时区和大整数需要明确业务规则。
loads 接受完整 JSON 字符串，解析大文件优先考虑流式库。
来自不可信来源的 JSON 仍要做 Schema 和范围校验。

6. CSV

import csv
from pathlib import Path

rows = [
    {"service": "api", "status": "ok"},
    {"service": "db", "status": "warn"},
]

with Path("status.csv").open("w", encoding="utf-8", newline="") as handle:
    writer = csv.DictWriter(handle, fieldnames=["service", "status"])
    writer.writeheader()
    writer.writerows(rows)

CSV 没有统一的类型系统和嵌套结构。解析时明确分隔符、引号、编码、表头和换行规则。

7. SQLite

import sqlite3

with sqlite3.connect("events.db") as connection:
    connection.execute(
        "CREATE TABLE IF NOT EXISTS events "
        "(id INTEGER PRIMARY KEY, service TEXT NOT NULL, level TEXT NOT NULL)"
    )
    connection.execute(
        "INSERT INTO events(service, level) VALUES (?, ?)",
        ("api", "ERROR"),
    )

始终使用参数占位符传递值，不要拼接 SQL。SQLite 适合单机工具、本地缓存和小规模状态；多写入者、高并发和跨主机共享应选择相应数据库。

8. 配置文件

常见格式：

.toml：结构化、可读，Python 3.11 起标准库可用 tomllib 读取。
.json：通用，跨语言支持好，但不方便手写注释。
.yaml：表达力强，但第三方解析器复杂，不应默认信任不可信输入。
.ini：适合简单分层配置，使用 configparser。

配置优先级通常为：命令行参数 > 环境变量 > 配置文件 > 默认值。

9. 路径安全

处理用户输入路径时：

解析绝对路径。
确认它位于允许根目录内。
不把未经约束的字符串传给 shell=True、rm 或远程命令。
对符号链接、相对路径、大小写和 Unicode 规范化做明确策略。

from pathlib import Path

def safe_child(root: Path, relative: str) -> Path:
    resolved_root = root.resolve()
    candidate = (resolved_root / relative).resolve()
    if candidate != resolved_root and resolved_root not in candidate.parents:
        raise ValueError("path escapes root")
    return candidate