Python函数参数与返回值详解

上下文管理器

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

工厂模式

class Animal:
    def speak(self):
        pass

class Dog(Animal):
    def speak(self):
        return "汪汪汪"

class Cat(Animal):
    def speak(self):
        return "喵喵喵"

class AnimalFactory:
    @staticmethod
    def create_animal(animal_type):
        if animal_type == "dog":
            return Dog()
        elif animal_type == "cat":
            return Cat()
        else:
            raise ValueError(f"未知的动物类型: {animal_type}")

dog = AnimalFactory.create_animal("dog")
print(dog.speak())

迭代器模式

class MyIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index < len(self.data):
            value = self.data[self.index]
            self.index += 1
            return value
        raise StopIteration

class MyCollection:
    def __init__(self):
        self.items = []

    def add(self, item):
        self.items.append(item)

    def __iter__(self):
        return MyIterator(self.items)

装饰器模式

class Coffee:
    def cost(self):
        return 10

    def description(self):
        return "基础咖啡"

class CoffeeDecorator:
    def __init__(self, coffee):
        self.coffee = coffee

    def cost(self):
        return self.coffee.cost()

    def description(self):
        return self.coffee.description()

class MilkDecorator(CoffeeDecorator):
    def cost(self):
        return super().cost() + 3

    def description(self):
        return super().description() + " + 牛奶"

class SugarDecorator(CoffeeDecorator):
    def cost(self):
        return super().cost() + 1

    def description(self):
        return super().description() + " + 糖"

观察者模式

class Subject:
    def __init__(self):
        self._observers = []

    def attach(self, observer):
        self._observers.append(observer)

    def detach(self, observer):
        self._observers.remove(observer)

    def notify(self, data=None):
        for observer in self._observers:
            observer.update(data)

class Observer:
    def update(self, data):
        pass

class EmailNotifier(Observer):
    def update(self, data):
        print(f"发送邮件通知: {data}")

class SMSNotifier(Observer):
    def update(self, data):
        print(f"发送短信通知: {data}")

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 time
import functools
import logging

logging.basicConfig(level=logging.INFO)

def log(msg):
    print(msg)
    with open(LOG_FILE, "a", encoding="utf-8") as f:
        f.write(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {msg}
")

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 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)

单例模式

class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

class SingletonMeta(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    def __init__(self):
        self.connection = None

日志工具

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()

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