「学习笔记」Python 进阶(二)：抽象设计 — 构建优雅系统

class A:
    def method(self):
        return "A"

class B(A):
    def method(self):
        return "B"

class C(A):
    def method(self):
        return "C"

class D(B, C):     # 继承顺序决定优先级
    pass

d = D()
print(d.method())  # "B" — 因为 B 在 C 之前
print(D.__mro__)   # (D, B, C, A, object) — 完整的查找链
print(D.mro())     # 同上，列表形式

class A:
    def method(self):
        print("A")

class B(A):
    def method(self):
        super().method()       # 等价于 super(B, self).method()
        print("B")

class C(A):
    def method(self):
        super().method()       # 等价于 super(C, self).method()
        print("C")

class D(B, C):
    def method(self):
        super().method()
        print("D")

D().method() # 输出: （A C B D）
# super() 按 MRO 顺序调用下一个类的同名方法，不是只调用直接父类！
# MRO 链的顺序是：D → B → C → A → object，所以A是最先打印的

class NormalClass:
    def __init__(self):
        self.name = None
        self.value = None

class OptimizedClass:
    __slots__ = ('name', 'value')   # 只允许这两个属性
    def __init__(self):
        self.name = None
        self.value = None

# __slots__ 的效果：
# 1. 禁止动态添加新属性 → obj.new_attr = 1 会报错 AttributeError
# 2. 节省内存（不用 __dict__ 字典存储属性）
# 3. 属性访问略快

class ValidatedString:
    """自定义描述符：对属性赋值进行校验"""
    def __init__(self, min_length=1):
        self.min_length = min_length
        self.data = {}

    def __get__(self, obj, objtype=None):
        return self.data.get(id(obj), "")

    def __set__(self, obj, value):
        if len(value) < self.min_length:
            raise ValueError(f"String too short (min {self.min_length})")
        self.data[id(obj)] = value

class User:
    name = ValidatedString(min_length=2)   # 类级别描述符

user = User()
user.name = "Tom"         # 触发 __set__，长度 >= 2，通过
# user.name = "X"          # ValueError!

from abc import ABC, abstractmethod

class LLMProvider(ABC):
    @abstractmethod
    def chat(self, messages: list[dict]) -> str:
        """子类必须实现聊天接口"""
        ...

    @abstractmethod
    def embed(self, texts: list[str]) -> list[list[float]]:
        """子类必须实现向量化接口"""
        ...

    # 普通方法，子类可直接继承
    def validate_config(self):
        return True

# 抽象类不能实例化
# provider = LLMProvider()  # TypeError!

# 必须实现所有抽象方法
class OpenAIProvider(LLMProvider):
    def chat(self, messages):
        return "Response from OpenAI"

    def embed(self, texts):
        return [[0.1, 0.2] for _ in texts]

from abc import ABC, abstractmethod

class Plugin(ABC):
    @property
    @abstractmethod
    def name(self) -> str:
        """子类必须提供此属性"""
        ...

    @classmethod
    @abstractmethod
    def from_config(cls, config: dict) -> "Plugin":
        """工厂方法"""
        ...

class MyPlugin(Plugin):
    @property
    def name(self) -> str:
        return "MyPlugin"

    @classmethod
    def from_config(cls, config: dict):
        return cls()

from abc import ABC

class Base(ABC):
    """所有 Embedding 模型的基类"""
    def __init__(self, key, model_name, **kwargs):
        pass

    def encode(self, texts: list) -> list:
        """编码文本列表为向量"""
        raise NotImplementedError("Please implement encode method!")

    def encode_queries(self, text: str) -> list:
        """编码查询文本为向量"""
        raise NotImplementedError("Please implement encode method!")

class OpenAIEmbed(Base):
    def encode(self, texts: list) -> list:
        resp = self.client.embeddings.create(
            input=texts, model=self.model_name
        )
        return [d.embedding for d in resp.data]

    def encode_queries(self, text: str) -> list:
        return self.encode([text])[0]

# 方式一：ABC（推荐用于接口定义）
class Interface(ABC):
    @abstractmethod
    def method(self): ...

# 方式二：NotImplementedError（轻量级，不阻止实例化）
class BaseClass:
    def method(self):
        raise NotImplementedError("Subclasses must implement")

# 区别：
# ABC + @abstractmethod → 实例化时就会报错（更早暴露）
# NotImplementedError → 只有调用时才会报错（更晚暴露）

from typing import Protocol, Any

class ToolCallSession(Protocol):
    """任何有 tool_call 方法的对象都匹配此 Protocol"""
    def tool_call(self, name: str, arguments: dict[str, Any]) -> str:
        ...

# 不继承任何 Protocol，但结构匹配
class MySession:
    def tool_call(self, name: str, arguments: dict[str, Any]) -> str:
        return f"Called {name} with {arguments}"

# 类型检查器认为 MySession 匹配 ToolCallSession
def use_session(session: ToolCallSession) -> str:
    return session.tool_call("search", {"query": "test"})

result = use_session(MySession())     # 类型检查通过！

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> None: ...

class Circle:
    def draw(self) -> None:
        print("Drawing circle")

print(isinstance(Circle(), Drawable))  # True（需要 @runtime_checkable）

from dataclasses import dataclass, field

@dataclass
class User:
    name: str                          # 必填字段
    age: int = 0                       # 带默认值
    tags: list = field(default_factory=list)  # 可变默认值必须用 field

user = User(name="Tom", age=25)
print(user)         # User(name='Tom', age=25, tags=[])
print(user == User(name="Tom", age=25, tags=[]))  # True

from dataclasses import dataclass, field

@dataclass
class Team:
    name: str
    members: list = field(default_factory=list)  # 避免多个实例共享同一个列表
    _id: int = field(repr=False, compare=False)  # 隐藏内部ID，不参与打印和相等比较

@dataclass(frozen=True)
class ExternalAccess:
    external_user_emails: set[str]
    is_public: bool = False

access = ExternalAccess({"a@b.com"}, True)
# access.is_public = False   # FrozenInstanceError!

@dataclass
class Person:
    name: str
    age: int
    
    def __post_init__(self):
        if self.age < 0:
            raise ValueError("年龄不能为负数")

from dataclasses import asdict, astuple

@dataclass
class GraphChange:
    removed_nodes: set[str] = field(default_factory=set)
    added_nodes: set[str] = field(default_factory=set)

change = GraphChange(removed_nodes={"A", "B"})
print(asdict(change))   # {'removed_nodes': {'A', 'B'}, 'added_nodes': set()}
print(astuple(change))  # ({'A', 'B'}, set())

from enum import Enum, IntEnum, auto

class Color(Enum):
    RED = "#FF0000"
    GREEN = "#00FF00"
    BLUE = "#0000FF"

# 访问
print(Color.RED)           # Color.RED
print(Color.RED.value)     # '#FF0000'
print(Color.RED.name)      # 'RED'
print(Color("#FF0000"))    # Color.RED（通过值查找）

# 遍历
for color in Color:
    print(color.name, color.value)

from enum import Enum, auto

class Status(Enum):
    PENDING = auto()
    RUNNING = auto()
    DONE = auto()

from enum import Enum, unique

@unique
class Shape(Enum):
    SQUARE = 1
    CIRCLE = 2
    # TRIANGLE = 1  # 这会报错，因为值 1 已被 SQUARE 使用

class VIP(Enum):
    BRONZE = 1
    GOLD = 3

    def level(self):
        return {1: '初级', 3: '高级'}.get(self.value, '未知')

print(VIP.GOLD.level())  # 输出: '高级'

from enum import IntEnum

class RetCode(IntEnum):
    SUCCESS = 0
    NOT_EFFECTIVE = 10
    EXCEPTION_ERROR = 100
    ARGUMENT_ERROR = 101
    UNAUTHORIZED = 401
    FORBIDDEN = 403
    SERVER_ERROR = 500

print(RetCode.SUCCESS == 0)           # True
print(RetCode.UNAUTHORIZED > 100)      # True（IntEnum 支持大小比较）

from enum import Flag

class Permission(Flag):
    READ = 1
    WRITE = 2
    EXECUTE = 4

# 组合权限
user_perms = Permission.READ | Permission.WRITE
print(Permission.READ in user_perms) # True

from collections import Counter

c = Counter('abracadabra')
print(c.most_common(3))  # 输出: [('a', 5), ('b', 2), ('r', 2)]

# 找重复元素
ids = ["id_1", "id_2", "id_1", "id_3", "id_2", "id_1"]
dups = [item for item, count in Counter(ids).items() if count > 1]
# ['id_1', 'id_2']

# 数学运算
c1, c2 = Counter(a=3, b=1), Counter(a=1, b=2)
print(c1 + c2)  # Counter({'a': 4, 'b': 3})

from collections import defaultdict

# 统计字符出现次数
d = defaultdict(int)
for char in 'hello':
    d[char] += 1
print(d)  # 输出: defaultdict(<class 'int'>, {'h': 1, 'e': 1, 'l': 2, 'o': 1})

# 按首字母分组（不需要先检查键是否存在）
words = ["apple", "banana", "avocado", "blueberry", "cherry"]
groups = defaultdict(list)
for word in words:
    groups[word[0]].append(word)      # 键不存在时自动创建空列表
print(dict(groups)) # 输出:  {'a': ['apple', 'avocado'], 'b': ['banana', 'blueberry'], 'c': ['cherry']}

# 实际使用：聚合实体，按名称分组
maybe_nodes = defaultdict(list)
for chunk in chunks:
    for entity in extract_entities(chunk):
        maybe_nodes[entity.name].append(entity)

from collections import deque

d = deque([1, 2, 3], maxlen=3)
d.appendleft(0)  # 队头添加
d.pop()          # 队尾弹出

from collections import OrderedDict

# Python 3.7+ 普通 dict 也保持插入顺序
# 但 OrderedDict 提供额外方法：move_to_end / popitem（按位置）

# 实现 LRU 缓存
class LRUCache:
    def __init__(self, capacity=100):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key):
        if key not in self.cache:
            return None
        self.cache.move_to_end(key)        # 移到末尾（标记为最近使用）
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)  # 弹出最旧的（头部）

from typing import NamedTuple

class ExtractionResult(NamedTuple):
    text_content: str
    embedded_images: list
    metadata: dict

result = ExtractionResult("hello", [], {"page": 1})
print(result.text_content)  # hello
print(result._fields)       # ('text_content', 'embedded_images', 'metadata')

class AppError(Exception):
    """项目基础异常"""

class DocumentParseError(AppError):
    """文档解析失败"""
    def __init__(self, doc_name, reason):
        self.doc_name = doc_name
        self.reason = reason
        super().__init__(f"Parse failed for '{doc_name}': {reason}")

class ModelCallError(AppError):
    """模型调用失败"""
    def __init__(self, model_name, status_code, detail):
        self.model_name = model_name
        self.status_code = status_code
        self.detail = detail
        super().__init__(f"Model '{model_name}' returned {status_code}: {detail}")

def parse_document(filepath):
    try:
        content = read_file(filepath)
        return process(content)
    except FileNotFoundError as e:
        raise DocumentParseError(filepath, "file not found") from e
    # from e 保留原始异常链，traceback 会显示完整调用栈

from contextlib import contextmanager

@contextmanager
def safe_db_transaction(db):
    """自动提交/回滚的事务管理器"""
    try:
        yield
        db.commit()
    except Exception:
        db.rollback()
        raise              # 异常继续向上传播

import importlib
from importlib.util import module_from_spec, spec_from_file_location

# 按模块名导入
module = importlib.import_module("myapp.llm.chat_model")
ChatClass = getattr(module, "OpenAIChat")

# 从文件路径动态导入（API 路由自动发现的常见模式）
def load_module_from_file(file_path, module_name):
    spec = spec_from_file_location(module_name, file_path)
    module = module_from_spec(spec)
    sys.modules[module_name] = module
    spec.loader.exec_module(module)
    return module

import inspect

class MyClass:
    """文档字符串"""
    def method_a(self, x: int, y: str = "hello") -> bool:
        """方法A"""
        return True

# 获取所有成员
for name, obj in inspect.getmembers(MyClass, predicate=inspect.isfunction):
    print(f"  {name}: {obj}")

# 获取函数签名
sig = inspect.signature(MyClass.method_a)
# (self, x: int, y: str = 'hello') -> bool

for param_name, param in sig.parameters.items():
    print(f"  {param_name}: type={param.annotation}, default={param.default}")

import importlib, inspect

ChatModel = {}   # {"OpenAI": OpenAIChat, "Ollama": OllamaChat, ...}

# 步骤1：导入模块
module = importlib.import_module("myapp.llm.chat_model")

# 步骤2：在模块中找到 Base 基类
base_class = None
for name, obj in inspect.getmembers(module):
    if inspect.isclass(obj) and name == "Base":
        base_class = obj

# 步骤3：自动发现所有 Base 子类并注册
for name, obj in inspect.getmembers(module):
    if (inspect.isclass(obj)                          # 是类
        and issubclass(obj, base_class)               # 是 Base 的子类
        and obj is not base_class                     # 不是 Base 自身
        and hasattr(obj, "_FACTORY_NAME")):           # 有注册名
        ChatModel[obj._FACTORY_NAME] = obj

# 使用：通过名称获取模型类
model_class = ChatModel["OpenAI"]
model = model_class(api_key="sk-xxx", model_name="gpt-4")

import pkgutil, importlib
from pathlib import Path

# 递归遍历包中所有子模块
pkg_dir = Path(__file__).resolve().parent
for modinfo in pkgutil.walk_packages([str(pkg_dir)], prefix=__name__ + "."):
    module = importlib.import_module(modinfo.name)
    # 处理发现的模块...