Python文件读写操作完整指南

装饰器工具集

import time
import functools
import logging

logging.basicConfig(level=logging.INFO)

def timer(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        logging.info(f"{func.__name__} 耗时: {end - start:.4f}秒")
        return result
    return wrapper

def log_execution(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logging.info(f"执行: {func.__name__}")
        result = func(*args, **kwargs)
        logging.info(f"完成: {func.__name__}")
        return result
    return wrapper

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

@timer
@log_execution
@retry(max_attempts=3)
def process_data():
    time.sleep(0.1)
    return "处理完成"

上下文管理器

import time
from contextlib import contextmanager

class DatabaseConnection:
    def __init__(self, db_name):
        self.db_name = db_name

    def __enter__(self):
        print(f"连接数据库: {self.db_name}")
        self.connection = f"已连接到 {self.db_name}"
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"关闭连接: {self.db_name}")
        self.connection = None
        return False

@contextmanager
def timer_context(name="代码块"):
    start = time.time()
    print(f"开始执行: {name}")
    try:
        yield
    finally:
        end = time.time()
        print(f"{name} 执行完成，耗时: {end - start:.4f}秒")

@contextmanager
def suppress(*exceptions):
    try:
        yield
    except exceptions:
        pass

HTTP客户端

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class HttpClient:
    def __init__(self, timeout=10):
        self.timeout = timeout
        self.session = requests.Session()
        self.session.proxies = {"http": None, "https": None}
        self.session.trust_env = False

        retry = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )

        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)

        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
        })

    def get(self, url, params=None):
        resp = self.session.get(url, params=params, timeout=self.timeout)
        resp.raise_for_status()
        return resp

    def post(self, url, data=None, json=None):
        resp = self.session.post(url, data=data, json=json, timeout=self.timeout)
        resp.raise_for_status()
        return resp

配置管理

import os
import json
from typing import Any

class Config:
    def __init__(self, config_file='config.json'):
        self.config_file = config_file
        self._data = self._load()

    def _load(self) -> dict:
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {}

    def get(self, key: str, default: Any = None) -> Any:
        keys = key.split('.')
        value = self._data
        for k in keys:
            if isinstance(value, dict) and k in value:
                value = value[k]
            else:
                return default
        return value

    def set(self, key: str, value: Any):
        keys = key.split('.')
        data = self._data
        for k in keys[:-1]:
            if k not in data:
                data[k] = {}
            data = data[k]
        data[keys[-1]] = value
        self._save()

    def _save(self):
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(self._data, f, indent=2, ensure_ascii=False)

日志工具

import logging
from logging.handlers import RotatingFileHandler

def setup_logger(name='app', log_file='app.log'):
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)

    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    console = logging.StreamHandler()
    console.setLevel(logging.INFO)
    console.setFormatter(formatter)

    file_handler = RotatingFileHandler(
        log_file, maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
    )
    file_handler.setLevel(logging.DEBUG)
    file_handler.setFormatter(formatter)

    logger.addHandler(console)
    logger.addHandler(file_handler)

    return logger

logger = setup_logger()

数据验证

from typing import List, Dict, Any
import re

class Validator:
    @staticmethod
    def email(value: str) -> bool:
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, value))

    @staticmethod
    def phone(value: str) -> bool:
        pattern = r'^1[3-9]\d{9}$'
        return bool(re.match(pattern, value))

    @staticmethod
    def required(value: Any) -> bool:
        if value is None:
            return False
        if isinstance(value, str) and not value.strip():
            return False
        return True

    @staticmethod
    def min_length(value: str, min_len: int) -> bool:
        return len(str(value)) >= min_len

最佳实践

遵循PEP 8规范，使用虚拟环境隔离依赖，添加类型提示提高可读性，使用Google风格文档字符串，捕获特定异常，大数据使用生成器，使用f-string进行字符串格式化。

技术分享，欢迎评论区交流讨论。