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 nskit/common/configuration/__init__.py

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."""
        # TODO This probably needs a tweak to handle complex structures.
        return (
            FileConfigSettingsSource(settings_cls),
            init_settings,
            env_settings,
            dotenv_settings,
            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 nskit/common/configuration/__init__.py

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 nskit/common/configuration/__init__.py

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 nskit/common/configuration/__init__.py

@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."""
    # TODO This probably needs a tweak to handle complex structures.
    return (
        FileConfigSettingsSource(settings_cls),
        init_settings,
        env_settings,
        dotenv_settings,
        file_secret_settings,
    )

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 nskit/common/contextmanagers/chdir.py

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.target_dir.__exit__(exc_type, exc_val, exc_tb)
            except PermissionError as e:
                # Handling circular imports with LoggingConfig
                logger_factory.get_logger(__name__).warn('Unable to delete temporary directory.')
                warnings.warn(e, stacklevel=2)

__enter__()

Change to the target directory.

Source code in nskit/common/contextmanagers/chdir.py

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 nskit/common/contextmanagers/chdir.py

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.target_dir.__exit__(exc_type, exc_val, exc_tb)
        except PermissionError as e:
            # Handling circular imports with LoggingConfig
            logger_factory.get_logger(__name__).warn('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 nskit/common/contextmanagers/chdir.py

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 nskit/common/contextmanagers/env.py

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')
            logger.debug(f'New env variables: {os.environ}')
        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 nskit/common/contextmanagers/env.py

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')
        logger.debug(f'New env variables: {os.environ}')
    else:
        logger.info('No arguments set (environ, override, remove)')

__exit__(*args, **kwargs)

Reset to the original environment variables.

Source code in nskit/common/contextmanagers/env.py

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 nskit/common/contextmanagers/env.py

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 nskit/common/contextmanagers/test_extensions.py

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 nskit/common/contextmanagers/test_extensions.py

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 nskit/common/contextmanagers/test_extensions.py

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 nskit/common/contextmanagers/test_extensions.py

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 nskit/common/extensions.py

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 nskit/common/extensions.py

@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 nskit/common/extensions.py

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 nskit/common/extensions.py

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 nskit/common/extensions.py

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__).warn(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 nskit/common/io/json.py

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 nskit/common/io/json.py

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 nskit/common/io/json.py

def load(fp: TextIO, **kwargs):
    """Load JSON from file."""
    return loads(fp.read(), **kwargs)

loads(s, **kwargs)

Load JSON from string.

Source code in nskit/common/io/json.py

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 nskit/common/io/toml.py

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 nskit/common/io/toml.py

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 nskit/common/io/toml.py

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 nskit/common/io/toml.py

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 nskit/common/io/yaml.py

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 nskit/common/io/yaml.py

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 nskit/common/io/yaml.py

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 nskit/common/io/yaml.py

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 nskit/common/logging/config.py

class LoggingConfig(BaseConfiguration):
    """This is a basic config for logging."""

    # TODO setup as settings
    level: str = Field(DEFAULT_LOGLEVEL, 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', alias='json')
    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 nskit/common/logging/library.py

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 nskit/common/logging/library.py

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 nskit/common/logging/library.py

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 nskit/common/logging/library.py

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 nskit/common/logging/library.py

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 nskit/common/logging/__init__.py

@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 nskit/common/logging/library.py

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)