跳转至

Kayaku🔗

kayaku.initialize 🔗

kayaku.initialize(specs: Dict[str, str], prettifier: Optional[Prettifier] = None, schema_generator_cls: Type[SchemaGenerator] = SchemaGenerator) -> None

初始化 Kayaku.

本函数需要最先被调用.

Parameters:

  • specs (Dict[str, str]) –

    domain 样式 -> 路径样式 的映射.

  • prettifier (Prettifier) –

    格式化器.

  • schema_generator_cls (Type[SchemaGenerator]) –

    JSON Schema 生成器的类.

Example 
from kayaku import bootstrap, config, initialize

initialize({"{**}.connection": "./config/connection.jsonc::{**}})

@config("my_mod.config.connection")
class Connection:
    account: int | None = None
    "Account"
    password: str | None = None
    "password"

bootstrap()

以上代码将会将 Connection 类的数据存储在 ./config/connection.jsonc 文件的 ["my_mod"]["config"] 里.

Source code in kayaku/domain.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def initialize(
    specs: Dict[str, str],
    prettifier: Optional[Prettifier] = None,
    schema_generator_cls: Type[SchemaGenerator] = SchemaGenerator,
) -> None:
    """初始化 Kayaku.

    本函数需要最先被调用.

    Args:
        specs (Dict[str, str]): `domain 样式 -> 路径样式` 的映射.
        prettifier (Prettifier, optional): 格式化器.
        schema_generator_cls (Type[SchemaGenerator], optional): JSON Schema 生成器的类.

    Example:

        ```py
        from kayaku import bootstrap, config, initialize

        initialize({"{**}.connection": "./config/connection.jsonc::{**}})

        @config("my_mod.config.connection")
        class Connection:
            account: int | None = None
            "Account"
            password: str | None = None
            "password"

        bootstrap()
        ```

        以上代码将会将 `Connection` 类的数据存储在 `./config/connection.jsonc` 文件的 `["my_mod"]["config"]` 里.
    """
    global _store
    _store = _GlobalStore(prettifier or Prettifier(), schema_generator_cls)
    exceptions: list[Exception] = []
    for src, path in specs.items():
        try:
            src_spec = parse_source(src)
            path_spec = parse_path(path)
            insert(src_spec, path_spec)
        except Exception as e:
            exceptions.append(e)
    if exceptions:
        raise ValueError(
            f"{len(exceptions)} occurred during spec initialization.", exceptions
        )

kayaku.config module-attribute 🔗

kayaku.config = config_stub if TYPE_CHECKING else config_impl

Note

Please note that config is routed to real config implementation at runtime.

将类型转换为 kayaku 承认的设置类。

domain 以外的参数将被直接传入 dataclasses.dataclass 进行转换。

Parameters:

  • domain (str) –

    唯一的 用于标识类的结构位置与类别的字符串，用 . 分开。

  • init (bool) –

    If true (the default), a __init__() method will be generated.

  • repr (bool) –

    If true (the default), a __repr__() method will be generated.

  • eq (bool) –

    If true (the default), an __eq__() method will be generated.

  • order (bool) –

    If specified, __lt__(), __le__(), __gt__(), and __ge__() methods will be generated.

  • unsafe_hash (bool) –

    Force dataclass() to create a __hash__() method. Otherwise a __hash__() method is generated according to how eq and frozen are set.

  • frozen (bool) –

    If specified, assigning to fields will generate an exception. This emulates read-only frozen instances.

  • match_args (bool) –

    If true (the default is True), the __match_args__ tuple will be created from the list of parameters to the generated __init__().

  • kw_only (bool) –

    If specified, then all fields will be marked as keyword-only.

  • slots (bool) –

    If specified, __slots__ attribute will be generated and new class will be returned instead of the original one.

Returns:

Source code in kayaku/model.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def config_impl(domain: str, **kwargs) -> Callable[[type], Type[ConfigModel]]:
    def wrapper(cls: type) -> type[ConfigModel]:
        from . import domain as domain_mod

        if not hasattr(domain_mod, "_store"):
            raise RuntimeError("You cannot call `config` before initialization!")
        from .domain import _store as g_store
        from .domain import insert_domain

        cls = cast(Type[ConfigModel], dataclass(**kwargs)(cls))
        domain_tup: Tuple[str, ...] = tuple(domain.split("."))
        if not all(domain_tup):
            raise ValueError(f"{domain!r} contains empty segment!")

        if domain_tup in g_store.models:
            m_store = g_store.models[domain_tup]
            other = m_store.cls
            if (
                cls.__module__ == other.__module__
                and cls.__qualname__ == other.__qualname__
            ):
                if m_store.instance is not None:
                    instance = m_store.instance
                    if (
                        sys.getrefcount(instance) > 3
                    ):  # parameter, local var `instance`, `m_store.instance`
                        logger.warning(f"Instance of {other!r} is stilled referred!")
                    m_store.instance = None
                f_store = g_store.files[m_store.location.path]
                # remove original occupations

                g_store.cls_domains.pop(other)
                other_def_name = f_store.generator.retrieve_name(other)
                other_schema = f_store.generator.defs.pop(other_def_name)
                mount_dest = tuple(m_store.location.mount_dest)
                for field in get_fields(other):
                    sub_dest = mount_dest + (field.name,)
                    f_store.field_mount_record.remove(sub_dest)

                # inject new ones
                insert_domain(domain_tup, cls)
                cls_def_name = f_store.generator.retrieve_name(cls)
                cls_schema = f_store.generator.defs[cls_def_name]
                if cls_schema != other_schema:
                    logger.warning(f"Schema of {cls} has changed!")
                return cls

            raise NameError(f"{domain!r} is already occupied by {other!r}")
        insert_domain(domain_tup, cls)
        return cls

    return wrapper

kayaku.default module-attribute 🔗

kayaku.default = copying_field

自动 deepcopy 默认值的 dataclass.field.

Parameters:

  • default (T) –

    The default value for this field, which will be deepcopied on each call.

  • init (bool) –

    If true (the default), this field is included in __init__().

  • repr (bool) –

    If true (the default), this field is included in repr().

  • hash (bool | None) –

    If true, this field is included in the generated __hash__() method. If None (the default), use the value of compare.

  • compare (bool) –

    If true (the default), this field is included in comparison methods.

  • metadata (Mapping[Any, Any] | None) –

    A mapping which is stored but not otherwise examined by dataclass.

  • kw_only (bool) –

    If true, this field will be marked as keyword-only. (3.10+)

Returns:

  • T( T ) –

    The field object, faked to be the same type of default in typ checkers' eyes.

kayaku.bootstrap 🔗

kayaku.bootstrap() -> None

预处理所有 ConfigModel 并写入默认值和 JSON Schema.

建议在加载完外部模块后调用.

Source code in kayaku/domain.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def bootstrap() -> None:
    """预处理所有 `ConfigModel` 并写入默认值和 JSON Schema.

    建议在加载完外部模块后调用.
    """

    exceptions = []

    for path, store in _store.files.items():
        document = loads(path.read_text("utf-8") or "{}")
        path.with_suffix(".schema.json").write_text(
            dumps(store.get_schema()), encoding="utf-8"
        )
        for mount_dest, domains in store.mount.items():
            container = document
            for sect in mount_dest:
                container = container.setdefault(sect, JObject())
            for domain in domains:
                model_store = _store.models[domain]
                if model_store.instance is None:
                    try:
                        model_store.instance = from_dict(model_store.cls, container)
                    except Exception as e:
                        exceptions.append((path, mount_dest, model_store.cls, e))
                format_with_model(container, model_store.cls)
        document.pop("$schema", None)
        document["$schema"] = path.with_suffix(".schema.json").as_uri()
        path.write_text(
            dumps(_store.prettifier.prettify(document), endline=True), "utf-8"
        )
    if exceptions:
        raise ValueError(exceptions)

kayaku.save_all 🔗

kayaku.save_all() -> None

保存所有容器，并格式化。

可以在退出前调用本函数。

Source code in kayaku/domain.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def save_all() -> None:
    """保存所有容器，并格式化。

    可以在退出前调用本函数。
    """

    for path, store in _store.files.items():
        document = loads(path.read_text("utf-8") or "{}")
        path.with_suffix(".schema.json").write_text(
            dumps(store.get_schema()), encoding="utf-8"
        )
        for mount_dest, domains in store.mount.items():
            container = document
            for sect in mount_dest:
                container = container.setdefault(sect, JObject())
            for domain in domains:
                model_store = _store.models[domain]
                if model_store.instance is not None:
                    update(container, model_store.instance)
                format_with_model(container, model_store.cls)
        document.pop("$schema", None)
        document["$schema"] = path.with_suffix(".schema.json").as_uri()
        path.write_text(
            dumps(_store.prettifier.prettify(document), endline=True), "utf-8"
        )

kayaku.create 🔗

kayaku.create(cls: Type[T], flush: bool = False) -> T

创建对应的模型。

给 flush 传入真值会强制重载。

Note

为了安全重载，我们推荐你让返回的实例存活时间尽量短。

Source code in kayaku/model.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def create(cls: Type[T], flush: bool = False) -> T:
    """创建对应的模型。

    给 `flush` 传入真值会强制重载。

    Note:
        为了安全重载，我们推荐你让返回的实例存活时间尽量短。
    """

    from .domain import _store

    if not (issubclass(cls, ConfigModel) and cls in _store.cls_domains):
        raise TypeError(f"{cls!r} is not a registered ConfigModel class!")
    domain = _store.cls_domains[cls]
    model_store = _store.models[domain]
    if flush or model_store.instance is None:

        fmt_path = model_store.location
        document = loads(fmt_path.path.read_text("utf-8") or "{}")
        container = document
        for sect in fmt_path.mount_dest:
            container = container.setdefault(sect, JObject())
        model_store.instance = from_dict(cls, container)

    return cast(T, model_store.instance)

kayaku.save 🔗

kayaku.save(model: Union[T, Type[T]]) -> None

保存某个类的设置。 对应的 JSON Schema 也会被更新。

Source code in kayaku/model.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def save(model: Union[T, Type[T]]) -> None:
    """保存某个类的设置。 对应的 JSON Schema 也会被更新。"""
    from .domain import _store

    cls = cast(Type[ConfigModel], model if isinstance(model, type) else model.__class__)
    m_store = _store.models[_store.cls_domains[cls]]
    inst = m_store.instance
    if inst is not None:
        document = loads(m_store.location.path.read_text("utf-8") or "{}")
        container = document
        for sect in m_store.location.mount_dest:
            container = container.setdefault(sect, JObject())
        update(container, inst)
        document.pop("$schema", None)
        document["$schema"] = m_store.location.path.with_suffix(".schema.json").as_uri()
        m_store.location.path.write_text(
            dumps(_store.prettifier.prettify(document), endline=True), "utf-8"
        )
    m_store.location.path.with_suffix(".schema.json").write_text(
        dumps(_store.files[m_store.location.path].get_schema()), "utf-8"
    )

kayaku.pretty.Prettifier 🔗

kayaku.pretty.Prettifier(indent: int = 4, trail_comma: bool = False, key_quote: Quote | None | Constant[False] = Quote.DOUBLE, string_quote: Quote | None = Quote.DOUBLE, unfold_single: bool = False)

容器格式化工具

Parameters:

  • indent (int) –

    缩进数量. Defaults to 4.

  • trail_comma (bool) –

    是否要为容器增加尾随逗号. Defaults to False.

  • key_quote (Quote | None | Constant[False]) –

    键使用的引号风格, None 为保留, Quote.DOUBLE 为双引号, Quote.SINGLE 为单引号, False 为尽可能去除引号. Defaults to Quote.DOUBLE.

  • string_quote (Quote | None) –

    值中字符串使用的引号风格, 解释同上. Defaults to Quote.DOUBLE.

  • unfold_single (bool) –

    单个元素的容器是否展开. Defaults to False.

Source code in kayaku/pretty.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    indent: int = 4,
    trail_comma: bool = False,
    key_quote: Quote | None | Constant[False] = Quote.DOUBLE,
    string_quote: Quote | None = Quote.DOUBLE,
    unfold_single: bool = False,
):
    """

    Args:
        indent (int, optional): 缩进数量. Defaults to 4.
        trail_comma (bool, optional): 是否要为容器增加尾随逗号. Defaults to False.
        key_quote (Quote | None | Constant[False], optional): 键使用的引号风格,
            None 为保留, Quote.DOUBLE 为双引号, Quote.SINGLE 为单引号, False 为尽可能去除引号.
            Defaults to Quote.DOUBLE.
        string_quote (Quote | None, optional): 值中字符串使用的引号风格, 解释同上. Defaults to Quote.DOUBLE.
        unfold_single (bool, optional): 单个元素的容器是否展开. Defaults to False.
    """
    self.indent: int = indent
    self.trail_comma: bool = trail_comma
    self.key_quote: Quote | Constant[False] | None = key_quote
    self.string_quote: Quote | None = string_quote
    self.unfold_single: bool = unfold_single
    self.layer: int = 0