Skip to content

nskit.common

Common utilities for nskit.

nskit.common.configuration

Base configuration class.

Includes
  • properties in model dumps
  • file based config loading (json/toml/yaml)
  • model dumping to toml & yaml

BaseConfiguration

Bases: PropertyDumpMixin, BaseSettings

A Pydantic BaseSettings type object with Properties included in model dump, and yaml and toml integrations.

Source code in src/nskit/common/configuration/__init__.py 
 38
 39
 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
 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
107
108
109
110
111
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
class BaseConfiguration(PropertyDumpMixin, _BaseSettings):
    """A Pydantic BaseSettings type object with Properties included in model dump, and yaml and toml integrations."""

    model_config = SettingsConfigDict(env_file_encoding="utf-8")

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[_BaseSettings],
        init_settings: _PydanticBaseSettingsSource,
        env_settings: _PydanticBaseSettingsSource,
        dotenv_settings: _PydanticBaseSettingsSource,
        file_secret_settings: _PydanticBaseSettingsSource,
    ) -> tuple[_PydanticBaseSettingsSource, ...]:
        """Create settings loading, including the FileConfigSettingsSource."""
        config_files = cls.model_config.get("config_file")
        config_file_encoding = cls.model_config.get("config_file_encoding")
        file_types = {"json": [".json", ".jsn"], "yaml": [".yaml", ".yml"], "toml": [".toml", ".tml"]}
        if config_files:
            if isinstance(config_files, (Path, str)):
                config_files = [config_files]
        else:
            config_files = []

        split_config_files = {}
        for file_type, suffixes in file_types.items():
            original = cls.model_config.get(f"{file_type}_file")
            if original and isinstance(original, (Path, str)):
                split_config_files[file_type] = [original]
            elif original:
                split_config_files[file_type] = original
            else:
                split_config_files[file_type] = []
            for config_file in config_files:
                if Path(config_file).suffix.lower() in suffixes:
                    split_config_files[file_type].append(config_file)
        return (
            init_settings,
            env_settings,
            JsonConfigSettingsSource(
                settings_cls,
                split_config_files["json"],
                cls.model_config.get("json_file_encoding") or config_file_encoding,
            ),
            YamlConfigSettingsSource(
                settings_cls,
                split_config_files["yaml"],
                cls.model_config.get("yaml_file_encoding") or config_file_encoding,
            ),
            TomlConfigSettingsSource(settings_cls, split_config_files["toml"]),
            DotEnvSettingsSource(
                settings_cls,
                dotenv_settings.env_file,
                dotenv_settings.env_file_encoding,
                dotenv_settings.case_sensitive,
                dotenv_settings.env_prefix,
                dotenv_settings.env_nested_delimiter,
                dotenv_settings.env_ignore_empty,
                dotenv_settings.env_parse_none_str,
                cls.model_config.get("dotenv_extra", "ignore"),
            ),
            file_secret_settings,
        )

    def model_dump_toml(
        self,
        *,
        indent: int | None = None,
        include: Any = None,
        exclude: Any = None,
        by_alias: bool = False,
        exclude_unset: bool = False,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
        round_trip: bool = False,
        warnings: bool = True,
    ):
        """Dump model to TOML."""
        # We go via JSON to include indent etc.
        return toml.dumps(
            json.loads(
                self.model_dump_json(
                    indent=indent,
                    include=include,
                    exclude=exclude,
                    by_alias=by_alias,
                    exclude_unset=exclude_unset,
                    exclude_defaults=exclude_defaults,
                    exclude_none=exclude_none,
                    round_trip=round_trip,
                    warnings=warnings,
                )
            )
        )

    def model_dump_yaml(
        self,
        *,
        indent: int | None = None,
        include: Any = None,
        exclude: Any = None,
        by_alias: bool = False,
        exclude_unset: bool = False,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
        round_trip: bool = False,
        warnings: bool = True,
    ):
        """Dump model to YAML."""
        # We go via JSON to include indent etc.
        return yaml.dumps(
            json.loads(
                self.model_dump_json(
                    indent=indent,
                    include=include,
                    exclude=exclude,
                    by_alias=by_alias,
                    exclude_unset=exclude_unset,
                    exclude_defaults=exclude_defaults,
                    exclude_none=exclude_none,
                    round_trip=round_trip,
                    warnings=warnings,
                )
            )
        )

model_dump_toml(*, indent=None, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True)

Dump model to TOML.

Source code in src/nskit/common/configuration/__init__.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def model_dump_toml(
    self,
    *,
    indent: int | None = None,
    include: Any = None,
    exclude: Any = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
):
    """Dump model to TOML."""
    # We go via JSON to include indent etc.
    return toml.dumps(
        json.loads(
            self.model_dump_json(
                indent=indent,
                include=include,
                exclude=exclude,
                by_alias=by_alias,
                exclude_unset=exclude_unset,
                exclude_defaults=exclude_defaults,
                exclude_none=exclude_none,
                round_trip=round_trip,
                warnings=warnings,
            )
        )
    )

model_dump_yaml(*, indent=None, include=None, exclude=None, by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, warnings=True)

Dump model to YAML.

Source code in src/nskit/common/configuration/__init__.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def model_dump_yaml(
    self,
    *,
    indent: int | None = None,
    include: Any = None,
    exclude: Any = None,
    by_alias: bool = False,
    exclude_unset: bool = False,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
    round_trip: bool = False,
    warnings: bool = True,
):
    """Dump model to YAML."""
    # We go via JSON to include indent etc.
    return yaml.dumps(
        json.loads(
            self.model_dump_json(
                indent=indent,
                include=include,
                exclude=exclude,
                by_alias=by_alias,
                exclude_unset=exclude_unset,
                exclude_defaults=exclude_defaults,
                exclude_none=exclude_none,
                round_trip=round_trip,
                warnings=warnings,
            )
        )
    )

settings_customise_sources(settings_cls, init_settings, env_settings, dotenv_settings, file_secret_settings) classmethod

Create settings loading, including the FileConfigSettingsSource.

Source code in src/nskit/common/configuration/__init__.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 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
@classmethod
def settings_customise_sources(
    cls,
    settings_cls: type[_BaseSettings],
    init_settings: _PydanticBaseSettingsSource,
    env_settings: _PydanticBaseSettingsSource,
    dotenv_settings: _PydanticBaseSettingsSource,
    file_secret_settings: _PydanticBaseSettingsSource,
) -> tuple[_PydanticBaseSettingsSource, ...]:
    """Create settings loading, including the FileConfigSettingsSource."""
    config_files = cls.model_config.get("config_file")
    config_file_encoding = cls.model_config.get("config_file_encoding")
    file_types = {"json": [".json", ".jsn"], "yaml": [".yaml", ".yml"], "toml": [".toml", ".tml"]}
    if config_files:
        if isinstance(config_files, (Path, str)):
            config_files = [config_files]
    else:
        config_files = []

    split_config_files = {}
    for file_type, suffixes in file_types.items():
        original = cls.model_config.get(f"{file_type}_file")
        if original and isinstance(original, (Path, str)):
            split_config_files[file_type] = [original]
        elif original:
            split_config_files[file_type] = original
        else:
            split_config_files[file_type] = []
        for config_file in config_files:
            if Path(config_file).suffix.lower() in suffixes:
                split_config_files[file_type].append(config_file)
    return (
        init_settings,
        env_settings,
        JsonConfigSettingsSource(
            settings_cls,
            split_config_files["json"],
            cls.model_config.get("json_file_encoding") or config_file_encoding,
        ),
        YamlConfigSettingsSource(
            settings_cls,
            split_config_files["yaml"],
            cls.model_config.get("yaml_file_encoding") or config_file_encoding,
        ),
        TomlConfigSettingsSource(settings_cls, split_config_files["toml"]),
        DotEnvSettingsSource(
            settings_cls,
            dotenv_settings.env_file,
            dotenv_settings.env_file_encoding,
            dotenv_settings.case_sensitive,
            dotenv_settings.env_prefix,
            dotenv_settings.env_nested_delimiter,
            dotenv_settings.env_ignore_empty,
            dotenv_settings.env_parse_none_str,
            cls.model_config.get("dotenv_extra", "ignore"),
        ),
        file_secret_settings,
    )

SettingsConfigDict

Bases: SettingsConfigDict

Customised Settings Config Dict.

Source code in src/nskit/common/configuration/__init__.py 
30
31
32
33
34
35
class SettingsConfigDict(_SettingsConfigDict):
    """Customised Settings Config Dict."""

    dotenv_extra: ExtraValues | None = "ignore"
    config_file: PathType | None = None
    config_file_encoding: str | None = None

nskit.common.contextmanagers

Reusable context managers.

ChDir

Bases: ContextDecorator

Context manager for running in a specified (or temporary) directory.

The optional argument is a path to a specified target directory, if this isn't provided, a temporary directory is created

Source code in src/nskit/common/contextmanagers/chdir.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class ChDir(ContextDecorator):
    """Context manager for running in a specified (or temporary) directory.

    The optional argument is a path to a specified target directory, if this isn't provided, a temporary directory is created
    """

    def __init__(self, target_dir: Optional[Path] = None):
        """Initialise the context manager.

        Keyword Args:
            target_dir (Optional[Path]): the target directory
        """
        self._temp_dir = None
        if not target_dir:
            # Handling circular imports with LoggingConfig
            logger_factory.get_logger(__name__).debug("No target_dir provided, using a temporary directory")
            self._temp_dir = tempfile.TemporaryDirectory()
            target_dir = self._temp_dir.name
        self.cwd = Path.cwd()
        self.target_dir = Path(target_dir)

    def __enter__(self):
        """Change to the target directory."""
        # Handling circular imports with LoggingConfig
        logger_factory.get_logger(__name__).info(f"Changing to {self.target_dir}")
        if not self.target_dir.exists():
            self.target_dir.mkdir()
        os.chdir(str(self.target_dir))
        if self._temp_dir:
            return self._temp_dir.__enter__()

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Reset to the original directory."""
        os.chdir(str(self.cwd))
        if self._temp_dir:
            try:
                self._temp_dir.__exit__(exc_type, exc_val, exc_tb)
            except PermissionError as e:
                # Handling circular imports with LoggingConfig
                logger_factory.get_logger(__name__).warning("Unable to delete temporary directory.")
                warnings.warn(e, stacklevel=2)

__enter__()

Change to the target directory.

Source code in src/nskit/common/contextmanagers/chdir.py 
34
35
36
37
38
39
40
41
42
def __enter__(self):
    """Change to the target directory."""
    # Handling circular imports with LoggingConfig
    logger_factory.get_logger(__name__).info(f"Changing to {self.target_dir}")
    if not self.target_dir.exists():
        self.target_dir.mkdir()
    os.chdir(str(self.target_dir))
    if self._temp_dir:
        return self._temp_dir.__enter__()

__exit__(exc_type, exc_val, exc_tb)

Reset to the original directory.

Source code in src/nskit/common/contextmanagers/chdir.py 
44
45
46
47
48
49
50
51
52
53
def __exit__(self, exc_type, exc_val, exc_tb):
    """Reset to the original directory."""
    os.chdir(str(self.cwd))
    if self._temp_dir:
        try:
            self._temp_dir.__exit__(exc_type, exc_val, exc_tb)
        except PermissionError as e:
            # Handling circular imports with LoggingConfig
            logger_factory.get_logger(__name__).warning("Unable to delete temporary directory.")
            warnings.warn(e, stacklevel=2)

__init__(target_dir=None)

Initialise the context manager.

Other Parameters:

Name Type Description
target_dir Optional[Path]

the target directory
Source code in src/nskit/common/contextmanagers/chdir.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, target_dir: Optional[Path] = None):
    """Initialise the context manager.

    Keyword Args:
        target_dir (Optional[Path]): the target directory
    """
    self._temp_dir = None
    if not target_dir:
        # Handling circular imports with LoggingConfig
        logger_factory.get_logger(__name__).debug("No target_dir provided, using a temporary directory")
        self._temp_dir = tempfile.TemporaryDirectory()
        target_dir = self._temp_dir.name
    self.cwd = Path.cwd()
    self.target_dir = Path(target_dir)

Env

Bases: ContextDecorator

Context manager for managing environment variables.

The optional arguments can provide either an exhaustive set of environment values, values to override or values to remove.

Source code in src/nskit/common/contextmanagers/env.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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
65
class Env(ContextDecorator):
    """Context manager for managing environment variables.

    The optional arguments can provide either an exhaustive set of environment values, values to override or values to remove.
    """

    def __init__(
        self,
        environ: Optional[dict[str, str]] = None,
        override: Optional[dict[str, str]] = None,
        remove: Optional[list[str]] = None,
    ):
        """Initialise the context manager.

        The parameters are applied in the following order (so can be combined): 1st - environ, 2nd - override, 3rd - remove

        Keyword Args:
            environ (Optional[Dict[str, str]]): an exhaustive set of environment values to set (replaces overall os.environ contents)
            override (Optional[Dict[str, str]]): a set of environment values to either override or set (replaces values in existing os.environ)
            remove (Optional[List[str]]): a set of environment values to remove (removes values if found in os.environ )
        """
        if environ is not None and not isinstance(environ, dict):
            raise TypeError("environ should be a dict")
        if override is not None and not isinstance(override, dict):
            raise TypeError("override should be a dict")
        if remove is not None and not isinstance(remove, (list, tuple, set)):
            raise TypeError("remove should be a (list, tuple, set)")
        self._environ = environ
        self._override = override
        self._remove = remove
        self._original = None

    def __enter__(self):
        """Change to the target environment variables."""
        # Handling circular imports with LoggingConfig
        logger = logger_factory.get_logger(__name__)
        if self._environ or self._override or self._remove:
            self._original = os.environ.copy()
            if self._environ is not None:
                # Here we pop all keys from it and then update
                os.environ.clear()
                os.environ.update(self._environ)
            if self._override:
                os.environ.update(self._override)
            if self._remove:
                for key in list(self._remove):
                    os.environ.pop(key, None)
            logger.info("Changing env variables")
        else:
            logger.info("No arguments set (environ, override, remove)")

    def __exit__(self, *args, **kwargs):  # noqa: U100
        """Reset to the original environment variables."""
        if self._original:
            os.environ.clear()
            os.environ.update(self._original.copy())

__enter__()

Change to the target environment variables.

Source code in src/nskit/common/contextmanagers/env.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __enter__(self):
    """Change to the target environment variables."""
    # Handling circular imports with LoggingConfig
    logger = logger_factory.get_logger(__name__)
    if self._environ or self._override or self._remove:
        self._original = os.environ.copy()
        if self._environ is not None:
            # Here we pop all keys from it and then update
            os.environ.clear()
            os.environ.update(self._environ)
        if self._override:
            os.environ.update(self._override)
        if self._remove:
            for key in list(self._remove):
                os.environ.pop(key, None)
        logger.info("Changing env variables")
    else:
        logger.info("No arguments set (environ, override, remove)")

__exit__(*args, **kwargs)

Reset to the original environment variables.

Source code in src/nskit/common/contextmanagers/env.py 
61
62
63
64
65
def __exit__(self, *args, **kwargs):  # noqa: U100
    """Reset to the original environment variables."""
    if self._original:
        os.environ.clear()
        os.environ.update(self._original.copy())

__init__(environ=None, override=None, remove=None)

Initialise the context manager.

The parameters are applied in the following order (so can be combined): 1st - environ, 2nd - override, 3rd - remove

Other Parameters:

Name Type Description
environ Optional[Dict[str, str]]

an exhaustive set of environment values to set (replaces overall os.environ contents)
override Optional[Dict[str, str]]

a set of environment values to either override or set (replaces values in existing os.environ)
remove Optional[List[str]]

a set of environment values to remove (removes values if found in os.environ )
Source code in src/nskit/common/contextmanagers/env.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def __init__(
    self,
    environ: Optional[dict[str, str]] = None,
    override: Optional[dict[str, str]] = None,
    remove: Optional[list[str]] = None,
):
    """Initialise the context manager.

    The parameters are applied in the following order (so can be combined): 1st - environ, 2nd - override, 3rd - remove

    Keyword Args:
        environ (Optional[Dict[str, str]]): an exhaustive set of environment values to set (replaces overall os.environ contents)
        override (Optional[Dict[str, str]]): a set of environment values to either override or set (replaces values in existing os.environ)
        remove (Optional[List[str]]): a set of environment values to remove (removes values if found in os.environ )
    """
    if environ is not None and not isinstance(environ, dict):
        raise TypeError("environ should be a dict")
    if override is not None and not isinstance(override, dict):
        raise TypeError("override should be a dict")
    if remove is not None and not isinstance(remove, (list, tuple, set)):
        raise TypeError("remove should be a (list, tuple, set)")
    self._environ = environ
    self._override = override
    self._remove = remove
    self._original = None

TestExtension

Bases: ContextDecorator

Context manager for running a test of an entrypoint.

Source code in src/nskit/common/contextmanagers/test_extensions.py 
 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
class TestExtension(ContextDecorator):
    """Context manager for running a test of an entrypoint."""

    def __init__(self, name: str, group: str, entrypoint: type, *, solo: bool = False):
        """Initialise the context manager.

        Keyword Args:
            name (str): the extension name
            group (str): the extension group
            entrypoint (type): the object/type to load in the entrypoint
            solo (bool): set so only that entrypoint will be found (can cause side-effects)
        """
        self.ep = _TestEntrypoint(name=name, group=group, entrypoint=entrypoint, solo=solo)
        self._clean = False

    def __enter__(self):
        """Add the extension so it can be loaded."""
        logger_factory.get_logger(__name__).info(f"Starting entrypoint for extension {self.ep.name} in {self.ep.group}")
        self.ep.start()

    def __exit__(self, *args, **kwargs):  # noqa: U100
        """Remove the extension and return to the original."""
        logger_factory.get_logger(__name__).info(
            f"Stoppings entrypoint for extension {self.ep.name} in {self.ep.group}"
        )
        self.ep.stop()

__enter__()

Add the extension so it can be loaded.

Source code in src/nskit/common/contextmanagers/test_extensions.py 
94
95
96
97
def __enter__(self):
    """Add the extension so it can be loaded."""
    logger_factory.get_logger(__name__).info(f"Starting entrypoint for extension {self.ep.name} in {self.ep.group}")
    self.ep.start()

__exit__(*args, **kwargs)

Remove the extension and return to the original.

Source code in src/nskit/common/contextmanagers/test_extensions.py 
 99
100
101
102
103
104
def __exit__(self, *args, **kwargs):  # noqa: U100
    """Remove the extension and return to the original."""
    logger_factory.get_logger(__name__).info(
        f"Stoppings entrypoint for extension {self.ep.name} in {self.ep.group}"
    )
    self.ep.stop()

__init__(name, group, entrypoint, *, solo=False)

Initialise the context manager.

Other Parameters:

Name Type Description
name str

the extension name
group str

the extension group
entrypoint type

the object/type to load in the entrypoint
solo bool

set so only that entrypoint will be found (can cause side-effects)
Source code in src/nskit/common/contextmanagers/test_extensions.py 
82
83
84
85
86
87
88
89
90
91
92
def __init__(self, name: str, group: str, entrypoint: type, *, solo: bool = False):
    """Initialise the context manager.

    Keyword Args:
        name (str): the extension name
        group (str): the extension group
        entrypoint (type): the object/type to load in the entrypoint
        solo (bool): set so only that entrypoint will be found (can cause side-effects)
    """
    self.ep = _TestEntrypoint(name=name, group=group, entrypoint=entrypoint, solo=solo)
    self._clean = False

nskit.common.extensions

Common extension helpers.

ExtensionsEnum

Bases: Enum

Enum created from available extensions on an entrypoint.

Source code in src/nskit/common/extensions.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class ExtensionsEnum(Enum):
    """Enum created from available extensions on an entrypoint."""

    @classmethod
    def from_entrypoint(cls, name: str, entrypoint: str):
        """Create the enum with name, from entrypoint options."""
        options = {u: u for u in get_extension_names(entrypoint)}
        kls = cls(name, options)
        kls.__entrypoint__ = entrypoint
        return kls

    @property
    def extension(self):
        """Load the extension."""
        return load_extension(self.__entrypoint__, self.value)

    @classmethod
    def _patch(cls):
        """Used for testing and patching objects."""
        options = {u: u for u in get_extension_names(cls.__entrypoint__)}
        # Loop over options not in members
        for key in options:
            if key not in cls._member_names_:
                extend_enum(cls, key, key)

extension property

Load the extension.

from_entrypoint(name, entrypoint) classmethod

Create the enum with name, from entrypoint options.

Source code in src/nskit/common/extensions.py 
44
45
46
47
48
49
50
@classmethod
def from_entrypoint(cls, name: str, entrypoint: str):
    """Create the enum with name, from entrypoint options."""
    options = {u: u for u in get_extension_names(entrypoint)}
    kls = cls(name, options)
    kls.__entrypoint__ = entrypoint
    return kls

get_extension_names(entrypoint)

Get all installed extension names for a given entrypoint.

Source code in src/nskit/common/extensions.py 
16
17
18
19
20
21
22
def get_extension_names(entrypoint: str):
    """Get all installed extension names for a given entrypoint."""
    extensions = []
    for ep in entry_points().select(group=entrypoint):
        extensions.append(ep.name)
    logger_factory.get_logger(__name__).debug(f"Identified extensions {extensions} for entrypoint {entrypoint}")
    return extensions

get_extensions(entrypoint)

Load all extensions for a given entrypoint.

Source code in src/nskit/common/extensions.py 
32
33
34
35
36
37
38
def get_extensions(entrypoint: str):
    """Load all extensions for a given entrypoint."""
    extensions = {}
    for ep in entry_points().select(group=entrypoint):
        extensions[ep.name] = ep
    logger_factory.get_logger(__name__).debug(f"Identified extensions {extensions} for entrypoint {entrypoint}")
    return extensions

load_extension(entrypoint, extension)

Load a given extension for a given entrypoint.

Source code in src/nskit/common/extensions.py 
25
26
27
28
29
def load_extension(entrypoint: str, extension: str):
    """Load a given extension for a given entrypoint."""
    for ep in entry_points().select(group=entrypoint, name=extension):
        return ep.load()
    logger_factory.get_logger(__name__).warning(f"Entrypoint {extension} not found for {entrypoint}")

nskit.common.io

File IO handlers.

nskit.common.io.json

Provide a JSON Load/Dump API consistent with stdlib JSON.

dump(data, f, /, default=None, option=None, **kwargs)

Dump JSON to file.

Source code in src/nskit/common/io/json.py 
25
26
27
def dump(data: Any, f: TextIO, /, default: Optional[Any] = None, option: Optional[int] = None, **kwargs):
    """Dump JSON to file."""
    f.write(dumps(data, default=default, option=option, **kwargs))
dumps(data, /, default=None, option=None, **kwargs)

Dump JSON to string.

Source code in src/nskit/common/io/json.py 
15
16
17
def dumps(data: Any, /, default: Optional[Any] = None, option: Optional[int] = None, **kwargs):
    """Dump JSON to string."""
    return orjson.dumps(data, default=default, option=option, **kwargs).decode()
load(fp, **kwargs)

Load JSON from file.

Source code in src/nskit/common/io/json.py 
20
21
22
def load(fp: TextIO, **kwargs):
    """Load JSON from file."""
    return loads(fp.read(), **kwargs)
loads(s, **kwargs)

Load JSON from string.

Source code in src/nskit/common/io/json.py 
10
11
12
def loads(s: str, **kwargs):
    """Load JSON from string."""
    return orjson.loads(s, **kwargs)

nskit.common.io.toml

Provide a TOML Load/Dump API consistent with JSON.

dump(data, fp, sort_keys=False, **kwargs)

Load TOML to file/stream.

Source code in src/nskit/common/io/toml.py 
26
27
28
def dump(data: Mapping, fp: TextIO, sort_keys: bool = False, **kwargs):
    """Load TOML to file/stream."""
    return tomlkit.dump(data, fp, sort_keys=sort_keys, **kwargs)
dumps(data, sort_keys=False, **kwargs)

Dump TOML to string.

Source code in src/nskit/common/io/toml.py 
16
17
18
def dumps(data: Mapping, sort_keys: bool = False, **kwargs):
    """Dump TOML to string."""
    return tomlkit.dumps(data, sort_keys=sort_keys, **kwargs)
load(fp, **kwargs)

Load TOML from file/stream.

Source code in src/nskit/common/io/toml.py 
21
22
23
def load(fp: TextIO, **kwargs):
    """Load TOML from file/stream."""
    return tomlkit.load(fp, **kwargs)
loads(s, **kwargs)

Load TOML from string.

Source code in src/nskit/common/io/toml.py 
11
12
13
def loads(s: str, **kwargs):
    """Load TOML from string."""
    return tomlkit.loads(s, **kwargs)

nskit.common.io.yaml

Provide a YAML 1.2 Load/Dump API consistent with JSON.

dump(data, stream, *, typ='rt', **kwargs)

Dump YAML to file/stream.

Source code in src/nskit/common/io/yaml.py 
30
31
32
def dump(data: Any, stream: TextIO, *, typ: str = "rt", **kwargs):
    """Dump YAML to file/stream."""
    return _YAML(typ=typ).dump(data, stream=stream, **kwargs)
dumps(data, *, typ='rt', **kwargs)

Dump YAML to string.

Source code in src/nskit/common/io/yaml.py 
18
19
20
21
22
def dumps(data: Any, *, typ: str = "rt", **kwargs):
    """Dump YAML to string."""
    s = StringIO()
    dump(data, s, typ=typ, **kwargs)
    return s.getvalue()
load(stream, *, typ='rt', **kwargs)

Load YAML from file/stream.

Source code in src/nskit/common/io/yaml.py 
25
26
27
def load(stream: TextIO, *, typ: str = "rt", **kwargs):
    """Load YAML from file/stream."""
    return _YAML(typ=typ).load(stream, **kwargs)
loads(s, *, typ='rt', **kwargs)

Load YAML from string.

Source code in src/nskit/common/io/yaml.py 
13
14
15
def loads(s: str, *, typ: str = "rt", **kwargs):
    """Load YAML from string."""
    return load(StringIO(s), typ=typ, **kwargs)

nskit.common.logging

Common logging helpers.

LoggingConfig

Bases: BaseConfiguration

This is a basic config for logging.

Source code in src/nskit/common/logging/config.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class LoggingConfig(BaseConfiguration):
    """This is a basic config for logging."""

    # TODO setup as settings
    level: str = Field(
        DEFAULT_LOGLEVEL,
        validation_alias=AliasChoices("level", LOGLEVEL_ENV_VAR),
        description="Set the log level for the logger",
    )
    logfile: Optional[Path] = Field(
        None, validation_alias=AliasChoices("logfile", LOGFILE_ENV_VAR), description="Set the log file for the logger"
    )
    format_string: str = Field(
        BASE_FORMAT_STR,
        validation_alias=AliasChoices("format_string", LOGFORMATSTRING_ENV_VAR),
        description="Set the log format for the logger",
    )
    json_format: bool = Field(
        True,
        validation_alias=AliasChoices("json_format", JSON_ENV_VAR),
        serialization_alias="json",
        description="Output JSON Logs",
    )
    extra: dict[str, Any] = Field(default_factory=dict, description="Extra kwargs")

    @property
    def formatter(self):
        """Return the logging formatter for the format string."""
        return LoggingFormatter(self.format_string)

formatter property

Return the logging formatter for the format string.

LibraryLoggerFactory

A factory for creating multiple library loggers.

Source code in src/nskit/common/logging/library.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class LibraryLoggerFactory:
    """A factory for creating multiple library loggers."""

    def __init__(self, library: str, version: str, base_config: Optional[Union[LoggingConfig, dict[str, Any]]] = None):
        """Initialise the logger factory."""
        self.__library = library
        self.__version = version
        self.__base_config = base_config

    def get_logger(self, name, config=None, **kwargs):
        """Get the library logger."""
        if config is None:
            config = self.__base_config
        return get_library_logger(self.library, self.version, name, config, **kwargs)

    def get(self, name, config=None, **kwargs):
        """Alias for the get_logger method."""
        return self.get_logger(name, config, **kwargs)

    def getLogger(self, name, config=None, **kwargs):
        """Alias for the get_logger method to provide parity with the standard logging API."""
        return self.get_logger(name, config, **kwargs)

    @property
    def library(self):
        """Return the library name."""
        return self.__library

    @property
    def version(self):
        """Return the version name."""
        return self.__version

library property

Return the library name.

version property

Return the version name.

__init__(library, version, base_config=None)

Initialise the logger factory.

Source code in src/nskit/common/logging/library.py 
32
33
34
35
36
def __init__(self, library: str, version: str, base_config: Optional[Union[LoggingConfig, dict[str, Any]]] = None):
    """Initialise the logger factory."""
    self.__library = library
    self.__version = version
    self.__base_config = base_config

get(name, config=None, **kwargs)

Alias for the get_logger method.

Source code in src/nskit/common/logging/library.py 
44
45
46
def get(self, name, config=None, **kwargs):
    """Alias for the get_logger method."""
    return self.get_logger(name, config, **kwargs)

getLogger(name, config=None, **kwargs)

Alias for the get_logger method to provide parity with the standard logging API.

Source code in src/nskit/common/logging/library.py 
48
49
50
def getLogger(self, name, config=None, **kwargs):
    """Alias for the get_logger method to provide parity with the standard logging API."""
    return self.get_logger(name, config, **kwargs)

get_logger(name, config=None, **kwargs)

Get the library logger.

Source code in src/nskit/common/logging/library.py 
38
39
40
41
42
def get_logger(self, name, config=None, **kwargs):
    """Get the library logger."""
    if config is None:
        config = self.__base_config
    return get_library_logger(self.library, self.version, name, config, **kwargs)

getLogger(name, config=None, **kwargs)

Get the logger object.

wraps get_logger.

Source code in src/nskit/common/logging/__init__.py 
10
11
12
13
14
15
16
@wraps(get_logger)
def getLogger(name, config=None, **kwargs):
    """Get the logger object.

    wraps get_logger.
    """
    return get_logger(name, config, **kwargs)

get_library_logger(library, version, name, config=None, **kwargs)

Get a (sub)logger for a library component, which includes the library and version.

Source code in src/nskit/common/logging/library.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def get_library_logger(
    library: str, version: str, name: str, config: Optional[Union[LoggingConfig, dict[str, Any]]] = None, **kwargs
):
    """Get a (sub)logger for a library component, which includes the library and version."""
    if config is None:
        config = {"extra": {}}
    elif isinstance(config, LoggingConfig):
        config = config.model_dump()
    formatstr = get_library_log_format_string(library, version)
    library = {"name": library, "version": version}
    config["format_string"] = formatstr
    config["extra"] = config.get("extra", {})
    config["extra"].update(kwargs.pop("extra", {}))
    config = LoggingConfig(**config)
    if config.json_format:
        config.extra["library"] = library
    return get_logger(name, config, **kwargs)