跳转至

GPConfigurable 类

GPConfigurable 是所有可配置对象的基类。通过继承此类,可以创建从 GPConfig 配置实例化的对象。

导入

from gpconfig import GPConfigurable

类定义

class GPConfigurable:
    def __init__(self, config: "GPConfig") -> None:
        """Initialize the configurable object from its config."""
        self._config = config

    @classmethod
    def from_config(
        cls,
        config: "GPConfig",
        *,
        context: GPConfigurableContext,
    ) -> "GPConfigurable":
        """Construct an object from config within a manager context.
        Override this to customize construction; the default calls cls(config)."""

    @property
    def config(self) -> "GPConfig":
        """Access the configuration object."""
        return self._config

from_config 是 GPConfigManager.get_object() 构造实例时使用的入口方法。关于如何重写它,请参阅上下文感知构造。

使用模式

基本用法

from typing import ClassVar
from gpconfig import GPConfig, GPConfigurable, GPConfigManager

# 1. 定义配置类
class DatabaseConfig(GPConfig):
    cfg_class_name: ClassVar[str] = "DatabaseConfig"
    host: str
    port: int = 5432
    username: str
    password: str
    database: str

# 2. 定义可配置对象类
class Database(GPConfigurable):
    """数据库连接对象"""

    def __init__(self, config: DatabaseConfig) -> None:
        super().__init__(config)
        self.host = config.host
        self.port = config.port
        self.username = config.username
        self.password = config.password
        self.database = config.database
        self._connection = None

    def connect(self):
        """建立数据库连接"""
        print(f"Connecting to {self.host}:{self.port}/{self.database}")
        # 实际连接逻辑...

    def close(self):
        """关闭连接"""
        if self._connection:
            self._connection.close()

注册并创建对象

# 3. 初始化管理器
manager = GPConfigManager("myapp")

# 4. 注册配置类和可配置类(分别注册)
GPConfigManager.register_config_class(DatabaseConfig)
GPConfigManager.register_configurable_class(Database)

# 5. 从配置创建对象实例
db = manager.get_object("database")

# 使用对象
db.connect()

register_configurable_class 按类的 __name__ 注册到注册表,因此 YAML 中设置的 configured_class_name 必须与 Python 类名完全一致(包括大小写)。

配置文件 (database.yaml):

cfg_class_name: "DatabaseConfig"
configured_class_name: "Database"
host: localhost
port: 5432
username: admin
password: secret
database: myapp

config 属性

通过 config 属性可以访问原始配置对象:

class Cache(GPConfigurable):
    def __init__(self, config: "CacheConfig") -> None:
        super().__init__(config)
        self.host = config.host
        self.port = config.port

    def reconnect(self):
        # 通过 config 属性访问配置
        print(f"Reconnecting to {self.config.host}:{self.config.port}")

cache = manager.get_object("cache")
print(cache.config.ttl)  # 访问配置中的字段

上下文感知构造

何时重写 from_config()

大多数子类不需要重写任何东西。默认的 GPConfigurable.from_config(config, *, context) 只是调用 cls(config),所以标准的 单参数子类无需任何改动即可继续工作:

class Database(GPConfigurable):
    def __init__(self, config: DatabaseConfig) -> None:
        super().__init__(config)   # 无需其它代码

仅当对象在构造时需要引用同一配置树中的其它配置时,才需要重写 from_config()。 context 参数是对象访问构造它的 manager 的唯一途径——按设计 GPConfig 不携带 manager 字段。

构造钩子与上下文

from gpconfig import GPConfigurable, GPConfigurableContext


class Worker(GPConfigurable):
    @classmethod
    def from_config(
        cls,
        config: WorkerConfig,
        *,
        context: GPConfigurableContext,
    ) -> "Worker":
        # context.manager:接收本次 get_object() 调用的 GPConfigManager。
        # context.path:   本对象 YAML 文件的规范点路径,
        #                  去掉了 .yaml 后缀和可选的项目名前缀
        #                  (例如 "services.api" 和 "myapp.services.api"
        #                  都得到 context.path == "services.api")。
        return cls(config)

GPConfigurableContext 是一个 frozen、slotted 的值对象,只有两个字段:

字段 含义
manager 接收本次 get_object() 请求的 GPConfigManager。
path 源 YAML 文件的规范点路径(不含 .yaml,不含项目名前缀)。

示例:引用另一份配置

一个需要数据库配置的 Worker。它通过 context.manager.get_config(...) 读取数据库 配置的数据,而不会构造 Database 对象:

from typing import ClassVar

from gpconfig import GPConfig, GPConfigurable, GPConfigManager, GPConfigurableContext


class DatabaseConfig(GPConfig):
    cfg_class_name: ClassVar[str] = "DatabaseConfig"
    host: str
    port: int = 5432


class WorkerConfig(GPConfig):
    cfg_class_name: ClassVar[str] = "WorkerConfig"
    worker_name: str
    concurrency: int = 4


class Worker(GPConfigurable):
    def __init__(self, config: WorkerConfig) -> None:
        super().__init__(config)
        self.worker_name = config.worker_name
        self.concurrency = config.concurrency
        self.database = None

    @classmethod
    def from_config(
        cls,
        config: WorkerConfig,
        *,
        context: GPConfigurableContext,
    ) -> "Worker":
        db_config = context.manager.get_config("database", DatabaseConfig)
        obj = cls(config)
        obj.database = db_config
        return obj

YAML 文件:

# database.yaml
cfg_class_name: "DatabaseConfig"
host: db.internal
port: 5432
# worker.yaml
cfg_class_name: "WorkerConfig"
configured_class_name: "Worker"
worker_name: ingest
concurrency: 8
GPConfigManager.register_config_class(DatabaseConfig)
GPConfigManager.register_config_class(WorkerConfig)
GPConfigManager.register_configurable_class(Worker)

manager = GPConfigManager("myapp")
worker = manager.get_object("worker")
print(worker.worker_name)        # ingest
print(worker.database.host)     # db.internal

在钩子中选择 get_config 还是 get_object

钩子里对 manager 的两种调用行为差别很大:

调用 会调用 from_config 吗 会递归吗 是否缓存
context.manager.get_config(path, Cfg) 否 否 是——配置数据,按文件缓存
context.manager.get_object(path) 是 是 否——每次调用都返回新对象

需要读取另一份配置的数据时用 get_config。只有需要另一个完整构造的对象时才用 get_object(并接受它会递归走它自己的 from_config)。

重要约束

  • 钩子必须返回已注册可配置类的实例(也允许返回其子类的实例)。其它返回值会触发 ConfigurableConstructionError(属性详见异常文档)。
  • 钩子抛出的异常——包括签名不兼容导致的 TypeError——会原样传播。钩子失败后 manager 绝不会回退重试旧式 cls(config) 构造器。
  • get_object() 不缓存对象;每次调用都返回新实例。get_config() 会按文件缓存配置数据。
  • 循环引用的检测由调用方负责。 manager 不做循环检测,也不限制深度。如果 get_object("a") 的钩子调 get_object("b"),而 get_object("b") 的钩子又调回 get_object("a"),就会无限递归直到栈溢出。

默认的 from_config() 保持了标准子类的契约,因此大多数用户无需重写它。

完整示例

多个可配置对象

from typing import ClassVar
from gpconfig import GPConfig, GPConfigurable, GPConfigManager

# 配置类
class LLMConfig(GPConfig):
    cfg_class_name: ClassVar[str] = "LLMConfig"
    api_key: str
    model: str
    temperature: float = 0.7
    max_tokens: int = 4096

# 可配置对象
class LLMProvider(GPConfigurable):
    """LLM 提供者"""

    def __init__(self, config: LLMConfig) -> None:
        super().__init__(config)
        self.api_key = config.api_key
        self.model = config.model
        self.temperature = config.temperature
        self.max_tokens = config.max_tokens

    def generate(self, prompt: str) -> str:
        """生成文本"""
        print(f"Using model: {self.model}")
        print(f"Temperature: {self.temperature}")
        # 实际调用 LLM API...
        return f"Response to: {prompt}"

# 初始化
manager = GPConfigManager("myapp")

# 分别注册配置类和可配置类
GPConfigManager.register_config_class(LLMConfig)
GPConfigManager.register_configurable_class(LLMProvider)

# 使用不同的配置创建不同的对象
openai = manager.get_object("llm.openai")
anthropic = manager.get_object("llm.anthropic")

print(openai.model)      # gpt-4
print(anthropic.model)   # claude-3-opus

YAML 配置文件:

# llm/openai.yaml
cfg_class_name: "LLMConfig"
configured_class_name: "LLMProvider"
api_key: sk-xxx
model: gpt-4
temperature: 0.7
max_tokens: 4096
# llm/anthropic.yaml
cfg_class_name: "LLMConfig"
configured_class_name: "LLMProvider"
api_key: sk-yyy
model: claude-3-opus
temperature: 0.8
max_tokens: 8192

访问配置元数据

class Service(GPConfigurable):
    def __init__(self, config: "ServiceConfig") -> None:
        super().__init__(config)
        self.name = config.name  # 配置名称
        self.url = config.url

    def info(self):
        return {
            "name": self.name,
            "config_file": str(self.config.cfg_file_path),
            "url": self.url
        }

service = manager.get_object("api_service")
print(service.info())
# {'name': 'api_service', 'config_file': '/path/to/api_service.yaml', 'url': '...'}

注意事项

每次调用创建新实例

get_object() 每次调用都会创建新的对象实例:

db1 = manager.get_object("database")
db2 = manager.get_object("database")

print(db1 is db2)  # False - 不同的实例

必须调用 super().init()

子类必须调用父类的 __init__ 方法:

class MyConfigurable(GPConfigurable):
    def __init__(self, config: MyConfig) -> None:
        super().__init__(config)  # 必须调用
        # 初始化逻辑...

类型提示

建议为 config 参数添加类型提示以获得更好的 IDE 支持:

class Database(GPConfigurable):
    def __init__(self, config: DatabaseConfig) -> None:  # 具体类型
        super().__init__(config)
        self.host = config.host  # IDE 可以自动补全