Pythonic 惯用法、PEP 8 标准、类型提示以及构建稳健、高效且可维护的 Python 应用程序的最佳实践。
name python-patterns description Pythonic 惯用法、PEP 8 标准、类型提示以及构建稳健、高效且可维护的 Python 应用程序的最佳实践。 origin ECC Python 开发模式 用于构建健壮、高效和可维护应用程序的惯用 Python 模式与最佳实践。 何时激活 编写新的 Python 代码 审查 Python 代码 重构现有的 Python 代码 设计 Python 包/模块 核心原则
def get_active_users ( users: list [User] ) -> list [User]: """Return only active users from the provided list.""" return [user for user in users if user.is_active]
def get_active_users ( u ): return [x for x in u if x.a] 2. 显式优于隐式 避免魔法;清晰说明你的代码在做什么。
import logging
'%(asctime)s - %(name)s - %(levelname)s - %(message)s' )
import some_module some_module.setup()
def get_value ( dictionary: dict , key: str ) -> Any : try : return dictionary[key] except KeyError: return default_value
True ) -> Optional [User]: """Process a user and return the updated User or None.""" if not active: return None return User(user_id, data) 现代类型提示(Python 3.9+)
def process_items ( items: list [ str ] ) -> dict [ str , int ]: return {item: len (item) for item in items}
from typing import List , Dict def process_items ( items: List [ str ] ) -> Dict [ str , int ]: return {item: len (item) for item in items} 类型别名和 TypeVar from typing import TypeVar, Union
JSON = Union [ dict [ str , Any ], list [ Any ], str , int , float , bool , None ] def parse_json ( data: str ) -> JSON: return json.loads(data)
T = TypeVar( 'T' ) def first ( items: list [T] ) -> T | None : """Return the first item or None if list is empty.""" return items[ 0 ] if items else None 基于协议的鸭子类型 from typing import Protocol class Renderable ( Protocol ): def render ( self ) -> str : """Render the object to a string.""" def render_all ( items: list [Renderable] ) -> str : """Render all items that implement the Renderable protocol.""" return "\n" .join(item.render() for item in items) 错误处理模式 特定异常处理
def load_config ( path: str ) -> Config: try : with open (path) as f: return Config.from_json(f.read()) except FileNotFoundError as e: raise ConfigError( f"Config file not found: {path} " ) from e except json.JSONDecodeError as e: raise ConfigError( f"Invalid JSON in config: {path} " ) from e
def load_config ( path: str ) -> Config: try : with open (path) as f: return Config.from_json(f.read()) except : return None
异常链 def process_data ( data: str ) -> Result: try : parsed = json.loads(data) except json.JSONDecodeError as e:
raise ValueError( f"Failed to parse data: {data} " ) from e 自定义异常层次结构 class AppError ( Exception ): """Base exception for all application errors.""" pass class ValidationError ( AppError ): """Raised when input validation fails.""" pass class NotFoundError ( AppError ): """Raised when a requested resource is not found.""" pass
def get_user ( user_id: str ) -> User: user = db.find_user(user_id) if not user: raise NotFoundError( f"User not found: {user_id} " ) return user 上下文管理器 资源管理
def process_file ( path: str ) -> str : with open (path, 'r' ) as f: return f.read()
def process_file ( path: str ) -> str : f = open (path, 'r' ) try : return f.read() finally : f.close() 自定义上下文管理器 from contextlib import contextmanager @contextmanager def timer ( name: str ): """Context manager to time a block of code.""" start = time.perf_counter() yield elapsed = time.perf_counter() - start print ( f" {name} took {elapsed: .4 f} seconds" )
with timer( "data processing" ): process_large_dataset() 上下文管理器类 class DatabaseTransaction : def init ( self, connection ): self .connection = connection def enter ( self ): self .connection.begin_transaction() return self def exit ( self, exc_type, exc_val, exc_tb ): if exc_type is None : self .connection.commit() else : self .connection.rollback() return False
with DatabaseTransaction(conn): user = conn.create_user(user_data) conn.create_profile(user. id , profile_data) 推导式和生成器 列表推导式
names = [user.name for user in users if user.is_active]
names = [] for user in users: if user.is_active: names.append(user.name)
0 ]
0 : result.append(x * 2 ) return result 生成器表达式
total = sum (x * x for x in range ( 1_000_000 ))
total = sum ([x * x for x in range ( 1_000_000 )]) 生成器函数 def read_large_file ( path: str ) -> Iterator[ str ]: """Read a large file line by line.""" with open (path) as f: for line in f: yield line.strip()
True
"123" , name= "Alice" , email= "[email protected]" ) 带验证的数据类 @dataclass class User : email: str age: int def post_init ( self ):
if "@" not in self .email: raise ValueError( f"Invalid email: {self.email} " )
if self .age < 0 or self .age > 150 : raise ValueError( f"Invalid age: {self.age} " ) 命名元组 from typing import NamedTuple class Point ( NamedTuple ): """Immutable 2D point.""" x: float y: float def distance ( self, other: 'Point' ) -> float : return (( self .x - other.x) ** 2
p1 = Point( 0 , 0 ) p2 = Point( 3 , 4 ) print (p1.distance(p2))
装饰器 函数装饰器 import functools import time def timer ( func: Callable ) -> Callable : """Decorator to time function execution.""" @functools.wraps( func ) def wrapper ( *args, **kwargs ): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print ( f" {func.name} took {elapsed: .4 f} s" ) return result return wrapper @timer def slow_function (): time.sleep( 1 )
参数化装饰器 def repeat ( times: int ): """Decorator to repeat a function multiple times.""" def decorator ( func: Callable ) -> Callable : @functools.wraps( func ) def wrapper ( *args, **kwargs ): results = [] for _ in range (times): results.append(func(*args, **kwargs)) return results return wrapper return decorator @repeat( times= 3 ) def greet ( name: str ) -> str : return f"Hello, {name} !"
基于类的装饰器 class CountCalls : """Decorator that counts how many times a function is called.""" def init ( self, func: Callable ): functools.update_wrapper( self , func) self .func = func self .count = 0 def call ( self, *args, **kwargs ): self .count += 1 print ( f" {self.func.name} has been called {self.count} times" ) return self .func(*args, **kwargs) @CountCalls def process (): pass
并发模式 用于 I/O 密集型任务的线程 import concurrent.futures import threading def fetch_url ( url: str ) -> str : """Fetch a URL (I/O-bound operation).""" import urllib.request with urllib.request.urlopen(url) as response: return response.read().decode() def fetch_all_urls ( urls: list [ str ] ) -> dict [ str , str ]: """Fetch multiple URLs concurrently using threads.""" with concurrent.futures.ThreadPoolExecutor(max_workers= 10 ) as executor: future_to_url = {executor.submit(fetch_url, url): url for url in urls} results = {} for future in concurrent.futures.as_completed(future_to_url): url = future_to_url[future] try : results[url] = future.result() except Exception as e: results[url] = f"Error: {e} " return results 用于 CPU 密集型任务的多进程 def process_data ( data: list [ int ] ) -> int : """CPU-intensive computation.""" return sum (x ** 2 for x in data) def process_all ( datasets: list [ list [ int ]] ) -> list [ int ]: """Process multiple datasets using multiple processes.""" with concurrent.futures.ProcessPoolExecutor() as executor: results = list (executor. map (process_data, datasets)) return results 用于并发 I/O 的异步/等待 import asyncio async def fetch_async ( url: str ) -> str : """Fetch a URL asynchronously.""" import aiohttp async with aiohttp.ClientSession() as session: async with session.get(url) as response: return await response.text() async def fetch_all ( urls: list [ str ] ) -> dict [ str , str ]: """Fetch multiple URLs concurrently.""" tasks = [fetch_async(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions= True ) return dict ( zip (urls, results)) 包组织 标准项目布局 myproject/ ├── src/ │ └── mypackage/ │ ├── init.py │ ├── main.py │ ├── api/ │ │ ├── init.py │ │ └── routes.py │ ├── models/ │ │ ├── init.py │ │ └── user.py │ └── utils/ │ ├── init.py │ └── helpers.py ├── tests/ │ ├── init.py │ ├── conftest.py │ ├── test_api.py │ └── test_models.py ├── pyproject.toml ├── README.md └── .gitignore 导入约定
import os import sys from pathlib import Path import requests from fastapi import FastAPI from mypackage.models import User from mypackage.utils import format_name
init .py 用于包导出
"""mypackage - A sample Python package.""" version = "1.0.0"
from mypackage.models import User, Post from mypackage.utils import format_name
all = [ "User" , "Post" , "format_name" ] 内存和性能 使用 slots 提高内存效率
class Point : def init ( self, x: float , y: float ): self .x = x self .y = y
class Point : slots = [ 'x' , 'y' ] def init ( self, x: float , y: float ): self .x = x self .y = y 生成器用于大数据
def read_lines ( path: str ) -> list [ str ]: with open (path) as f: return [line.strip() for line in f]
def read_lines ( path: str ) -> Iterator[ str ]: with open (path) as f: for line in f: yield line.strip() 避免在循环中进行字符串拼接
result = "" for item in items: result += str (item)
result = "" .join( str (item) for item in items)
from io import StringIO
buffer = StringIO() for item in items: buffer.write( str (item)) result = buffer.getvalue() Python 工具集成 基本命令
black . isort .
ruff check . pylint mypackage/
mypy .
pytest --cov=mypackage --cov-report=html
bandit -r .
"--cov=mypackage --cov-report=term-missing" 快速参考:Python 惯用法 惯用法 描述 EAFP 请求宽恕比请求许可更容易 上下文管理器 使用 with 进行资源管理 列表推导式 用于简单的转换 生成器 用于惰性求值和大数据集 类型提示 注解函数签名 数据类 用于具有自动生成方法的数据容器 slots 用于内存优化 f-strings 用于字符串格式化(Python 3.6+) pathlib.Path 用于路径操作(Python 3.4+) enumerate 用于循环中的索引-元素对 要避免的反模式
def append_to ( item, items=[] ): items.append(item) return items
def append_to ( item, items= None ): if items is None : items = [] items.append(item) return items
if type (obj) == list : process(obj)
if isinstance (obj, list ): process(obj)
if value == None : process()
if value is None : process()
from os.path import *
from os.path import join, exists
try : risky_operation() except : pass
try : risky_operation() except SpecificError as e: logger.error( f"Operation failed: {e} " ) 记住 :Python 代码应该具有可读性、显式性,并遵循最小意外原则。如有疑问,优先考虑清晰性而非巧妙性。