Python工具类封装案例如何编写通用工具

wen python案例 2026-06-28 255

本文目录导读：

基础工具类设计
高级工具类模式
使用示例
最佳实践总结

我来分享一个Python工具类封装的完整案例,包含通用设计模式和最佳实践。

基础工具类设计

import os
import json
import yaml
import logging
import hashlib
from datetime import datetime
from typing import Any, Dict, List, Optional, Union
from pathlib import Path
class BaseTool:
    """工具基类，提供通用功能"""
    def __init__(self, config: Optional[Dict] = None):
        self.config = config or {}
        self.logger = self._setup_logger()
    def _setup_logger(self) -> logging.Logger:
        """配置日志器"""
        logger = logging.getLogger(self.__class__.__name__)
        logger.setLevel(logging.INFO)
        if not logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            )
            handler.setFormatter(formatter)
            logger.addHandler(handler)
        return logger
    def validate_config(self, required_keys: List[str]) -> bool:
        """验证配置完整性"""
        missing_keys = [k for k in required_keys if k not in self.config]
        if missing_keys:
            self.logger.warning(f"Missing config keys: {missing_keys}")
            return False
        return True
class FileUtils(BaseTool):
    """文件操作工具类"""
    def __init__(self, config: Optional[Dict] = None):
        super().__init__(config)
        self.supported_extensions = config.get('extensions', ['.txt', '.json', '.csv'])
    def read_file(self, file_path: str, encoding: str = 'utf-8') -> str:
        """读取文件内容"""
        try:
            with open(file_path, 'r', encoding=encoding) as f:
                return f.read()
        except FileNotFoundError:
            self.logger.error(f"File not found: {file_path}")
            raise
        except Exception as e:
            self.logger.error(f"Error reading file: {e}")
            raise
    def write_file(self, file_path: str, content: str, mode: str = 'w') -> bool:
        """写入文件"""
        try:
            os.makedirs(os.path.dirname(file_path) or '.', exist_ok=True)
            with open(file_path, mode, encoding='utf-8') as f:
                f.write(content)
            self.logger.info(f"File written successfully: {file_path}")
            return True
        except Exception as e:
            self.logger.error(f"Error writing file: {e}")
            return False
    def json_reader(self, file_path: str) -> Dict:
        """读取JSON文件"""
        content = self.read_file(file_path)
        return json.loads(content)
    def json_writer(self, file_path: str, data: Any, indent: int = 2) -> bool:
        """写入JSON文件"""
        content = json.dumps(data, indent=indent, ensure_ascii=False)
        return self.write_file(file_path, content)
    def get_file_info(self, file_path: str) -> Dict:
        """获取文件信息"""
        path = Path(file_path)
        if not path.exists():
            return {}
        return {
            'name': path.name,
            'extension': path.suffix,
            'size': path.stat().st_size,
            'modified_time': datetime.fromtimestamp(path.stat().st_mtime).isoformat(),
            'absolute_path': str(path.absolute())
        }
class DataTransformer(BaseTool):
    """数据转换工具类"""
    @staticmethod
    def to_json(data: Any) -> str:
        """转换为JSON字符串"""
        return json.dumps(data, ensure_ascii=False, default=str)
    @staticmethod
    def from_json(json_str: str) -> Any:
        """从JSON字符串转换"""
        return json.loads(json_str)
    @staticmethod
    def to_yaml(data: Any) -> str:
        """转换为YAML字符串"""
        return yaml.dump(data, default_flow_style=False, allow_unicode=True)
    @staticmethod
    def from_yaml(yaml_str: str) -> Any:
        """从YAML字符串转换"""
        return yaml.safe_load(yaml_str)
    def transform_list(self, data: List, operation: str, **kwargs) -> List:
        """列表转换操作"""
        operations = {
            'flatten': self._flatten_list,
            'unique': lambda x: list(set(x)),
            'sort': lambda x: sorted(x, **kwargs),
            'filter': lambda x: [i for i in x if i not in kwargs.get('exclude', [])]
        }
        if operation in operations:
            return operations[operation](data)
        else:
            self.logger.error(f"Unknown operation: {operation}")
            return data
    @staticmethod
    def _flatten_list(nested_list: List) -> List:
        """展平嵌套列表"""
        result = []
        for item in nested_list:
            if isinstance(item, list):
                result.extend(DataTransformer._flatten_list(item))
            else:
                result.append(item)
        return result
class SecurityUtils(BaseTool):
    """安全工具类"""
    @staticmethod
    def hash_password(password: str, salt: str = '') -> str:
        """密码哈希"""
        combined = password + salt
        return hashlib.sha256(combined.encode()).hexdigest()
    @staticmethod
    def md5_hash(text: str) -> str:
        """MD5哈希"""
        return hashlib.md5(text.encode()).hexdigest()
    @staticmethod
    def generate_token(length: int = 32) -> str:
        """生成随机令牌"""
        import secrets
        return secrets.token_hex(length // 2)
    def sanitize_filename(self, filename: str) -> str:
        """清理文件名，移除不安全字符"""
        import re
        # 移除非字母数字和.-_的字符
        sanitized = re.sub(r'[^\w\.\-]', '_', filename)
        # 限制长度
        max_length = self.config.get('max_filename_length', 255)
        return sanitized[:max_length]
class APIClient(BaseTool):
    """API客户端工具类"""
    def __init__(self, config: Optional[Dict] = None):
        super().__init__(config)
        self.base_url = config.get('base_url', '')
        self.timeout = config.get('timeout', 30)
        self.retry_count = config.get('retry_count', 3)
    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
        """GET请求"""
        import requests
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        for attempt in range(self.retry_count):
            try:
                response = requests.get(url, params=params, timeout=self.timeout)
                response.raise_for_status()
                return response.json()
            except requests.RequestException as e:
                self.logger.error(f"Request failed (attempt {attempt + 1}): {e}")
                if attempt == self.retry_count - 1:
                    raise
    def post(self, endpoint: str, data: Dict) -> Dict:
        """POST请求"""
        import requests
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            response = requests.post(url, json=data, timeout=self.timeout)
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            self.logger.error(f"POST request failed: {e}")
            raise

高级工具类模式

class CacheUtils(BaseTool):
    """缓存工具类，支持多级缓存"""
    def __init__(self, config: Optional[Dict] = None):
        super().__init__(config)
        self.cache = {}
        self.cache_type = config.get('cache_type', 'memory')
        self.max_size = config.get('max_cache_size', 100)
        if self.cache_type == 'file':
            self.cache_dir = Path(config.get('cache_dir', './cache'))
            self.cache_dir.mkdir(exist_ok=True)
    def get(self, key: str) -> Any:
        """获取缓存"""
        if self.cache_type == 'memory':
            return self.cache.get(key)
        else:
            return self._get_file_cache(key)
    def set(self, key: str, value: Any, ttl: int = 3600) -> None:
        """设置缓存"""
        if self.cache_type == 'memory':
            self._set_memory_cache(key, value)
        else:
            self._set_file_cache(key, value)
        self.logger.debug(f"Cache set for key: {key}")
    def _set_memory_cache(self, key: str, value: Any) -> None:
        """内存缓存管理"""
        if len(self.cache) >= self.max_size:
            # LRU淘汰策略
            self.cache.pop(next(iter(self.cache)))
        self.cache[key] = value
    def _get_file_cache(self, key: str) -> Optional[Any]:
        """文件缓存读取"""
        cache_file = self.cache_dir / f"{hashlib.md5(key.encode()).hexdigest()}.json"
        if cache_file.exists():
            with open(cache_file, 'r') as f:
                return json.load(f)
        return None
    def _set_file_cache(self, key: str, value: Any) -> None:
        """文件缓存写入"""
        cache_file = self.cache_dir / f"{hashlib.md5(key.encode()).hexdigest()}.json"
        with open(cache_file, 'w') as f:
            json.dump(value, f)
    def clear(self) -> None:
        """清空缓存"""
        self.cache.clear()
        self.logger.info("Cache cleared")
class ValidatorUtils(BaseTool):
    """数据验证工具类"""
    @staticmethod
    def validate_email(email: str) -> bool:
        """验证邮箱格式"""
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))
    @staticmethod
    def validate_phone(phone: str) -> bool:
        """验证手机号格式"""
        import re
        pattern = r'^1[3-9]\d{9}$'
        return bool(re.match(pattern, phone))
    @staticmethod
    def validate_url(url: str) -> bool:
        """验证URL格式"""
        from urllib.parse import urlparse
        try:
            result = urlparse(url)
            return all([result.scheme, result.netloc])
        except:
            return False
    def validate_json_schema(self, data: Dict, schema: Dict) -> bool:
        """JSON Schema验证"""
        try:
            import jsonschema
            jsonschema.validate(instance=data, schema=schema)
            return True
        except jsonschema.exceptions.ValidationError as e:
            self.logger.error(f"Validation error: {e}")
            return False

使用示例

def main():
    """使用示例"""
    # 1. 文件工具使用
    file_tool = FileUtils({
        'extensions': ['.txt', '.json', '.csv', '.yaml']
    })
    # 写入文件
    file_tool.write_file('config.json', '{"name": "test"}')
    # 读取JSON
    config = file_tool.json_reader('config.json')
    print(f"Config: {config}")
    # 2. 数据转换
    transformer = DataTransformer()
    nested_list = [1, [2, 3], [4, [5, 6]]]
    flat_list = transformer.transform_list(nested_list, 'flatten')
    print(f"Flattened: {flat_list}")
    # 3. 安全工具
    security = SecurityUtils()
    password_hash = security.hash_password("my_password", "unique_salt")
    token = security.generate_token()
    print(f"Token: {token}")
    # 4. 缓存使用
    cache = CacheUtils({
        'cache_type': 'memory',
        'max_cache_size': 50
    })
    cache.set('user_42', {'name': 'Alice', 'age': 30})
    user = cache.get('user_42')
    print(f"Cached user: {user}")
    # 5. 验证器使用
    validator = ValidatorUtils()
    email = "test@example.com"
    print(f"Email valid: {validator.validate_email(email)}")
if __name__ == "__main__":
    main()