李猪兰
0%

Python代码规范

发表于 更新于 分类于 ，
本文字数： 16k 阅读时长 ≈ 15 分钟

Python Code Review 检查事项

一、审查目标

  1. 代码可读性检查

  2. 代码组织结构和风格一致性检查

  3. 代码最佳实践检查和推广

  4. 静态代码分析工具(pylint/flake8等), 或者Pycharm等提示的警告必须检查并修复

二、代码审查重点

  1. 代码风格

    1. 是否符合PEP 8规范

    2. 命名是否规范且有意义

    3. 代码格式是否一致

  2. 代码质量

    1. 是否有适当的错误处理

    2. 是否有充分的测试覆盖

    3. 是否有必要的文档注释

    4. 是否存在代码重复

  3. 业务逻辑

    1. 业务逻辑是否正确

    2. 是否考虑了边界情况

    3. 是否有性能问题

    4. 是否有安全隐患

一、Python PEP8 代码规范（重点）

1.1 代码布局

  • 缩进使用4个空格（不使用Tab）

  • 每行最大长度限制在79个字符（文档字符串/注释）或88个字符（代码）

  • 顶层函数和类定义之间空两行

  • 类中的方法定义之间空一行

  • 函数内逻辑块之间空一行

  • 括号内的参数不要有空格

  • 逗号、冒号、分号前不要有空格

  • 操作符两边要有空格：x = 1

  • 不要在行尾加空格

  • 文件必须使用UTF-8编码

  • 文件要以一个空行结束

1.2 命名规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 包名和模块名：简短的、全小写的名字
import os
import sys_info
 
# 类名：首字母大写的驼峰命名 
class UserProfile: 
	pass
	 
# 函数名和变量名：小写字母和下划线 
def calculate_total(): 
	user_count = 0
	 
# 常量：全大写字母和下划线 
MAX_CONNECTIONS = 100 
DEFAULT_TIMEOUT = 30
 
# 保护变量：单下划线开头 
_internal_value = 10
 
# 私有变量：双下划线开头 
__private_method = 20

1.3 导入规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# 1. 标准库导入
import os 
import sys 
from datetime import datetime 

# 2. 第三方库导入 
import numpy as np 
import pandas as pd 
from sqlalchemy import create_engine 

# 3. 本地应用/库的特定导入 
from mypackage import example 
from core.models import User 

# 不推荐的导入方式 
from module import * # 避免使用通配符导入

1.4 字符串规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 字符串引号使用
single_quoted = 'single' # 简单字符串用单引号
double_quoted = "double" # 包含单引号的字符串用双引号
triple_quoted = '''
	多行字符串使用三引号
	可以跨越多行
	'''

# 字符串格式化（按推荐顺序）
name = "World"
# 1. f-string（Python 3.6+）
greeting = f"Hello {name}"

# 2. str.format()
greeting = "Hello {}".format(name)

# 3. %-formatting（不推荐）
greeting = "Hello %s" % name

二、Python实践规范（重点）

2.1 基本设计规范

2.1.1 函数定义

  • 函数应该做一件事，并且做好这件事

  • 函数的参数最好少于5个

  • 函数长度最好不要超过50行

  • 函数应该有清晰的输入和输出

1
2
3
def process_data(data, type, validate=True, user=None, logger=None):
	# 函数参数过多
	pass
1
2
3
4
5
6
7
8
9
10
@dataclass
class ProcessConfig:
	type: str
	validate: bool = True
	user: Optional[str] = None
	logger: Optional[Logger] = None

def process_data(data: dict, config: ProcessConfig) -> dict:
"""处理数据并返回结果"""
	pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# 推荐2: 使用**kwargs
def process_data(data: dict, **kwargs) -> dict:
	"""
	处理数据并返回结果
	支持kwargs方式灵活传参
	"""
pass
## 使用示例
params = {
	"data": data,
	"type": "test",
	"validate": False
}
result = process_data(**params)

2.1.2 类型注解

1
2
3
4
5
6
7
from typing import List, Optional, Dict
def get_user_info(user_id: int) -> Optional[Dict[str, any]]:
	"""获取用户信息"""
	pass
def process_items(items: List[str]) -> List[str]:
	"""处理项目列表"""
	return [item.upper() for item in items]

2.1.3 类的设计规范

  • 遵循单一职责原则（SRP）

  • 使用封装隐藏实现细节

  • 提供清晰的公共接口

  • 合理使用继承和组合

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 好的示例：职责单一，封装良好
class User:
	def __init__(self, name: str, email: str):
		self._name = name
		self._email = email
		self._is_active = False
	@property
	def name(self) -> str:
		return self._name
	@property
	def email(self) -> str:
		return self._email
	@property
	def is_active(self) -> bool:
		return self._is_active
	def activate(self):
		"""激活用户"""
		self._is_active = True
	def deactivate(self):
		"""停用用户"""
		self._is_active = False

2.1.4 接口设计规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from abc import ABC, abstractmethod
class PaymentProcessor(ABC):
	@abstractmethod
	def process_payment(self, amount: Decimal) -> bool:
		"""处理支付"""
		pass
	@abstractmethod
	def refund(self, transaction_id: str) -> bool:
		"""处理退款"""
		pass
class StripePaymentProcessor(PaymentProcessor):
	def process_payment(self, amount: Decimal) -> bool:
		# Stripe支付实现
		pass
	def refund(self, transaction_id: str) -> bool:
		# Stripe退款实现
		pass

2.2 最佳实践

2.2.1 枚举类型的最佳实践

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from enum import Enum, auto

# 不好的示例：使用普通常量
PAYMENT_STATUS_PENDING = 1
PAYMENT_STATUS_SUCCESS = 2
PAYMENT_STATUS_FAILED = 3

def process_payment(status: int):
    if status == PAYMENT_STATUS_PENDING:
        pass
    # 容易输入任意整数，不安全

# 好的示例：使用枚举类
class PaymentStatus(Enum):
    PENDING = auto()
    SUCCESS = auto()
    FAILED = auto()
    
    @classmethod
    def is_final_status(cls, status: 'PaymentStatus') -> bool:
        """判断是否为最终状态"""
        return status in (cls.SUCCESS, cls.FAILED)

# 枚举类的高级用法
class OrderStatus(Enum):
    CREATED = ('创建', 10)
    PAID = ('已支付', 20)
    SHIPPED = ('已发货', 30)
    COMPLETED = ('已完成', 40)
    
    def __init__(self, label: str, sequence: int):
        self.label = label
        self.sequence = sequence
    
    def can_transition_to(self, new_status: 'OrderStatus') -> bool:
        """判断是否可以转换到新状态"""
        return self.sequence < new_status.sequence

2.2.2 保持函数抽象层次一致

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# 不好的示例：混合了不同抽象层次
class ReportCard:
    def generate_report(self):
        # 打印头部
        print("=== 成绩单 ===")
        print(f"学生：{self.student_name}")
        
        # 直接在这里处理成绩统计的细节
        total = 0
        for course in self.courses:
            print(f"{course.name}: {course.score}")
            total += course.score
        
        avg = total / len(self.courses) if self.courses else 0
        print(f"\n平均分：{avg:.2f}")

# 好的示例：保持一致的抽象层次
class ReportCard:
    def generate_report(self):
        """顶层函数只包含主要流程"""
        self._print_header()
        self._print_scores()
        avg = self._calculate_average()
        self._print_footer(avg)
    
    def _print_header(self):
        """处理具体的打印细节"""
        print("=== 成绩单 ===")
        print(f"学生：{self.student_name}")
    
    def _print_scores(self):
        """处理具体的成绩打印"""
        for course in self.courses:
            print(f"{course.name}: {course.score}")
    
    def _calculate_average(self) -> float:
        """处理具体的计算逻辑"""
        if not self.courses:
            return 0.0
        return sum(course.score for course in self.courses) / len(self.courses)
    
    def _print_footer(self, average: float):
        """处理具体的页脚打印"""
        print(f"\n平均分：{average:.2f}")

2.2.3 避免复杂的条件表达式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
# 不好的示例：混合了不同抽象层次
class ReportCard:
    def generate_report(self):
        # 打印头部
        print("=== 成绩单 ===")
        print(f"学生：{self.student_name}")
        
        # 直接在这里处理成绩统计的细节
        total = 0
        for course in self.courses:
            print(f"{course.name}: {course.score}")
            total += course.score
        
        avg = total / len(self.courses) if self.courses else 0
        print(f"\n平均分：{avg:.2f}")

# 好的示例：保持一致的抽象层次
class ReportCard:
    def generate_report(self):
        """顶层函数只包含主要流程"""
        self._print_header()
        self._print_scores()
        avg = self._calculate_average()
        self._print_footer(avg)
    
    def _print_header(self):
        """处理具体的打印细节"""
        print("=== 成绩单 ===")
        print(f"学生：{self.student_name}")
    
    def _print_scores(self):
        """处理具体的成绩打印"""
        for course in self.courses:
            print(f"{course.name}: {course.score}")
    
    def _calculate_average(self) -> float:
        """处理具体的计算逻辑"""
        if not self.courses:
            return 0.0
        return sum(course.score for course in self.courses) / len(self.courses)
    
    def _print_footer(self, average: float):
        """处理具体的页脚打印"""
        print(f"\n平均分：{average:.2f}")

2.2.4 命令查询分离（CQS）原则

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 不好的示例：混合命令和查询
class UserManager:
    def get_and_update_last_login(self, user_id: int) -> User:
        """获取用户信息并更新最后登录时间"""
        user = self.db.get_user(user_id)
        if user:
            user.last_login = datetime.now()
            self.db.save(user)
        return user

# 好的示例：分离命令和查询
class UserManager:
    def get_user(self, user_id: int) -> Optional[User]:
        """查询：获取用户信息"""
        return self.db.get_user(user_id)
    
    def update_last_login(self, user_id: int) -> None:
        """命令：更新最后登录时间"""
        user = self.db.get_user(user_id)
        if user:
            user.last_login = datetime.now()
            self.db.save(user)

2.2.5 保持单一职责

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# 不好的示例：一个函数做多件事
def process_order(order):
    # 验证订单
    if not order.items:
        raise ValueError("订单为空")
    
    # 计算总价
    total = 0
    for item in order.items:
        total += item.price * item.quantity
    
    # 扣减库存
    for item in order.items:
        item.product.stock -= item.quantity
        db.save(item.product)
    
    # 创建支付记录
    payment = Payment(order_id=order.id, amount=total)
    db.save(payment)

# 好的示例：职责分离
class OrderProcessor:
    def process_order(self, order):
        """主流程只负责调度"""
        self._validate_order(order)
        total = self._calculate_total(order)
        self._update_inventory(order)
        self._create_payment(order, total)
    
    def _validate_order(self, order):
        """只负责验证"""
        if not order.items:
            raise ValueError("订单为空")
    
    def _calculate_total(self, order) -> Decimal:
        """只负责计算总价"""
        return sum(
            item.price * item.quantity 
            for item in order.items
        )
    
    def _update_inventory(self, order):
        """只负责更新库存"""
        for item in order.items:
            item.product.stock -= item.quantity
            db.save(item.product)
    
    def _create_payment(self, order, total):
        """只负责创建支付记录"""
        payment = Payment(order_id=order.id, amount=total)
        db.save(payment)

2.3 错误处理规范

2.3.1 异常处理基本原则

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 不推荐：捕获所有异常
try:
    do_something()
except Exception as e:
    print(f"Error: {e}")  # 1. 使用print而不是logger
                          # 2. 吞掉异常而不是处理或重新抛出
                          # 3. 没有提供上下文信息

# 推荐：具体异常处理
try:
    do_something()
except ValueError as e:
    logger.error(f"处理数据时发生值错误: {e}", extra={
        'action': 'process_data',
        'input_data': input_data,
        'error_type': 'ValueError'
    })
    raise  # 重新抛出异常，保持调用栈
except IOError as e:
    logger.error(f"读取文件时发生IO错误: {e}", extra={
        'action': 'read_file',
        'file_path': file_path,
        'error_type': 'IOError'
    })
    raise CustomIOError(f"无法读取文件: {file_path}") from e  # 使用from保留原始异常
finally:
    cleanup_resources()  # 确保资源被清理

2.3.2 异常处理最佳实践

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
# 1. 使用上下文管理器自动处理资源
# 不推荐
f = open('file.txt', 'r')
try:
    data = f.read()
finally:
    f.close()

# 推荐
with open('file.txt', 'r') as f:
    data = f.read()

# 2. 自定义上下文管理器
from contextlib import contextmanager
import time

@contextmanager
def operation_timer(operation_name: str):
    """计时器上下文管理器"""
    start_time = time.time()
    try:
        yield
    finally:
        elapsed_time = time.time() - start_time
        logger.info(f"{operation_name} 耗时: {elapsed_time:.2f}秒")

# 使用示例
with operation_timer("数据处理"):
    process_large_dataset()

# 3. 异常重试机制
from functools import wraps
from typing import Type, Tuple, Optional
import time

def retry(
    exceptions: Tuple[Type[Exception], ...],
    tries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    logger: Optional[logging.Logger] = None
):
    """重试装饰器
    
    Args:
        exceptions: 需要重试的异常类型
        tries: 最大重试次数
        delay: 初始延迟时间
        backoff: 延迟时间的增长因子
        logger: 日志记录器
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            _tries, _delay = tries, delay
            while _tries > 1:
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    msg = f"{func.__name__} 失败，{_tries-1}次重试剩余，异常: {e}"
                    if logger:
                        logger.warning(msg)
                    time.sleep(_delay)
                    _tries -= 1
                    _delay *= backoff
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 使用示例
@retry((RequestException, ConnectionError), tries=3, delay=1)
def fetch_data_from_api(url: str) -> dict:
    return requests.get(url).json()

2.3.3 自定义异常层次结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class BusinessError(Exception):
    """业务逻辑相关的异常基类"""
    def __init__(self, message: str, error_code: str = None):
        super().__init__(message)
        self.error_code = error_code
        self.message = message

class ValidationError(BusinessError):
    """数据验证异常"""
    def __init__(self, message: str, field: str = None):
        super().__init__(message, error_code="VALIDATION_ERROR")
        self.field = field

class ResourceError(BusinessError):
    """资源相关异常基类"""
    pass

class ResourceNotFoundError(ResourceError):
    """资源未找到异常"""
    def __init__(self, resource_type: str, resource_id: str):
        message = f"未找到{resource_type}: {resource_id}"
        super().__init__(message, error_code="RESOURCE_NOT_FOUND")
        self.resource_type = resource_type
        self.resource_id = resource_id

class ResourceConflictError(ResourceError):
    """资源冲突异常"""
    def __init__(self, resource_type: str, conflict_detail: str):
        message = f"{resource_type}资源冲突: {conflict_detail}"
        super().__init__(message, error_code="RESOURCE_CONFLICT")
        
 # 使用示例
def create_user(user_data: dict) -> User:
    try:
        validate_user_data(user_data)
    except ValidationError as e:
        logger.error(f"用户数据验证失败: {e.message}", extra={
            'field': e.field,
            'error_code': e.error_code
        })
        raise

    try:
        return user_repository.create(user_data)
    except DuplicateKeyError as e:
        raise ResourceConflictError(
            resource_type="User",
            conflict_detail=f"邮箱 {user_data['email']} 已存在"
        ) from e

2.4 性能优化规范

2.4.1 数据结构使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
# 1. 使用集合进行成员检查
# 不推荐
if item in large_list:  # O(n)
    pass

# 推荐
item_set = set(large_list)  # 转换成本O(n)
if item in item_set:  # O(1)
    pass

# 2. 使用生成器处理大数据
def process_large_file(file_path: str):
    with open(file_path) as f:
        for line in f:  # 逐行处理，避免一次性加载
            yield process_line(line)
            
# 3. 使用deque进行队列操作
# 不推荐：使用列表作为队列
queue = []
queue.append(item)      # O(1)
queue.pop(0)            # O(n)

# 推荐：使用deque
queue: Deque = deque()
queue.append(item)      # O(1)
queue.popleft()         # O(1)

# 4. 集合运算
# 不推荐：使用列表进行集合运算
common_items = [x for x in list1 if x in list2]  # O(n^2)

# 推荐：使用集合进行集合运算
set1, set2 = set(list1), set(list2)
common_items = set1 & set2  # O(min(len(set1), len(set2)))
unique_items = set1 | set2  # 并集
only_in_set1 = set1 - set2  # 差集

2.4.2 循环优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# 1. 列表推导式与生成器表达式
# 不推荐：在不需要列表的场合使用列表推导式
sum([x * x for x in range(1000)])  # 创建了不必要的列表

# 推荐：使用生成器表达式
sum(x * x for x in range(1000))  # 内存效率更高

# 2. 多重循环优化
# 不推荐：嵌套循环
result = []
for x in range(1000):
    for y in range(1000):
        if x * y > 100:
            result.append((x, y))

# 推荐：使用生成器表达式和any/all
result = ((x, y) 
          for x in range(1000) 
          for y in range(1000) 
          if x * y > 100)

# 3. 字典推导式
# 不推荐
square_dict = {}
for x in range(100):
    square_dict[x] = x * x

# 推荐
square_dict = {x: x * x for x in range(100)}

# 4. 批量数据处理
# 不推荐：一次性处理所有数据
def process_large_dataset(items: List[dict]) -> List[dict]:
    return [process_item(item) for item in items]  # 可能消耗大量内存

# 推荐：分批处理
from typing import Iterator, TypeVar, List
T = TypeVar('T')

def batch_processor(items: Iterator[T], batch_size: int) -> Iterator[List[T]]:
    batch = []
    for item in items:
        batch.append(item)
        if len(batch) >= batch_size:
            yield batch
            batch = []
    if batch:
        yield batch

def process_large_dataset(items: Iterator[dict], batch_size: int = 1000):
    for batch in batch_processor(items, batch_size):
        processed_batch = [process_item(item) for item in batch]
        save_to_database(processed_batch)

2.4.3 字符串操作优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
# 1. 字符串拼接
# 不推荐：在循环中使用+=
result = ""
for item in items:
    result += str(item)  # 每次都创建新字符串

# 推荐：使用join
result = "".join(str(item) for item in items)

# 2. 字符串格式化
# 不推荐：%-格式化
message = "Hello, %s. You have %d new messages." % (name, count)

# 推荐：f-string（Python 3.6+）
message = f"Hello, {name}. You have {count} new messages."

# 3. 字符串分割和合并
# 不推荐：多次调用split
parts = text.split()[0].split(',')[1].split('-')[0]

# 推荐：使用正则表达式
import re
pattern = re.compile(r'^[^,]*,([^-]*)')
match = pattern.search(text)
if match:
    part = match.group(1)

# 4. 字符串替换
# 不推荐：多次replace
text = text.replace('old', 'new').replace('foo', 'bar').replace('baz', 'qux')

# 推荐：使用正则表达式或str.translate
replacements = {'old': 'new', 'foo': 'bar', 'baz': 'qux'}
pattern = re.compile('|'.join(map(re.escape, replacements.keys())))
text = pattern.sub(lambda m: replacements[m.group()], text)

# 或使用translate
trans_table = str.maketrans(replacements)
text = text.translate(trans_table)

三、其他规范

3.1 测试规范

3.1.1 测试原则

  • 测试应该是独立的

  • 测试应该是可重复的

  • 测试应该是简单的

  • 测试应该是有意义的

3.1.2 测试工具

  • pytest: 单元测试框架

  • unittest: 标准库测试框架

3.2 文档规范

3.2.1 代码注释

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 1. 函数文档
def complex_function(param1: str, param2: int) -> bool:
    """函数的简短描述。

    详细的函数说明，可以多行。

    Args:
        param1: 参数1的说明
        param2: 参数2的说明

    Returns:
        返回值的说明

    Raises:
        ValueError: 异常情况说明
    """
    pass

# 2. 类文档
class ComplexClass:
    """类的简短描述。

    详细的类说明，包括主要功能、使用场景等。

    Attributes:
        attr1: 属性1的说明
        attr2: 属性2的说明
    """
    pass

3.3 安全规范

3.3.1 输入验证

1
2
3
4
5
6
7
8
9
10
11
def process_user_input(user_input: str) -> str:
    # 1. 长度检查
    if len(user_input) > MAX_LENGTH:
        raise ValueError("输入超过最大长度限制")
    
    # 2. 格式检查
    if not VALID_INPUT_PATTERN.match(user_input):
        raise ValueError("输入格式不正确")
    
    # 3. 特殊字符过滤
    return sanitize_input(user_input)

3.3.2 SQL注入防护

1
2
3
4
5
6
7
# 不推荐
query = f"SELECT * FROM users WHERE id = {user_id}"

# 推荐
query = "SELECT * FROM users WHERE id = %s"
cursor.execute(query, (user_id,))

-------------本文结束感谢您的阅读-------------