if backend == "torch":
    ...
elif backend == "tensorrt":
    ...
elif backend == "onnx":
    ...
elif backend == "custom":
    ...

class ModelConfig:
    def __init__(self, model_name: str, device: str) -> None:
        self.model_name = model_name
        self.device = device


config = ModelConfig("llama", "cuda")

field_name = "device"
value = getattr(config, field_name)

print(value)  # cuda

batch_size = getattr(config, "batch_size", 1)

setattr(config, "batch_size", 8)
setattr(ModelConfig, "version", "v1")

assert config.batch_size == 8
assert ModelConfig.version == "v1"

def collect_public_attributes(obj: object) -> dict[str, object]:
    result: dict[str, object] = {}

    for name in dir(obj):
        if name.startswith("_"):
            continue

        value = getattr(obj, name)

        if not callable(value):
            result[name] = value

    return result

import inspect


def load_model(
    model_name: str,
    device: str = "cuda",
    *,
    batch_size: int = 1,
) -> object:
    ...


signature = inspect.signature(load_model)

for name, parameter in signature.parameters.items():
    print(name, parameter.annotation, parameter.default)

def validate_loader(loader: object) -> None:
    if not callable(loader):
        raise TypeError("loader must be callable")

    signature = inspect.signature(loader)
    parameters = signature.parameters

    if "model_name" not in parameters:
        raise TypeError("loader must accept model_name")

class Worker:
    max_batch_size = 16

    def __init__(self) -> None:
        self.device = "cuda"
        self.running = True


worker = Worker()

print(worker.__dict__) # {'device': 'cuda', 'running': True}
print(Worker.__dict__) # {'__module__': '__main__', 'max_batch_size': 16, '__init__': <function Worker.__init__ at 0x...>, '__dict__': <attribute '__dict__' of 'Worker' objects>, '__weakref__': <attribute '__weakref__' of 'Worker' objects>, '__doc__': None}

class Model:
    pass


model = Model()

print(type(model))  # Model
print(type(Model))  # type

Model = type(
    "Model",
    (),
    {
        "name": "demo",
        "describe": lambda self: f"model={self.name}",
    },
)

model = Model()
print(model.describe())

from collections.abc import Callable
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def traced(
    func: Callable[P, R],
) -> Callable[P, R]:
    @wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        print(f"calling {func.__name__}")
        result = func(*args, **kwargs)
        print(f"completed {func.__name__}")
        return result

    return wrapper

import time
from collections.abc import Awaitable, Callable
from functools import wraps
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def measure_async(
    func: Callable[P, Awaitable[R]],
) -> Callable[P, Awaitable[R]]:
    @wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()

        try:
            return await func(*args, **kwargs)
        finally:
            elapsed = time.perf_counter() - start
            print(f"{func.__name__}: {elapsed:.4f}s")

    return wrapper

@retry
@cache
@trace
@limit
@fallback
async def infer(...):
    ...

class PositiveInteger:
    # __set_name__ 在类创建阶段由元类自动调用，
    # 将属性名注入描述符——这本身就是一种元编程钩子。
    def __set_name__(self, owner: type, name: str) -> None:
        self.name = name

    def __get__(
        self,
        instance: object | None,
        owner: type | None = None,
    ) -> int | "PositiveInteger":
        if instance is None:
            return self
        return instance.__dict__.get(self.name, 0)

    def __set__(self, instance: object, value: int) -> None:
        if not isinstance(value, int) or value <= 0:
            raise ValueError(f"{self.name} must be a positive integer")

        instance.__dict__[self.name] = value


class BatchConfig:
    batch_size = PositiveInteger()

    def __init__(self, batch_size: int) -> None:
        self.batch_size = batch_size

config = BatchConfig(16)
print(config.batch_size)

config.batch_size = -1
# ValueError

class Backend:
    registry: dict[str, type["Backend"]] = {}
    backend_name: str | None = None

    def __init_subclass__(
        cls,
        *,
        name: str | None = None,
        **kwargs,
    ) -> None:
        super().__init_subclass__(**kwargs)

        if name is not None:
            if name in cls.registry:
                raise ValueError(
                    f"backend already registered: {name}"
                )

            cls.registry[name] = cls
            cls.backend_name = name

class TorchBackend(Backend, name="torch"):
    pass


class TensorRTBackend(Backend, name="tensorrt"):
    pass

backend_cls = Backend.registry["torch"]

class RegistryMeta(type):
    # 所有使用此元类的类共享同一个 registry——
    # 这是元类的常见模式，因为元类实例本身就是类。
    registry: dict[str, type] = {}

    def __new__(
        mcls,
        name: str,
        bases: tuple[type, ...],
        namespace: dict,
    ):
        cls = super().__new__(mcls, name, bases, namespace)

        if name != "BaseBackend":
            backend_name = namespace.get("backend_name")
            if backend_name:
                mcls.registry[backend_name] = cls

        return cls

class BaseBackend(metaclass=RegistryMeta):
    backend_name: str | None = None


class TorchBackend(BaseBackend):
    backend_name = "torch"

print(RegistryMeta.registry["torch"])

import importlib

module = importlib.import_module("my_package.backends.torch")
backend_class = getattr(module, "TorchBackend")

def load_object(path: str) -> object:
    module_name, object_name = path.rsplit(":", 1)

    module = importlib.import_module(module_name)
    return getattr(module, object_name)

backend_cls = load_object(
    "my_package.backends.torch:TorchBackend"
)

def create_loader(format_name: str):
    if format_name == "onnx":
        return ONNXLoader()
    elif format_name == "tensorrt":
        return TensorRTLoader()
    else:
        raise ValueError(f"不支持的格式: {format_name}")

from collections.abc import Callable
from typing import Any


class BackendRegistry:
    def __init__(self) -> None:
        self._factories: dict[
            str,
            Callable[..., Any],
        ] = {}

    def register(
        self,
        name: str,
        factory: Callable[..., Any],
    ) -> None:
        if name in self._factories:
            raise ValueError(f"backend already registered: {name}")

        self._factories[name] = factory

    def create(self, name: str, **kwargs: Any) -> Any:
        try:
            factory = self._factories[name]
        except KeyError as exc:
            available = ", ".join(sorted(self._factories))
            raise ValueError(
                f"unknown backend {name!r}; "
                f"available: {available}"
            ) from exc

        return factory(**kwargs)

from typing import Protocol


class InferenceBackend(Protocol):
    async def infer(self, request: dict) -> dict:
        ...

class TorchBackend:
    def __init__(self, model_path: str) -> None:
        self.model_path = model_path

    async def infer(self, request: dict) -> dict:
        return {"backend": "torch", "result": request}


registry = BackendRegistry()
registry.register("torch", TorchBackend)

backend = registry.create(
    "torch",
    model_path="/models/demo",
)

registry = BackendRegistry()


def register_backend(name: str):
    def decorator(factory):
        registry.register(name, factory)
        return factory

    return decorator


@register_backend("torch")
class TorchBackend:
    ...

@LOADERS.register("onnx")
class ONNXLoader:
    ...

class ONNXLoader(ModelLoader, format_name="onnx"):
    ...

# my_project/plugins/__init__.py
def load_builtin_plugins() -> None:
    from my_project.plugins import onnx_loader      # noqa: F401
    from my_project.plugins import tensorrt_loader  # noqa: F401

import importlib
import logging

logger = logging.getLogger(__name__)

OPTIONAL_PLUGINS = (
    "my_project.plugins.onnx_loader",
    "my_project.plugins.tensorrt_loader",
)


def load_builtin_plugins() -> None:
    for module_name in OPTIONAL_PLUGINS:
        try:
            importlib.import_module(module_name)
        except ImportError as exc:
            # 没装 TensorRT 的 CPU 机器不该因此起不来
            logger.warning("plugin %s unavailable: %s", module_name, exc)

import importlib
import pkgutil
from types import ModuleType


def load_plugins(package: ModuleType) -> list[str]:
    loaded: list[str] = []

    for module_info in pkgutil.iter_modules(package.__path__):
        if module_info.name.startswith("_"):
            continue

        module_name = f"{package.__name__}.{module_info.name}"
        importlib.import_module(module_name)
        loaded.append(module_name)

    return loaded

from importlib.metadata import entry_points


def discover_backends() -> dict[str, object]:
    discovered: dict[str, object] = {}

    for entry_point in entry_points(
        group="my_ai.backends"
    ):
        discovered[entry_point.name] = entry_point.load()

    return discovered

class LazyPlugin:
    def __init__(self, entry_point) -> None:
        self.entry_point = entry_point
        self._loaded = None

    def load(self):
        if self._loaded is None:
            self._loaded = self.entry_point.load()
        return self._loaded

from typing import cast


def load_backend(obj: object) -> InferenceBackend:
    if not hasattr(obj, "infer"):
        raise TypeError("backend must provide infer()")

    return cast(InferenceBackend, obj)

import inspect


def validate_backend(backend_cls: type) -> None:
    infer = getattr(backend_cls, "infer", None)

    if infer is None or not callable(infer):
        raise TypeError("backend must define infer()")

    signature = inspect.signature(infer)

    if "request" not in signature.parameters:
        raise TypeError(
            "infer() must accept request parameter"
        )

class ModelSpec:
    inputs = ["input_ids", "attention_mask"]
    outputs = ["logits"]
    supports_streaming = True

def inspect_model_spec(spec_cls: type) -> dict[str, object]:
    return {
        "inputs": getattr(spec_cls, "inputs", []),
        "outputs": getattr(spec_cls, "outputs", []),
        "supports_streaming": getattr(
            spec_cls,
            "supports_streaming",
            False,
        ),
    }

from typing import Protocol


class SchedulerPlugin(Protocol):
    def submit(self, task: dict) -> str:
        ...

    def cancel(self, task_id: str) -> None:
        ...

    def status(self, task_id: str) -> str:
        ...

config = {
    "name": "priority",
    "options": {
        "max_pending": 1000,
        "default_priority": 10,
    },
}

from typing import Protocol


class LifecyclePlugin(Protocol):
    async def start(self) -> None:
        ...

    async def stop(self) -> None:
        ...

class BackendCapabilities:
    supports_streaming: bool
    supports_batching: bool
    supports_cancellation: bool

class Backend(Protocol):
    async def infer(self, request: dict) -> dict:
        ...

async def health_check(self) -> bool:
    ...

from dataclasses import dataclass


@dataclass
class PluginMetadata:
    api_version: str = "2.0"
    plugin_version: str = "1.4.0"

def check_compatibility(metadata: PluginMetadata) -> None:
    if metadata.api_version.split(".")[0] != "2":
        raise RuntimeError(
            f"unsupported plugin api: {metadata.api_version}"
        )

if plugin.capabilities.supports_streaming:
    await plugin.stream(request)
else:
    result = await plugin.infer(request)

# core/protocol.py
from typing import Protocol


class InferenceBackend(Protocol):
    name: str

    async def start(self) -> None:
        ...

    async def infer(self, request: dict) -> dict:
        ...

    async def stop(self) -> None:
        ...

# core/registry.py
from collections.abc import Callable
from typing import Any


class Registry:
    def __init__(self) -> None:
        self._items: dict[str, Callable[..., Any]] = {}

    def register(
        self,
        name: str,
        factory: Callable[..., Any],
    ) -> None:
        if name in self._items:
            raise ValueError(f"duplicate plugin: {name}")

        self._items[name] = factory

    def get(self, name: str) -> Callable[..., Any]:
        try:
            return self._items[name]
        except KeyError as exc:
            raise LookupError(
                f"plugin not found: {name}"
            ) from exc

    def names(self) -> tuple[str, ...]:
        return tuple(sorted(self._items))

# backends/torch_backend.py
class TorchBackend:
    name = "torch"

    def __init__(self, model_path: str) -> None:
        self.model_path = model_path

    async def start(self) -> None:
        # 加载模型和初始化资源
        pass

    async def infer(self, request: dict) -> dict:
        return {
            "backend": self.name,
            "request": request,
        }

    async def stop(self) -> None:
        # 释放模型和设备资源
        pass

async def start_backend(
    registry: Registry,
    name: str,
    config: dict,
) -> InferenceBackend:
    backend_factory = registry.get(name)
    backend = backend_factory(**config)

    await backend.start()
    return backend

def dispatch(path: str, request: dict) -> dict:
    if path == "/v1/chat/completions":
        return chat_completion(request)
    elif path == "/v1/embeddings":
        return embeddings(request)
    elif path == "/v1/models":
        return list_models(request)
    ...

from collections.abc import Callable
from typing import Any

Handler = Callable[..., Any]

ROUTES: dict[tuple[str, str], Handler] = {}


def route(path: str, method: str = "GET") -> Callable[[Handler], Handler]:
    def decorator(handler: Handler) -> Handler:
        key = (method.upper(), path)

        if key in ROUTES:
            existing = ROUTES[key]
            raise ValueError(
                f"duplicate route {method.upper()} {path}: "
                f"{existing.__module__}.{existing.__qualname__} "
                f"vs {handler.__module__}.{handler.__qualname__}"
            )

        ROUTES[key] = handler
        return handler

    return decorator

@route("/v1/models", method="GET")
def list_models() -> dict:
    return {"data": ["model-a", "model-b"]}


@route("/v1/chat/completions", method="POST")
def chat_completion(
    model: str,
    messages: list[dict],
    max_tokens: int = 128,
    temperature: float = 0.7,
    stream: bool = False,
) -> dict:
    ...

import inspect
from collections.abc import Callable
from typing import Any


def build_binder(handler: Callable[..., Any]):
    """启动阶段调用一次，返回一个只做字典查找的绑定函数。"""
    signature = inspect.signature(handler)

    required: list[str] = []
    optional: list[str] = []
    accepts_extra = False

    for name, parameter in signature.parameters.items():
        if parameter.kind is inspect.Parameter.VAR_KEYWORD:
            accepts_extra = True   # **kwargs：额外字段原样透传
            continue
        if parameter.kind is inspect.Parameter.VAR_POSITIONAL:
            continue               # *args：路由 handler 不支持，忽略
        if parameter.default is inspect.Parameter.empty:
            required.append(name)
        else:
            optional.append(name)

    known = frozenset(required) | frozenset(optional)

    def bind(request: dict) -> dict:
        missing = [name for name in required if name not in request]
        if missing:
            raise ValueError(f"缺少必要参数: {', '.join(missing)}")

        arguments = {name: request[name] for name in required}
        arguments.update(
            (name, request[name]) for name in optional if name in request
        )

        if accepts_extra:
            arguments.update(
                (k, v) for k, v in request.items() if k not in known
            )

        return arguments

    return bind

class CompiledRoute:
    __slots__ = ("handler", "bind")

    def __init__(self, handler: Callable[..., Any]) -> None:
        self.handler = handler
        self.bind = build_binder(handler)   # 签名分析只发生在这里

    def invoke(self, request: dict) -> Any:
        return self.handler(**self.bind(request))


# 启动阶段一次性编译
COMPILED_ROUTES: dict[tuple[str, str], CompiledRoute] = {
    key: CompiledRoute(handler) for key, handler in ROUTES.items()
}


def dispatch(method: str, path: str, request: dict) -> Any:
    route = COMPILED_ROUTES.get((method.upper(), path))

    if route is None:
        raise LookupError(f"未找到路由: {method} {path}")

    return route.invoke(request)   # 一次字典查找 + 一次字典构造

from pydantic import TypeAdapter

adapter = TypeAdapter(ChatRequest)   # 启动阶段构造，内部会编译校验器
payload = adapter.validate_python(request)

class APIHandler:
    def handle_models(self, request: dict) -> dict:
        ...

    def handle_chat(self, request: dict) -> dict:
        ...


def dispatch(action: str, request: dict):
    handler = APIHandler()
    method = getattr(handler, f"handle_{action}")   # 危险
    return method(request)

dispatch("chat", req)             # 正常
dispatch("models", req)           # 正常
dispatch("chat.__globals__", req) # 这次 getattr 会走到哪里？

HANDLERS: dict[str, Handler] = {
    "models": handle_models,
    "chat": handle_chat,
}


def dispatch(action: str, request: dict):
    try:
        handler = HANDLERS[action]
    except KeyError as exc:
        raise LookupError(
            f"unknown action {action!r}; "
            f"available: {', '.join(sorted(HANDLERS))}"
        ) from exc

    return handler(request)

def build_generate_handler(model, tokenizer):
    def handler(prompt: str, max_tokens: int = 128) -> dict:
        inputs = tokenizer(prompt)
        return model.generate(inputs, max_tokens=max_tokens)

    return handler


ROUTES[("POST", "/v1/generate")] = build_generate_handler(model, tokenizer)

VERSIONED_ROUTES = {
    ("v1", "chat"): chat_v1,
    ("v2", "chat"): chat_v2,
}

# 1. 每次请求都重新解析路径并导入
async def handle(request):
    backend = load_object(config["backend_path"])   # importlib + getattr
    return await backend().infer(request)

# 2. 每个 token 都做一次能力探测
async for token in stream:
    if hasattr(plugin, "on_token"):                # 每 token 一次 hasattr
        plugin.on_token(token)

# 3. 每次请求都重新校验签名
def dispatch(fn, payload):
    sig = inspect.signature(fn)                    # 微秒级，且可预先计算
    return fn(**{k: payload[k] for k in sig.parameters})

class Handler:
    def __init__(self, plugin: object) -> None:
        # 启动时一次性解析，反射结果被固化成普通引用
        self._backend = load_object(config["backend_path"])()
        self._on_token = getattr(plugin, "on_token", None)
        self._param_names = tuple(
            inspect.signature(self._backend.infer).parameters
        )

    async def handle(self, request: dict) -> dict:
        # 热路径上没有任何反射
        return await self._backend.infer(request)

    def emit(self, token: str) -> None:
        if self._on_token is not None:  # 只剩一次 None 判断
            self._on_token(token)

@traced
@retry(times=3)
@measure_async
async def infer(request: dict) -> dict:
    raise ValueError("bad input")

import inspect
from functools import wraps


def traced(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)

    return wrapper


@traced
def infer(request: dict, *, timeout: float = 1.0) -> dict:
    ...


print(infer.__name__)            # infer，而不是 wrapper
print(inspect.signature(infer))  # (request: dict, *, timeout: float = 1.0) -> dict
print(infer.__wrapped__)         # <function infer at 0x...>

try:
    module = importlib.import_module(module_name)
except ImportError as exc:
    raise PluginLoadError(
        f"cannot load plugin {name!r} from {module_name!r}"
    ) from exc

# 1. 注册表返回 Any，下游全部丢失类型信息
backend = registry.create("torch")   # -> Any
backend.inferr(request)              # 拼错方法名，pyright 不报错

# 2. setattr 注入的属性，检查器不认识
for name, value in config.items():
    setattr(self, name, value)
self.batch_size                      # error: 属性不存在

# 3. 元类生成的类属性，补全为空
class User(Model):                   # 元类注入了 objects / fields
    name = CharField()

User.objects.filter(...)             # 检查器不知道 objects 存在

from typing import Protocol


class InferenceBackend(Protocol):
    async def infer(self, request: dict) -> dict:
        ...


class Registry[T]:  # Python 3.12+ 泛型语法；3.11 及以前用 Generic[T] + TypeVar
    def __init__(self) -> None:
        self._items: dict[str, type[T]] = {}

    def register(self, name: str, factory: type[T]) -> None:
        self._items[name] = factory

    def create(self, name: str, **kwargs: object) -> T:
        return self._items[name](**kwargs)


backends: Registry[InferenceBackend] = Registry()

# config.pyi
class ServiceConfig:
    batch_size: int
    device: str
    max_tokens: int

validate_backend(obj)                    # 先在运行时确认结构
backend = cast(InferenceBackend, obj)    # 之后才向检查器做出承诺

getattr(obj, user_input)

eval(expression)
exec(source_code)

import enum
import importlib
import logging
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


class Criticality(enum.Enum):
    REQUIRED = "required"   # 加载失败 -> 拒绝启动
    OPTIONAL = "optional"   # 加载失败 -> 记录并降级


@dataclass
class PluginLoadResult:
    name: str
    criticality: Criticality
    loaded: bool
    error: str | None = None


@dataclass
class PluginLoader:
    results: list[PluginLoadResult] = field(default_factory=list)

    def load(
        self, name: str, module: str, criticality: Criticality
    ) -> None:
        try:
            importlib.import_module(module)
        except Exception as exc:
            if criticality is Criticality.REQUIRED:
                # 核心插件：立刻失败，并保留原始异常链
                raise RuntimeError(
                    f"required plugin {name!r} failed to load"
                ) from exc

            logger.warning("optional plugin %s unavailable: %r", name, exc)
            self.results.append(
                PluginLoadResult(name, criticality, False, repr(exc))
            )
            return

        self.results.append(PluginLoadResult(name, criticality, True))

    def degraded(self) -> list[str]:
        return [r.name for r in self.results if not r.loaded]

import inspect


def dump_registry(registry: dict[str, type]) -> list[dict[str, str]]:
    return [
        {
            "name": name,
            "implementation": f"{obj.__module__}.{obj.__qualname__}",
            "source": getattr(
                inspect.getmodule(obj), "__file__", "<unknown>"
            ),
        }
        for name, obj in sorted(registry.items())
    ]