最佳实践
高质量 Python 代码通常具备以下特点:
正确
清晰
简单
稳定
可测试
可维护
可扩展
最佳实践不是固定规则,而是一组优先原则:
可读性优先于炫技
明确优先于隐式
简单优先于复杂
可维护优先于少写几行
先保证正确,再优化性能
PEP 8 是 Python 官方代码风格指南。
推荐:
def calculate_total(
price: float,
quantity: int,
) -> float:
return price * quantity
不推荐使用 Tab 和空格混合缩进。
推荐:
stock_code = "600519"
def calculate_profit() -> float:
...
不推荐:
stockCode = "600519"
def CalculateProfit():
...
class StockAnalyzer:
pass
MAX_RETRY_COUNT = 3
DEFAULT_ENCODING = "utf-8"
DATABASE_PORT = 5432
Python 不会真正禁止修改常量,但大写表示:
这个变量不应该被修改。
通常建议每行不超过 88 个字符左右。
不推荐:
result = calculate_stock_profit(stock_code, buy_price, sell_price, quantity, commission_rate, tax_rate)
推荐:
result = calculate_stock_profit(
stock_code=stock_code,
buy_price=buy_price,
sell_price=sell_price,
quantity=quantity,
commission_rate=commission_rate,
tax_rate=tax_rate,
)
命名是代码可读性的基础。
不推荐:
a = 100
b = 20
c = a * b
推荐:
price = 100
quantity = 20
total_amount = price * quantity
推荐:
load_stock_data()
save_report()
calculate_profit()
validate_stock_code()
不推荐:
stock()
data()
result()
推荐:
is_valid = True
has_permission = False
can_trade = True
should_retry = False
推荐:
stocks = []
stock_codes = set()
errors = []
单个对象使用单数:
stock = {}
error = None
不推荐:
usr_cfg
stk_lst
prc
qty
推荐:
user_config
stock_list
price
quantity
常见且明确的缩写可以保留:
url
id
api
json
csv
html
sql
不推荐:
def process_stock_data():
# 读取文件
# 解析 JSON
# 清洗数据
# 筛选股票
# 写入数据库
# 保存报告
# 发送邮件
...
推荐拆分:
def load_stock_data(
file_path: Path,
) -> list[dict[str, object]]:
...
def clean_stock_data(
stocks: list[dict[str, object]],
) -> list[dict[str, object]]:
...
def filter_stocks(
stocks: list[dict[str, object]],
) -> list[dict[str, object]]:
...
def save_stock_data(
file_path: Path,
stocks: list[dict[str, object]],
) -> None:
...
主函数负责组合:
def main() -> None:
stocks = load_stock_data(
Path("stocks.json")
)
stocks = clean_stock_data(stocks)
stocks = filter_stocks(stocks)
save_stock_data(
Path("output/stocks.json"),
stocks,
)
不推荐:
def create_order(
code,
name,
price,
quantity,
direction,
account,
strategy,
remark,
):
...
参数过多时,可以使用数据类:
from dataclasses import dataclass
from enum import StrEnum
class Direction(StrEnum):
BUY = "buy"
SELL = "sell"
@dataclass(slots=True)
class Order:
code: str
price: float
quantity: int
direction: Direction
account: str
strategy: str | None = None
remark: str | None = None
函数:
def submit_order(
order: Order,
) -> None:
...
参数较多或容易混淆时:
create_order(
code="600519",
price=1500.0,
quantity=100,
direction="buy",
)
比下面更清晰:
create_order(
"600519",
1500.0,
100,
"buy",
)
可以通过 * 强制使用关键字参数:
def create_order(
code: str,
*,
price: float,
quantity: int,
direction: str,
) -> None:
...
错误写法:
def add_item(
item: str,
items: list[str] = [],
) -> list[str]:
items.append(item)
return items
默认列表只创建一次,多次调用会共享数据。
正确写法:
def add_item(
item: str,
items: list[str] | None = None,
) -> list[str]:
if items is None:
items = []
items.append(item)
return items
数据类中使用:
from dataclasses import dataclass, field
@dataclass
class Portfolio:
stocks: list[str] = field(
default_factory=list
)
不推荐:
def calculate_profit(
buy_price: float,
sell_price: float,
) -> None:
print(sell_price - buy_price)
推荐:
def calculate_profit(
buy_price: float,
sell_price: float,
) -> float:
return sell_price - buy_price
调用方决定如何处理:
profit = calculate_profit(
buy_price=10,
sell_price=12,
)
print(profit)
优点:
- 更容易测试
- 更容易复用
- 可以写入文件
- 可以继续参与计算
- 可以由不同界面展示
不推荐:
def calculate_total() -> float:
price = float(
input("请输入价格:")
)
quantity = int(
input("请输入数量:")
)
return price * quantity
推荐:
def calculate_total(
price: float,
quantity: int,
) -> float:
return price * quantity
用户输入放在外部:
def main() -> None:
price = float(
input("请输入价格:")
)
quantity = int(
input("请输入数量:")
)
total = calculate_total(
price,
quantity,
)
print(total)
纯函数:
- 相同输入得到相同输出
- 不修改外部状态
- 不修改传入对象
推荐:
def add_stock(
stocks: list[str],
code: str,
) -> list[str]:
return [
*stocks,
code,
]
不推荐:
def add_stock(
stocks: list[str],
code: str,
) -> None:
stocks.append(code)
并不是所有函数都必须是纯函数,但数据计算和转换逻辑应尽量纯粹。
不推荐:
tax_rate = 0.001
def calculate_tax(
amount: float,
) -> float:
return amount * tax_rate
更推荐显式传入:
def calculate_tax(
amount: float,
tax_rate: float,
) -> float:
return amount * tax_rate
调用:
tax = calculate_tax(
amount=10000,
tax_rate=0.001,
)
配置数据可以封装:
from dataclasses import dataclass
@dataclass(frozen=True)
class TradingConfig:
tax_rate: float
commission_rate: float
推荐:
def calculate_profit(
buy_price: float,
sell_price: float,
quantity: int,
) -> float:
return (
sell_price - buy_price
) * quantity
类型提示可以帮助:
- 编辑器自动补全
- 静态类型检查
- 阅读函数接口
- 提前发现类型错误
- 大型项目维护
Python 3.10 以上:
def find_stock(
code: str,
) -> dict[str, object] | None:
...
不需要旧式写法:
from typing import Optional
Python 3.12:
type StockCode = str
type StockData = dict[
str,
object,
]
type StockList = list[
StockData
]
from typing import NotRequired, TypedDict
class StockDict(TypedDict):
code: str
name: str
price: float
industry: NotRequired[str]
from typing import Protocol
class StockRepository(Protocol):
def find(
self,
code: str,
) -> Stock | None:
...
这样不要求具体类继承某个父类,只要具有对应方法即可。
适合使用类:
- 数据具有固定字段
- 对象需要维护状态
- 数据具有相关行为
- 需要多个同类型对象
- 需要接口、多态或依赖注入
不一定需要类:
- 简单转换
- 一次性脚本
- 无状态计算
- 几个简单函数即可完成的任务
不要为了面向对象而强行使用类。
不推荐手写大量初始化代码:
class Stock:
def __init__(
self,
code: str,
name: str,
price: float,
) -> None:
self.code = code
self.name = name
self.price = price
推荐:
from dataclasses import dataclass
@dataclass(slots=True)
class Stock:
code: str
name: str
price: float
如果对象不应该修改:
@dataclass(
frozen=True,
slots=True,
)
class Stock:
code: str
name: str
price: float
继承适合:
子类确实是一种父类
例如:
class Order:
...
class MarketOrder(Order):
...
组合适合:
一个对象拥有另一个对象
class StockService:
def __init__(
self,
repository: StockRepository,
) -> None:
self.repository = repository
实际项目中,组合通常比多层继承更灵活。
不推荐:
def process() -> bool:
try:
...
return True
except Exception as error:
print(error)
return False
问题:
- 捕获范围太大
- 隐藏真正 Bug
- 无法区分失败原因
- 丢失完整堆栈
推荐捕获明确异常:
def read_json(
file_path: Path,
) -> object:
try:
text = file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError as error:
raise FileNotFoundError(
f"文件不存在:{file_path}"
) from error
try:
return json.loads(text)
except json.JSONDecodeError as error:
raise ValueError(
f"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列"
) from error
不推荐:
try:
data = load_data()
data = clean_data(data)
data = analyze_data(data)
save_data(data)
except ValueError:
...
推荐分开处理可能发生的异常。
当前函数不能解决时,让异常继续传播。
def save_data(
data: object,
) -> None:
database.save(data)
不需要每一层都写:
except Exception:
try:
price = float(value)
except ValueError as error:
raise InvalidPriceError(
f"价格格式错误:{value!r}"
) from error
不推荐:
except:
...
推荐:
except Exception:
...
裸捕获还会拦截:
KeyboardInterruptSystemExitGeneratorExit
开发测试时可以使用:
print()
正式项目应使用:
import logging
配置:
logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s "
"%(levelname)s "
"%(name)s "
"%(message)s"
),
)
获取日志器:
logger = logging.getLogger(
__name__
)
记录:
logger.info(
"开始处理股票:%s",
code,
)
logger.warning(
"股票数据不完整:%s",
code,
)
logger.error(
"文件读取失败:%s",
file_path,
)
异常中:
try:
process()
except ValueError:
logger.exception(
"数据处理失败"
)
logger.exception() 会记录完整堆栈。
推荐:
from pathlib import Path
file_path = (
Path("data")
/ "stocks.json"
)
不推荐:
file_path = (
"data/"
+ "stocks.json"
)
文本读取:
text = file_path.read_text(
encoding="utf-8",
)
写入:
file_path.write_text(
text,
encoding="utf-8",
)
目录创建:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
推荐:
with file_path.open(
"r",
encoding="utf-8",
) as file:
content = file.read()
不要依赖系统默认编码:
open("data.txt")
不同系统默认编码可能不同。
推荐:
with file_path.open(
"r",
encoding="utf-8",
) as file:
content = file.read()
不要:
file = open("data.txt")
content = file.read()
file.close()
with 可以保证异常发生时仍然释放资源。
同样适用于:
- 数据库连接
- 文件锁
- 网络连接
- 临时文件
- 线程锁
不推荐:
content = file_path.read_text(
encoding="utf-8",
)
处理大文件时:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line in file:
process_line(line)
处理大 JSON 数据时,可以考虑:
- JSON Lines
- 分批处理
- 数据库
- 流式解析器
直接覆盖重要文件可能产生半写入状态。
推荐流程:
写入临时文件
→ 写入成功
→ 原子替换正式文件
from pathlib import Path
import os
import tempfile
def safe_write_text(
file_path: Path,
content: str,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=file_path.parent,
delete=False,
) as temp_file:
temp_file.write(content)
temp_path = Path(
temp_file.name
)
try:
os.replace(
temp_path,
file_path,
)
except Exception:
temp_path.unlink(
missing_ok=True,
)
raise
JSON 格式正确不等于业务数据正确。
data = json.loads(text)
还应继续检查:
if not isinstance(data, list):
raise TypeError(
"JSON 根节点必须是数组"
)
单条记录:
if not isinstance(item, dict):
raise TypeError(
"股票记录必须是对象"
)
字段:
required_fields = {
"code",
"name",
"price",
}
missing_fields = (
required_fields
- item.keys()
)
if missing_fields:
raise ValueError(
"缺少字段:"
+ ", ".join(
sorted(missing_fields)
)
)
不推荐直接使用浮点数处理精确金额:
print(0.1 + 0.2)
结果可能是:
0.30000000000000004
推荐:
from decimal import Decimal
amount = (
Decimal("0.1")
+ Decimal("0.2")
)
结果:
0.3
从浮点数转换时:
Decimal(str(value))
不要直接:
Decimal(value)
财务数据保存到 JSON 时,可以保存为字符串:
{
"amount": "1500.50"
}
读取后:
amount = Decimal(
data["amount"]
)
常见选择:
有序可重复数据 list
不可变序列 tuple
键值映射 dict
去重和成员判断 set
先进先出队列 deque
计数 Counter
自动默认值分组 defaultdict
例如,判断股票代码是否存在:
不推荐列表:
if code in code_list:
...
大量查询时推荐集合:
code_set = set(code_list)
if code in code_set:
...
不推荐:
if len(process_data(data)) > 10:
print(process_data(data))
推荐:
result = process_data(data)
if len(result) > 10:
print(result)
昂贵计算可以使用缓存:
from functools import cache
@cache
def calculate_value(
code: str,
) -> float:
...
缓存只适用于:
- 输入可哈希
- 相同输入结果稳定
- 数据不会频繁变化
- 缓存不会无限膨胀
正确顺序:
先实现正确
再提高可读性
添加测试
测量性能
最后优化
不要凭感觉优化。
使用:
from time import perf_counter
或者:
from timeit import timeit
程序级分析:
python -m cProfile -s cumulative main.py
Python 标准库已经包含大量成熟工具:
pathlib
json
csv
decimal
datetime
collections
itertools
functools
logging
tempfile
shutil
concurrent.futures
asyncio
dataclasses
typing
引入第三方库前,先确认标准库是否已经能解决问题。
推荐:
codes = [
stock["code"]
for stock in stocks
if stock["price"] > 100
]
不推荐:
result = [
transform(item)
for group in groups
if validate_group(group)
for item in group
if validate_item(item)
if another_condition(item)
]
复杂逻辑改用普通循环:
result = []
for group in groups:
if not validate_group(group):
continue
for item in group:
if not validate_item(item):
continue
if not another_condition(item):
continue
result.append(
transform(item)
)
适合:
stocks.sort(
key=lambda stock: stock.price,
)
不适合复杂逻辑:
lambda item: (
complicated_result(item)
if condition(item)
else another_result(item)
)
复杂逻辑使用命名函数:
def calculate_score(
stock: Stock,
) -> float:
...
不推荐:
def process_code(
code: str,
) -> str | None:
if code:
if len(code) == 6:
if code.isdigit():
return code
return None
推荐:
def process_code(
code: str,
) -> str | None:
if not code:
return None
if len(code) != 6:
return None
if not code.isdigit():
return None
return code
这种写法也称为守卫语句。
不推荐:
if direction == "buy":
...
字符串可能拼错:
"BUY"
"Buy"
"byu"
推荐:
from enum import StrEnum
class Direction(StrEnum):
BUY = "buy"
SELL = "sell"
使用:
def submit_order(
direction: Direction,
) -> None:
if direction is Direction.BUY:
...
不推荐:
if retry_count >= 3:
...
推荐:
MAX_RETRY_COUNT = 3
if retry_count >= MAX_RETRY_COUNT:
...
更复杂配置:
@dataclass(frozen=True)
class RetryConfig:
max_attempts: int = 3
delay_seconds: float = 1.0
不推荐:
class StockService:
def find_stock(
self,
code: str,
) -> Stock | None:
database = PostgreSQLDatabase()
return database.find(code)
推荐:
class StockService:
def __init__(
self,
repository: StockRepository,
) -> None:
self.repository = repository
def find_stock(
self,
code: str,
) -> Stock | None:
return self.repository.find(
code
)
优点:
- 容易替换数据库
- 容易编写测试
- 降低模块依赖
- 支持模拟对象
最简单的测试:
def add(
a: int,
b: int,
) -> int:
return a + b
assert add(1, 2) == 3
正式项目可以使用:
unittest
pytest
测试应覆盖:
正常输入
边界输入
空数据
错误类型
异常情况
极端数据
例如:
import unittest
class TestValidateStockCode(
unittest.TestCase
):
def test_valid_code(
self,
) -> None:
self.assertEqual(
validate_stock_code(
"600519"
),
"600519",
)
def test_invalid_length(
self,
) -> None:
with self.assertRaises(
ValueError
):
validate_stock_code(
"60051"
)
不推荐测试直接连接:
- 正式数据库
- 真实邮件服务
- 实际支付接口
- 外部网络 API
应使用:
- 测试数据库
- 临时目录
- 模拟对象
- 依赖注入
- Mock
文件测试可以使用:
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as directory:
file_path = (
Path(directory)
/ "data.json"
)
推荐:
stock_project/
├── pyproject.toml
├── README.md
├── src/
│ └── stock_project/
│ ├── __init__.py
│ ├── models.py
│ ├── services.py
│ ├── repositories.py
│ ├── exceptions.py
│ ├── config.py
│ └── main.py
└── tests/
├── test_models.py
├── test_services.py
└── test_repositories.py
常见职责:
models.py 数据模型
services.py 业务逻辑
repositories.py 数据访问
exceptions.py 自定义异常
config.py 配置处理
main.py 程序入口
tests/ 自动化测试
例如:
models.py 导入 services.py
services.py 又导入 models.py
容易产生循环导入。
解决方式:
- 重新划分模块职责
- 抽取公共接口
- 把类型放到独立模块
- 仅在类型检查时导入
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .models import Stock
推荐:
def main() -> int:
run_application()
return 0
if __name__ == "__main__":
raise SystemExit(main())
优点:
- 模块被导入时不会自动执行
- 便于测试
- 可以返回退出码
- 程序结构清晰
不推荐:
DATABASE_HOST = "192.168.1.10"
DATABASE_PASSWORD = "123456"
敏感配置应来自:
- 环境变量
- 配置文件
- 密钥管理系统
读取环境变量:
import os
database_host = os.environ.get(
"DATABASE_HOST",
"localhost",
)
必需配置:
database_password = os.environ[
"DATABASE_PASSWORD"
]
不要把密码、Token、API Key 提交到代码仓库。
日志中不要输出:
密码
API 密钥
访问令牌
身份证号码
银行卡号
完整用户隐私信息
数据库连接密码
不推荐:
logger.info(
"连接数据库,密码:%s",
password,
)
推荐只记录必要信息:
logger.info(
"开始连接数据库:%s",
database_host,
)
危险:
result = eval(
user_input
)
用户输入可能执行任意代码。
解析数据应使用专用工具:
json.loads()
int()
float()
Decimal()
ast.literal_eval()
即使 ast.literal_eval() 相对安全,也只应用于可信格式数据。
不推荐:
sql = (
"SELECT * FROM users "
f"WHERE name = '{name}'"
)
存在 SQL 注入风险。
推荐:
cursor.execute(
"""
SELECT *
FROM users
WHERE name = %s
""",
(name,),
)
不同数据库驱动占位符可能不同,应遵循对应驱动文档。
不推荐:
response = requests.get(url)
网络可能永久等待。
推荐:
response = requests.get(
url,
timeout=10,
)
重试只适合临时错误:
- 超时
- 短暂网络故障
- 服务器临时不可用
不应该重试:
- 参数错误
- 身份认证失败
- 权限不足
- 数据格式错误
一般原则:
网络和文件等待 多线程或 asyncio
CPU 密集计算 多进程
简单任务 保持同步
不要因为异步看起来高级就全部使用 asyncio。
异步适合:
- 大量网络请求
- 并发数据库操作
- WebSocket
- 消息队列
- 高并发服务
纯计算任务使用异步通常不会变快。
并发代码中不推荐多个线程共同修改:
shared_list = []
shared_count = 0
优先让每个任务返回结果:
def process_item(
item: str,
) -> Result:
...
然后在主线程统一汇总。
必要时使用:
Lock- 队列
- 不可变对象
- 消息传递
def calculate_profit(
buy_price: float,
sell_price: float,
quantity: int,
) -> float:
"""
计算股票交易毛利润。
参数:
buy_price: 买入价格。
sell_price: 卖出价格。
quantity: 股票数量。
返回:
未扣除手续费和税费的毛利润。
异常:
ValueError: 价格或数量小于零。
"""
...
简单函数可以使用单行文档:
def is_even(
number: int,
) -> bool:
"""判断整数是否为偶数。"""
return number % 2 == 0
不要写毫无信息量的文档:
"""计算数据。"""
不推荐:
## 数量加一
quantity += 1
代码本身已经说明做了什么。
推荐:
## API 页码从 1 开始,而列表索引从 0 开始。
page_number = index + 1
注释应该解释:
- 为什么这样设计
- 业务背景
- 非直观限制
- 临时兼容原因
同一个项目中保持统一:
命名方式
异常风格
日志格式
类型提示
目录结构
配置方式
测试方式
不要同一个项目中混用:
getUser()
get_user()
GetUser()
统一比个人偏好更重要。
推荐工具:
ruff 代码检查和格式化
mypy 静态类型检查
pytest 自动化测试
coverage 测试覆盖率
示例 pyproject.toml:
[project]
name = "stock-project"
version = "0.1.0"
requires-python = ">=3.12"
[project.optional-dependencies]
dev = [
"ruff",
"mypy",
"pytest",
"coverage",
]
[tool.ruff]
line-length = 88
[tool.ruff.format]
quote-style = "double"
[tool.mypy]
python_version = "3.12"
strict = true
常用命令:
ruff check .
ruff format .
mypy src
pytest
coverage run -m pytest
coverage report
不要在不同环境中随意安装不同版本。
可以在 pyproject.toml 中声明依赖:
[project]
dependencies = [
"requests>=2.32,<3",
"openpyxl>=3.1,<4",
]
虚拟环境:
python -m venv .venv
激活:
macOS 或 Linux:
source .venv/bin/activate
Windows:
.venv\Scripts\activate
安装:
python -m pip install -e .
不推荐:
def normalize_codes(
codes: list[str],
) -> None:
for index, code in enumerate(
codes
):
codes[index] = code.strip()
调用者的数据被悄悄修改。
推荐返回新列表:
def normalize_codes(
codes: list[str],
) -> list[str]:
return [
code.strip()
for code in codes
]
如果函数设计就是原地修改,应从名称或文档中明确表达:
def normalize_codes_in_place(
codes: list[str],
) -> None:
...
不推荐:
def find_stock(
code: str,
):
if found:
return {
"code": code,
}
return False
函数可能返回字典或布尔值,调用方很难处理。
推荐:
def find_stock(
code: str,
) -> dict[str, object] | None:
if found:
return {
"code": code,
}
return None
或者抛出明确异常:
raise StockNotFoundError(code)
下面接口过于简单:
def save_data() -> bool:
try:
...
return True
except Exception:
return False
调用方不知道为什么失败。
更推荐:
def save_data(
data: object,
) -> None:
...
失败时抛出异常。
调用方:
try:
save_data(data)
except PermissionError as error:
logger.error(
"保存失败:%s",
error,
)
批量任务确实需要返回状态时,可以定义结果对象:
from dataclasses import dataclass
@dataclass(slots=True)
class OperationResult:
success: bool
message: str
error: Exception | None = None
不推荐一次性返回大量数据:
def load_lines(
file_path: Path,
) -> list[str]:
return file_path.read_text(
encoding="utf-8",
).splitlines()
大文件推荐:
from collections.abc import Iterator
def read_lines(
file_path: Path,
) -> Iterator[str]:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line in file:
yield line.rstrip("\n")
不推荐大量裸字典:
stock = {
"code": "600519",
"name": "贵州茅台",
"price": 1500.0,
}
长期业务对象推荐:
@dataclass(
frozen=True,
slots=True,
)
class Stock:
code: str
name: str
price: Decimal
优点:
- 字段明确
- 类型明确
- 编辑器提示
- 更容易校验
- 更容易重构
业务函数不需要层层捕获所有异常。
可以在程序入口统一处理:
import logging
logger = logging.getLogger(
__name__
)
def main() -> int:
try:
run_application()
except ConfigurationError as error:
logger.error(
"配置错误:%s",
error,
)
return 1
except KeyboardInterrupt:
logger.info(
"用户中断程序"
)
return 130
except Exception:
logger.exception(
"程序发生未处理异常"
)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
下面是一个较完整的股票 JSON 数据处理示例。
from dataclasses import asdict, dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
import json
import logging
import os
import tempfile
logger = logging.getLogger(__name__)
class StockDataError(Exception):
"""股票数据异常。"""
@dataclass(
frozen=True,
slots=True,
)
class Stock:
code: str
name: str
price: Decimal
def parse_stock(
value: object,
) -> Stock:
if not isinstance(value, dict):
raise StockDataError(
"股票记录必须是对象"
)
try:
raw_code = value["code"]
raw_name = value["name"]
raw_price = value["price"]
except KeyError as error:
raise StockDataError(
f"缺少字段:{error.args[0]}"
) from error
code = str(raw_code).strip()
name = str(raw_name).strip()
if (
len(code) != 6
or not code.isdigit()
):
raise StockDataError(
f"股票代码无效:{code!r}"
)
if not name:
raise StockDataError(
"股票名称不能为空"
)
try:
price = Decimal(
str(raw_price)
)
except InvalidOperation as error:
raise StockDataError(
f"股票价格无效:"
f"{raw_price!r}"
) from error
if not price.is_finite():
raise StockDataError(
"股票价格必须是有限数值"
)
if price < 0:
raise StockDataError(
"股票价格不能小于零"
)
return Stock(
code=code,
name=name,
price=price,
)
def read_json(
file_path: Path,
) -> object:
try:
text = file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError as error:
raise StockDataError(
f"文件不存在:{file_path}"
) from error
except PermissionError as error:
raise StockDataError(
f"没有读取权限:"
f"{file_path}"
) from error
try:
return json.loads(text)
except json.JSONDecodeError as error:
raise StockDataError(
"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列"
) from error
def process_stocks(
raw_data: object,
) -> tuple[
list[Stock],
list[tuple[int, str]],
]:
if not isinstance(raw_data, list):
raise StockDataError(
"JSON 根节点必须是数组"
)
stocks: list[Stock] = []
errors: list[
tuple[int, str]
] = []
seen_codes: set[str] = set()
for index, item in enumerate(
raw_data,
start=1,
):
try:
stock = parse_stock(item)
except StockDataError as error:
errors.append(
(
index,
str(error),
)
)
continue
if stock.code in seen_codes:
continue
seen_codes.add(stock.code)
stocks.append(stock)
stocks.sort(
key=lambda stock: stock.code
)
return stocks, errors
def json_default(
value: object,
) -> object:
if isinstance(value, Decimal):
return str(value)
raise TypeError(
f"不支持序列化类型:"
f"{type(value).__name__}"
)
def safe_write_json(
file_path: Path,
data: object,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
allow_nan=False,
default=json_default,
)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=file_path.parent,
delete=False,
) as temp_file:
temp_file.write(text)
temp_path = Path(
temp_file.name
)
try:
os.replace(
temp_path,
file_path,
)
except Exception:
temp_path.unlink(
missing_ok=True,
)
raise
def main() -> int:
input_path = Path(
"stocks.json"
)
output_path = Path(
"output/stocks.json"
)
try:
raw_data = read_json(
input_path
)
stocks, errors = (
process_stocks(
raw_data
)
)
output_data = [
asdict(stock)
for stock in stocks
]
safe_write_json(
output_path,
output_data,
)
except StockDataError as error:
logger.error(
"股票数据处理失败:%s",
error,
)
return 1
except OSError:
logger.exception(
"文件保存失败"
)
return 1
logger.info(
"成功保存 %s 条股票数据",
len(stocks),
)
for index, error in errors:
logger.warning(
"第 %s 条数据错误:%s",
index,
error,
)
return 0
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s "
"%(levelname)s "
"%(message)s"
),
)
raise SystemExit(main())
这个案例体现了:
类型提示
数据类
Decimal
自定义异常
异常链
JSON 结构校验
业务数据校验
函数职责拆分
安全文件写入
日志记录
程序退出码
应尽量避免:
整个函数都使用 except Exception
大量使用全局变量
函数同时做很多事情
函数参数数量过多
使用可变默认参数
所有错误都返回 False
使用 print 代替日志
硬编码路径和密码
重要文件直接覆盖
大文件一次性读入内存
复杂代码没有测试
用 eval 处理用户输入
使用字符串拼接 SQL
过早进行性能优化
为了高级语法而牺牲可读性
提交代码前可以检查:
变量名和函数名是否清晰?
函数是否只负责一个任务?
参数是否过多?
返回值类型是否稳定?
是否使用了可变默认参数?
异常捕获是否过宽?
是否保留了异常原因?
是否应该使用日志而不是 print?
文件是否明确指定编码?
路径是否使用 pathlib?
重要文件是否安全写入?
金额是否应该使用 Decimal?
是否有输入和数据校验?
是否泄露敏感信息?
是否存在重复代码?
复杂逻辑是否有测试?
是否进行了不必要的优化?
一、可读性优先于少写代码。
二、函数只做一件事。
三、显式传递依赖,少用全局变量。
四、类型提示和数据模型保持清晰。
五、只捕获能够处理的异常。
六、正式项目使用日志而不是 print。
七、文件和路径优先使用 pathlib 与 with。
八、金额和精确数值使用 Decimal。
九、先写测试和正确代码,再优化性能。
十、保持简单,不为高级而高级。
Python 最佳实践的最终目标不是让代码看起来“专业”,而是:
让代码在几个月以后仍然容易理解,在出现问题时容易定位,在需求变化时容易修改。