Python工程与模块命名规范：构建可维护的大型项目架构

project-root/
├── src/                    # 源代码目录
│   └── package_name/       # 主包
│       ├── __init__.py
│       ├── core/           # 核心模块
│       │   ├── __init__.py
│       │   ├── models.py
│       │   └── utils.py
│       ├── api/            # API模块
│       │   ├── __init__.py
│       │   ├── routes.py
│       │   └── schemas.py
│       └── services/       # 服务层
│           ├── __init__.py
│           ├── user_service.py
│           └── payment_service.py
├── tests/                  # 测试目录
│   ├── __init__.py
│   ├── conftest.py
│   ├── unit/
│   │   ├── test_models.py
│   │   └── test_services.py
│   └── integration/
│       ├── test_api.py
│       └── test_database.py
├── docs/                   # 文档
├── scripts/                # 脚本文件
├── data/                   # 数据文件
├── requirements.txt        # 依赖文件
├── pyproject.toml          # 项目配置
└── README.md               # 项目说明

2.2 工程命名规范

2.2.1 项目根目录命名

# 好的项目名
financial-analysis-tool  # 使用连字符
data_processing_pipeline  # 使用下划线
imageClassifier  # 驼峰式（较少用）# 避免的项目名
myproject  # 太泛
project123  # 无意义数字
test-project  # 可能冲突

2.2.2 包和模块命名

# 包名（目录）
models/          # 复数形式
utils/           # 工具函数
services/        # 服务层
adapters/        # 适配器# 模块名（文件）
user_model.py    # 明确单一职责
database_connection.py
config_loader.py

三、详细命名约定与模式

3.1 变量命名规范

3.1.1 基础变量命名

# 基本数据类型
count = 10                  # 整数
price = 29.99               # 浮点数
name = "Alice"              # 字符串
is_active = True            # 布尔值
items = []                  # 列表
user_map = {}               # 字典# 集合类型
user_list = []              # 列表
user_dict = {}              # 字典  
user_set = set()            # 集合
user_tuple = ()             # 元组

3.1.2 复合变量命名

# 使用有意义的复合名称
max_retry_count = 3         # 最大重试次数
default_timeout_seconds = 30  # 默认超时秒数
is_user_authenticated = True  # 用户是否认证# 避免的命名
temp = get_data()           # 无意义
var1 = calculate()          # 编号命名
flag = check_status()       # 过于泛化

3.2 函数与方法命名

3.2.1 动作-对象命名模式

# 好的函数名
def calculate_total_price(items): ...
def validate_user_input(input_data): ...
def send_email_notification(recipient, message): ...
def create_user_profile(user_data): ...# 动词选择指南
# 获取数据: get_, fetch_, retrieve_, load_
# 修改数据: update_, modify_, set_, change_
# 创建数据: create_, add_, insert_, make_
# 删除数据: delete_, remove_, drop_, erase_
# 检查状态: is_, has_, can_, should_, check_

3.2.2 布尔函数命名

# 返回布尔值的函数
def is_valid_email(email): ...
def has_permission(user, resource): ...
def should_retry_request(response): ...
def can_user_access(user, feature): ...# 使用正面的布尔命名
def is_available(): ...     # 而不是 is_not_available
def has_content(): ...      # 而不是 lacks_content

3.3 类命名规范

3.3.1 类与对象命名

# 类名使用驼峰式
class UserAccount: def __init__(self, username):self.username = username  # 实例变量使用下划线self._private_data = {}   # 私有变量前加下划线class DatabaseConnection:def connect(self):passclass HTTPRequestHandler:def handle_request(self):pass# 避免的类名
class myClass: ...          # 应使用驼峰
class User_Account: ...     # 不应使用下划线

3.3.2 特殊方法命名

class Vector:def __init__(self, x, y):self.x = xself.y = y# 运算符重载def __add__(self, other):return Vector(self.x + other.x, self.y + other.y)# 字符串表示def __str__(self):return f"Vector({self.x}, {self.y})"# 容器协议def __len__(self):return 2# 上下文管理器def __enter__(self):return selfdef __exit__(self, exc_type, exc_val, exc_tb):pass

3.4 常量与配置命名

# 常量使用全大写+下划线
MAX_CONNECTIONS = 100
DEFAULT_TIMEOUT = 30
API_BASE_URL = "https://api.example.com"
SUPPORTED_FILE_FORMATS = ['.json', '.csv', '.xml']# 配置类常量
class Config:DATABASE_URL = "postgresql://user:pass@localhost/db"LOG_LEVEL = "INFO"CACHE_TIMEOUT = 3600

四、模块化设计与命名策略

4.1 模块职责划分

graph TBA[项目架构] --> B[核心模块]A --> C[业务模块]A --> D[工具模块]A --> E[接口模块]B --> B1[models.py]B --> B2[exceptions.py]B --> B3[constants.py]C --> C1[user_service.py]C --> C2[order_service.py]C --> C3[payment_service.py]D --> D1[database.py]D --> D2[logger.py]D --> D3[validators.py]E --> E1[api/E1 --> E11[routes.py]E1 --> E12[schemas.py]E1 --> E13[controllers.py]

4.2 模块命名模式

4.2.1 按功能划分模块

# 数据模型模块
# models.py
class User:passclass Product:passclass Order:pass# 服务模块
# user_service.py
class UserService:def create_user(self, user_data): ...def get_user(self, user_id): ...# 工具模块
# validation_utils.py
def validate_email(email): ...
def validate_phone(phone): ...

4.2.2 按层级划分模块

# 数据访问层
# repositories/
# user_repository.py
class UserRepository:def save(self, user): ...def find_by_id(self, user_id): ...# 业务逻辑层  
# services/
# user_service.py
class UserService:def __init__(self, user_repo):self.user_repo = user_repodef register_user(self, user_data): ...# 表示层
# web/
# controllers.py
class UserController:def post_user(self, request): ...

五、高级命名模式与技巧

5.1 设计模式相关命名

# 工厂模式
class ConnectionFactory:def create_connection(self, config): ...# 单例模式
class DatabaseManager:_instance = Nonedef __new__(cls):if cls._instance is None:cls._instance = super().__new__(cls)return cls._instance# 观察者模式
class EventPublisher:def __init__(self):self._subscribers = []def subscribe(self, subscriber): ...def notify_subscribers(self, event): ...class EventSubscriber:def on_event(self, event): ...# 策略模式
class CompressionStrategy:def compress(self, data): ...class ZipCompression(CompressionStrategy):def compress(self, data): ...class GzipCompression(CompressionStrategy):def compress(self, data): ...

5.2 领域驱动设计(DDD)命名

# 实体
class Customer:def __init__(self, customer_id, name):self.customer_id = customer_id  # 唯一标识self.name = name# 值对象
class Money:def __init__(self, amount, currency):self.amount = amountself.currency = currency# 聚合根
class Order:def __init__(self, order_id, order_items):self.order_id = order_idself.order_items = order_itemsdef add_item(self, product, quantity): ...def calculate_total(self): ...# 领域服务
class OrderProcessingService:def process_order(self, order): ...# 仓储接口
class OrderRepository:def save(self, order): ...def find_by_id(self, order_id): ...

5.3 异步编程命名

import asyncioclass AsyncDataProcessor:# 异步方法使用async前缀或相关动词async def fetch_data(self, url): ...async def process_data(self, data): ...async def save_results(self, results): ...# 回调函数命名def on_data_received(self, data): ...def on_processing_complete(self, result): ...def on_error(self, error): ...# 异步上下文管理器
class AsyncDatabaseConnection:async def __aenter__(self):await self.connect()return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):await self.close()async def connect(self): ...async def close(self): ...

六、完整项目示例

6.1 电子商务项目结构

ecommerce-platform/
├── src/
│   └── ecommerce/
│       ├── __init__.py
│       ├── core/
│       │   ├── __init__.py
│       │   ├── models.py
│       │   ├── exceptions.py
│       │   └── types.py
│       ├── domain/
│       │   ├── __init__.py
│       │   ├── product.py
│       │   ├── order.py
│       │   ├── user.py
│       │   └── payment.py
│       ├── application/
│       │   ├── __init__.py
│       │   ├── product_service.py
│       │   ├── order_service.py
│       │   ├── user_service.py
│       │   └── payment_service.py
│       ├── infrastructure/
│       │   ├── __init__.py
│       │   ├── database/
│       │   │   ├── __init__.py
│       │   │   ├── connection.py
│       │   │   ├── repositories.py
│       │   │   └── migrations/
│       │   ├── cache/
│       │   │   ├── __init__.py
│       │   │   └── redis_client.py
│       │   └── external/
│       │       ├── __init__.py
│       │       ├── email_service.py
│       │       └── payment_gateway.py
│       ├── api/
│       │   ├── __init__.py
│       │   ├── routes/
│       │   │   ├── __init__.py
│       │   │   ├── product_routes.py
│       │   │   ├── order_routes.py
│       │   │   └── user_routes.py
│       │   ├── schemas/
│       │   │   ├── __init__.py
│       │   │   ├── product_schemas.py
│       │   │   ├── order_schemas.py
│       │   │   └── user_schemas.py
│       │   └── middleware/
│       │       ├── __init__.py
│       │       ├── authentication.py
│       │       ├── logging.py
│       │       └── error_handling.py
│       └── utils/
│           ├── __init__.py
│           ├── validation.py
│           ├── logging.py
│           ├── security.py
│           └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── unit/
│   │   ├── __init__.py
│   │   ├── test_models.py
│   │   ├── test_services.py
│   │   └── test_utils.py
│   ├── integration/
│   │   ├── __init__.py
│   │   ├── test_api.py
│   │   ├── test_database.py
│   │   └── test_external_services.py
│   └── conftest.py
├── scripts/
│   ├── setup_database.py
│   ├── run_migrations.py
│   └── start_server.py
├── docs/
│   ├── api.md
│   ├── architecture.md
│   └── development.md
├── requirements/
│   ├── base.txt
│   ├── development.txt
│   └── production.txt
├── config/
│   ├── development.yaml
│   ├── production.yaml
│   └── test.yaml
├── .pre-commit-config.yaml
├── pyproject.toml
├── Dockerfile
└── README.md

6.2 核心模块代码示例

# src/ecommerce/core/models.py
"""数据模型定义"""from datetime import datetime
from typing import Optional, List
from decimal import Decimalclass BaseModel:"""所有模型的基类"""created_at: datetimeupdated_at: datetimeis_active: bool = Trueclass Product(BaseModel):"""产品实体"""def __init__(self,product_id: str,name: str,price: Decimal,description: Optional[str] = None,stock_quantity: int = 0):self.product_id = product_idself.name = nameself.price = priceself.description = descriptionself.stock_quantity = stock_quantityself.created_at = datetime.now()self.updated_at = datetime.now()def update_price(self, new_price: Decimal) -> None:"""更新产品价格"""if new_price <= 0:raise ValueError("价格必须大于0")self.price = new_priceself.updated_at = datetime.now()def reduce_stock(self, quantity: int) -> None:"""减少库存"""if quantity > self.stock_quantity:raise ValueError("库存不足")self.stock_quantity -= quantityself.updated_at = datetime.now()class Order(BaseModel):"""订单实体"""def __init__(self, order_id: str, customer_id: str):self.order_id = order_idself.customer_id = customer_idself.order_items: List[OrderItem] = []self.status = "pending"self.total_amount = Decimal('0.00')self.created_at = datetime.now()self.updated_at = datetime.now()def add_item(self, product: Product, quantity: int) -> None:"""添加订单项"""if quantity <= 0:raise ValueError("数量必须大于0")# 检查库存if quantity > product.stock_quantity:raise ValueError("产品库存不足")item_total = product.price * quantityorder_item = OrderItem(product, quantity, item_total)self.order_items.append(order_item)self.total_amount += item_totalself.updated_at = datetime.now()def calculate_total(self) -> Decimal:"""计算订单总额"""return sum(item.total_price for item in self.order_items)class OrderItem:"""订单项值对象"""def __init__(self, product: Product, quantity: int, total_price: Decimal):self.product = productself.quantity = quantityself.total_price = total_price

6.3 服务层代码示例

# src/ecommerce/application/order_service.py
"""订单服务层"""from typing import List, Optional
from decimal import Decimal
from datetime import datetime
from ..core.models import Order, Product
from ..core.exceptions import (OrderNotFoundException,ProductNotFoundException,InsufficientStockException
)class OrderService:"""订单业务服务"""def __init__(self, order_repository, product_repository):self.order_repository = order_repositoryself.product_repository = product_repositorydef create_order(self, customer_id: str, items: List[dict]) -> Order:"""创建新订单"""order = Order(order_id=self._generate_order_id(),customer_id=customer_id)for item in items:product = self.product_repository.find_by_id(item['product_id'])if not product:raise ProductNotFoundException(f"产品 {item['product_id']} 不存在")try:order.add_item(product, item['quantity'])except ValueError as e:raise InsufficientStockException(str(e))self.order_repository.save(order)return orderdef get_order(self, order_id: str) -> Order:"""获取订单详情"""order = self.order_repository.find_by_id(order_id)if not order:raise OrderNotFoundException(f"订单 {order_id} 不存在")return orderdef update_order_status(self, order_id: str, new_status: str) -> Order:"""更新订单状态"""order = self.get_order(order_id)order.status = new_statusorder.updated_at = datetime.now()self.order_repository.save(order)return orderdef calculate_order_discount(self,order: Order,discount_percentage: Decimal) -> Decimal:"""计算订单折扣"""if not 0 <= discount_percentage <= 100:raise ValueError("折扣百分比必须在0-100之间")discount_factor = discount_percentage / Decimal('100.00')return order.total_amount * discount_factordef _generate_order_id(self) -> str:"""生成订单ID"""timestamp = datetime.now().strftime("%Y%m%d%H%M%S")return f"ORD{timestamp}"

6.4 API层代码示例

# src/ecommerce/api/routes/order_routes.py
"""订单API路由"""from fastapi import APIRouter, Depends, HTTPException, status
from typing import List
from ...application.order_service import OrderService
from ...infrastructure.database.repositories import get_order_repository
from ...infrastructure.database.repositories import get_product_repository
from ..schemas.order_schemas import (OrderCreateRequest,OrderResponse,OrderStatusUpdate
)router = APIRouter(prefix="/orders", tags=["orders"])@router.post("/", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
async def create_order(request: OrderCreateRequest,order_service: OrderService = Depends(get_order_service)
) -> OrderResponse:"""创建新订单"""try:order = order_service.create_order(customer_id=request.customer_id,items=request.items)return OrderResponse.from_domain(order)except Exception as e:raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail=str(e))@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(order_id: str,order_service: OrderService = Depends(get_order_service)
) -> OrderResponse:"""获取订单详情"""try:order = order_service.get_order(order_id)return OrderResponse.from_domain(order)except Exception as e:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=str(e))@router.patch("/{order_id}/status", response_model=OrderResponse)
async def update_order_status(order_id: str,status_update: OrderStatusUpdate,order_service: OrderService = Depends(get_order_service)
) -> OrderResponse:"""更新订单状态"""try:order = order_service.update_order_status(order_id=order_id,new_status=status_update.status)return OrderResponse.from_domain(order)except Exception as e:raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,detail=str(e))def get_order_service():"""获取订单服务依赖"""order_repo = get_order_repository()product_repo = get_product_repository()return OrderService(order_repo, product_repo)

七、自动化检查与工具

7.1 使用pre-commit进行命名检查

# .pre-commit-config.yaml
repos:- repo: https://github.com/pre-commit/pre-commit-hooksrev: v4.4.0hooks:- id: trailing-whitespace- id: end-of-file-fixer- id: check-yaml- id: check-added-large-files- repo: https://github.com/PyCQA/flake8rev: 6.0.0hooks:- id: flake8additional_dependencies: [flake8-naming]- repo: https://github.com/psf/blackrev: 23.3.0hooks:- id: black- repo: https://github.com/pycqa/isortrev: 5.12.0hooks:- id: isort- repo: https://github.com/pre-commit/mirrors-mypyrev: v1.3.0hooks:- id: mypy

7.2 自定义命名检查规则

# scripts/check_naming.py
"""自定义命名检查脚本"""import ast
import re
from pathlib import Path
from typing import List, Dict, Anyclass NamingChecker(ast.NodeVisitor):"""AST节点访问器，检查命名规范"""def __init__(self):self.violations: List[Dict[str, Any]] = []self.class_names = set()self.function_names = set()def visit_ClassDef(self, node: ast.ClassDef) -> None:"""检查类命名"""if not re.match(r'^[A-Z][a-zA-Z0-9]*$', node.name):self.violations.append({'type': 'class','name': node.name,'line': node.lineno,'message': '类名应使用驼峰命名法'})self.class_names.add(node.name)self.generic_visit(node)def visit_FunctionDef(self, node: ast.FunctionDef) -> None:"""检查函数命名"""if not re.match(r'^[a-z_][a-z0-9_]*$', node.name):self.violations.append({'type': 'function','name': node.name,'line': node.lineno,'message': '函数名应使用蛇形命名法'})# 检查布尔函数命名if node.name.startswith('is_') or node.name.startswith('has_'):# 应该返回布尔值passself.function_names.add(node.name)self.generic_visit(node)def visit_Name(self, node: ast.Name) -> None:"""检查变量命名"""if isinstance(node.ctx, ast.Store):if not re.match(r'^[a-z_][a-z0-9_]*$', node.id):self.violations.append({'type': 'variable','name': node.id,'line': node.lineno,'message': '变量名应使用蛇形命名法'})self.generic_visit(node)def visit_Constant(self, node: ast.Constant) -> None:"""检查常量命名（如果被赋值）"""if hasattr(node, 'parent') and isinstance(node.parent, ast.Assign):for target in node.parent.targets:if isinstance(target, ast.Name) and target.id.isupper():if not re.match(r'^[A-Z_][A-Z0-9_]*$', target.id):self.violations.append({'type': 'constant','name': target.id,'line': node.lineno,'message': '常量名应使用全大写蛇形命名法'})def check_file_naming(file_path: Path) -> List[Dict[str, Any]]:"""检查单个文件的命名规范"""try:with open(file_path, 'r', encoding='utf-8') as f:content = f.read()tree = ast.parse(content)checker = NamingChecker()checker.visit(tree)return checker.violationsexcept Exception as e:return [{'type': 'error','name': str(file_path),'line': 0,'message': f'解析文件时出错: {str(e)}'}]def main():"""主检查函数"""project_root = Path(__file__).parent.parentpython_files = list(project_root.rglob('*.py'))all_violations = []for file_path in python_files:if 'venv' in str(file_path) or '.pytest_cache' in str(file_path):continueviolations = check_file_naming(file_path)for violation in violations:violation['file'] = str(file_path.relative_to(project_root))all_violations.append(violation)if all_violations:print("命名规范检查发现以下问题:")for violation in all_violations:print(f"{violation['file']}:{violation['line']} - "f"{violation['type']} '{violation['name']}': "f"{violation['message']}")return 1else:print("所有文件命名规范检查通过!")return 0if __name__ == "__main__":exit(main())