Python 工程实践核心（面试向）

概述

本篇是「Python 面试核心」系列第二篇，聚焦工程实践中天天要用的标准库与三方库：日志、命令行参数、路径、配置序列化、HTTP 请求、子进程、并发三件套。这些模块的熟练度直接决定「会写 Python」与「能用 Python 干活」的分界线。

所有示例均在 CentOS 测试机（Python 3.10.12 + fastapi 0.141 / requests 2.29 / pyyaml 最新版）实测通过。

logging

print 只能调试，logging 才能运维：分级、落盘、按模块命名是三要素。

import logging

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
    handlers=[logging.StreamHandler(),
              logging.FileHandler("/tmp/app.log", encoding="utf-8")],
)
log = logging.getLogger("myapp.db")     # 点分命名形成层级：myapp.db 继承 myapp
log.info("服务启动")
log.warning("连接超时，准备重试")
log.error("写库失败", exc_info=True)     # exc_info 把堆栈一起打出来

生产常用按大小/时间轮转，防止单文件无限膨胀：

import logging
from logging.handlers import RotatingFileHandler

log = logging.getLogger("myapp")
log.setLevel(logging.INFO)
handler = RotatingFileHandler("/tmp/app.log", maxBytes=1024*1024,
                              backupCount=5, encoding="utf-8")
handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
log.addHandler(handler)
log.info("轮转日志：超过 1MB 自动切 app.log.1，最多留 5 份")

面试要点：

• 等级从低到高：DEBUG / INFO / WARNING / ERROR / CRITICAL，默认 root 是 WARNING
• Logger 树形传播：子 logger 处理完向父级传播，级别在 logger 上设、格式在 handler 上设
• 别在库里 basicConfig，入口（main）统一配置；库只 getLogger(__name__)
• exc_info=True 或 log.exception(...) 是记录异常堆栈的正确姿势

argparse

标准库命令行解析，写运维脚本/CLI 工具的第一选择：

import argparse

p = argparse.ArgumentParser(description="部署工具示例")
p.add_argument("host", help="目标主机")                    # 位置参数
p.add_argument("-p", "--port", type=int, default=22)      # 可选参数，带类型与默认值
p.add_argument("--dry-run", action="store_true")          # 布尔开关
p.add_argument("-v", "--verbose", action="count", default=0)

args = p.parse_args(["web-01", "--port", "2222", "-vv"])  # 实际使用传 sys.argv
print(args.host, args.port, args.dry_run, args.verbose)   # web-01 2222 False 2

面试要点：

• -h/--help 自动生成，type=int 顺带完成类型校验，非法输入自动报错退出
• action 常用三兄弟：store_true（开关）、count（-vvv 累计）、append（多值收集）
• 复杂 CLI 用 add_subparsers 做 git 风格子命令；现代替代品 click/typer 思路一致（装饰器注册）

pathlib

面向对象的路径操作，比字符串拼接的 os.path 更安全直观，跨平台自动处理分隔符：

from pathlib import Path

p = Path("/tmp/kb_demo") / "sub" / "a.txt"    # 用 / 拼路径
p.parent.mkdir(parents=True, exist_ok=True)   # 建目录：parents 递归，exist_ok 幂等
p.write_text("hello pathlib", encoding="utf-8")

print(p.name, p.suffix, p.stem)       # a.txt .txt a
print(p.read_text(encoding="utf-8"))  # hello pathlib
print(p.exists(), p.is_file())        # True True

# glob 通配与批量处理
print([f.name for f in p.parent.parent.glob("**/*.txt")])   # ['a.txt']
print(p.with_suffix(".log").name)     # a.log

面试要点：

• 「读写文件不再需要 open/close」：read_text/write_text/read_bytes 一步到位，大文件仍用 open()
• mkdir(parents=True, exist_ok=True) 一行抵掉 os.makedirs 的三行判断
• 遗留代码里 os.path 很常见，两者可共存：Path(os.path.join(a, b))，新代码统一 pathlib

os

与操作系统打交道的总入口：环境变量、文件系统、进程信息：

import os

# 环境变量：配置注入的常规入口（12-Factor App 原则）
os.environ["KB_MODE"] = "prod"
print(os.environ.get("KB_MODE", "dev"))       # prod（get 带默认值不抛异常）

# 文件系统
os.makedirs("/tmp/os_demo/sub", exist_ok=True)
with open("/tmp/os_demo/sub/f.txt", "w") as f:
    f.write("data")
print(os.listdir("/tmp/os_demo"))             # ['sub']
print(os.path.getsize("/tmp/os_demo/sub/f.txt"))   # 4

# 进程信息
print(os.getpid() > 0, os.cpu_count() >= 1)   # True True

面试要点：

• 读环境变量用 os.environ.get(key, 默认值)，配置缺失要有兜底而不是裸 KeyError
• os.cpu_count() 是并发池大小的常见依据（min(32, cpu_count() * 2) 之类经验公式）
• os.system 有注入风险且拿不到输出，工程里一律 subprocess 模块替代

json

接口与配置的事实标准。核心考点：中文、缩进、datetime 序列化：

import json
from datetime import datetime

data = {"name": "youngsu", "tags": ["a", "b"], "ok": True, "n": None}
s = json.dumps(data, ensure_ascii=False, indent=2)   # 中文直出 + 可读缩进
print(s)
print(json.loads(s) == data)                          # True

# datetime 默认不可序列化 —— 自定义 encoder（面试手写题）
class Encoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.strftime("%Y-%m-%d %H:%M:%S")
        return super().default(o)
print(Encoder().encode({"now": datetime(2026, 9, 8, 10, 0)}))
# {"now": "2026-09-08 10:00:00"}

面试要点：

• ensure_ascii=False 才能输出中文，否则变 \uXXXX 转义；文件读写记得 encoding="utf-8"
• json.loads 接收字符串、json.load 接收文件对象；dumps/dump 同理
• None/True/False 与 null/true/false 的对应关系；数字没有 int/float 之分（反序列化后整型可能变浮点）
• dict 顺序在 dumps 时默认保留（3.7+ dict 有序）

yaml

k8s、CI 配置全在用 YAML。唯一必考点：一定用 safe_load：

import yaml                     # pip install pyyaml

text = """
service:
  name: linux-kb
  replicas: 2
features: [seo, sitemap]
debug: false
"""
cfg = yaml.safe_load(text)       # 安全解析
print(cfg["service"]["name"], cfg["features"])   # linux-kb ['seo', 'sitemap']
print(yaml.safe_dump(cfg, allow_unicode=True, sort_keys=False))

面试要点：

• yaml.load 不指定 Loader 可执行任意对象构造（经典安全问题），永远用 safe_load / safe_dump
• YAML 的坑：no 会被解析成 False、挪威国家代码 NO 变布尔是真实事故；严格场景用 JSON 或加引号
• 缩进即语法（同 Python），Tab 是非法字符，只能空格
• 与 json 互转：yaml.safe_load 结果可直接 json.dumps

requests

Python 最流行的 HTTP 客户端。timeout、raise_for_status、Session 是三件必备：

import functools, http.server, threading, requests

# 起本地 HTTP 服务当靶子（真实场景换成远端 API）
Handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory="/tmp")
srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)   # 随机空闲端口
threading.Thread(target=srv.serve_forever, daemon=True).start()
base = f"http://127.0.0.1:{srv.server_port}"

r = requests.get(base + "/", timeout=3)          # 永远带 timeout
print(r.status_code, r.headers["content-type"][:15])

s = requests.Session()                           # 复用 TCP 连接 + 统一 headers
s.headers.update({"User-Agent": "kb-demo"})
try:
    r2 = s.post(base + "/", data={"k": "v"}, timeout=3)
    r2.raise_for_status()                        # 非 2xx 抛 HTTPError
except requests.RequestException as e:
    print("请求失败(预期，演示用):", type(e).__name__)   # HTTPError
srv.shutdown()

面试要点：

• 不设 timeout 的 requests 调用可能永久挂起 —— 生产事故高频根因
• raise_for_status() 让 HTTP 错误显式化，requests 默认不因 4xx/5xx 抛异常
• Session 复用连接池，批量请求性能显著优于裸 get；request.RequestException 是统一异常基类
• 重试交给 urllib3.util.Retry 挂到 HTTPAdapter，不要手写 for 循环重试

subprocess

在 Python 里调用外部命令的标准姿势：

import subprocess

# 推荐列表参数（不经过 shell，无注入风险）
r = subprocess.run(["python3", "-c", "print(1+1)"],
                   capture_output=True, text=True, check=True)
print(r.stdout.strip())                    # 2

# check=True：非零退出码抛 CalledProcessError，比手工判断 returncode 稳
try:
    subprocess.run(["python3", "-c", "import sys; sys.exit(3)"], check=True)
except subprocess.CalledProcessError as e:
    print("失败 exit=", e.returncode)      # 失败 exit= 3

# 确需管道/通配符时才用 shell=True，且输入必须可信
r2 = subprocess.run("echo hello | tr a-z A-Z", shell=True,
                    capture_output=True, text=True)
print(r2.stdout.strip())                   # HELLO

面试要点：

• 列表参数 vs shell=True：后者有命令注入风险，字符串里有用户输入就别用
• capture_output=True, text=True 等于 stdout=PIPE, stderr=PIPE 并自动解码
• run 阻塞到结束；要非阻塞/流式读输出用 Popen，超时控制 timeout= 参数
• 环境传递：env= 传完整字典（通常基于 os.environ.copy() 追加）

asyncio

单线程并发跑 IO 密集任务的答案，协程靠事件循环调度：

import asyncio, time

async def fetch(i):
    await asyncio.sleep(0.3)               # 模拟 IO 等待
    return i * 2

async def main():
    t0 = time.time()
    rs = await asyncio.gather(fetch(1), fetch(2), fetch(3))   # 三个协程并发
    print(rs, f"总耗时 {time.time()-t0:.2f}s")   # [2, 4, 6] ≈0.30s 而非 0.90s

asyncio.run(main())                        # 程序入口：创建事件循环跑 main

面试要点：

• 协程在 await 处让出控制权，事件循环调度下一个任务 —— 并发的是「等待」，不是计算
• 裸 time.sleep 会卡死整个循环，协程里必须 await asyncio.sleep
• gather 并发收集结果；TaskGroup（3.11+）提供更好的结构化并发与异常处理
• CPU 密集任务上 asyncio 无收益（GIL + 无让出），那是 multiprocessing 的战场

threading

线程适合 IO 密集：等待网络/磁盘时释放 GIL 让别的线程跑：

import threading

counter = 0
lock = threading.Lock()                    # 锁保护共享状态，防竞态条件

def work():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1                   # += 是「读-改-写」，不是原子操作

ts = [threading.Thread(target=work) for _ in range(4)]
[t.start() for t in ts]
[t.join() for t in ts]
print(counter)                             # 稳定 400000；去掉锁大概率丢计数

面试要点：

• GIL 只保证字节码级别不撕裂数据结构，不保证业务原子性：counter += 1 照样丢更新，锁不可少
• with lock: 自动加锁解锁，比手动 acquire/release 安全（异常也不会忘记释放）
• 线程间通信用 queue.Queue（线程安全队列），别用裸列表 + 锁硬凑
• 线程是 IO 密集方案；CPU 密集换 multiprocessing，同一份任务代码几乎不用改

multiprocessing

独立进程、独立解释器、独立 GIL —— 真正吃满多核：

from multiprocessing import Pool
import os, math

def cpu_task(n):
    return sum(math.sqrt(i) for i in range(n))

if __name__ == "__main__":                 # 多进程入口必须有这个保护
    with Pool(2) as pool:                  # 进程池：每个子进程有独立 GIL
        rs = pool.map(cpu_task, [200000, 200000])
    print([f"{r:.0f}" for r in rs], "父进程 pid =", os.getpid())
# ['365148', '365148'] 父进程 pid = <当前进程>

面试要点：

• if __name__ == "__main__": 是硬要求：子进程 spawn 时会重新 import 主模块，没保护就无限繁殖
• 进程间不共享内存，传参经序列化（pickle），所以传大对象要先掂量成本
• 对比记忆：线程共享内存轻量但受 GIL 限制；进程隔离重但真并行；IO 密集线程/协程，CPU 密集进程
• Pool.map 与内置 map 语义一致但并发执行；apply_async 适合异构任务

高频面试题速答