import logging
import pickle
from collections.abc import Iterable, Mapping, Sequence
from copy import deepcopy
from datetime import timedelta
from time import perf_counter
from typing import (
TYPE_CHECKING,
Any,
Generic,
Literal,
NoReturn,
Self,
Unpack,
overload,
)
from rics.collections.dicts import InheritedKeysDict, MakeType
from rics.collections.misc import as_list
from rics.env.read import read_bool
from rics.misc import tname
from rics.paths import AnyPath, any_path_to_path
from rics.strings import format_seconds as fmt_sec
from rics.types import LiteralHelper
from . import logging as _logging
from ._tasks import MappingTask, NamesTask, TranslationTask
from ._utils.emit_warning import emit_warning
from .exceptions import (
ConfigurationChangedError,
ConnectionStatusError,
MissingNamesError,
TransformerConflictError,
TranslationDisabledWarning,
)
from .fetching import Fetcher, MultiFetcher
from .fetching.types import IdsToFetch
from .mapping import Mapper
from .mapping.matrix import ScoreMatrix
from .mapping.types import UserOverrideFunction
from .offline import Format, TranslationMap
from .offline.types import (
FormatType,
PlaceholderTranslations,
SourcePlaceholderTranslations,
)
from .testing import TestFetcher, TestMapper
from .toml import TranslatorFactory, meta
from .transform import TransformerStack, as_transformer
from .transform._impl.stack import has_member
from .transform.types import OnExistingTransformer, Transformer, Transformers, TransformersArg
from .translator_typing import CopyParams, FetcherCopyMode, FetcherTypes
from .types import (
ID,
CopyTranslatable,
DictToId,
DictToList,
DictToOneTuple,
DictToSet,
DictToThreeTuple,
DictToTwoTuple,
DictToVarTuple,
HasSources,
IdType,
IdTypes,
InplaceTranslatable,
Names,
NameToSource,
NameType,
NameTypes,
SourceType,
Translatable,
)
LOGGER = logging.getLogger(__package__).getChild("Translator")
ON_EXISTING_HELPER: LiteralHelper[OnExistingTransformer] = LiteralHelper(
OnExistingTransformer,
default_name="on_existing",
type_name="OnExistingTransformer",
)
FETCHER_COPY_MODE_HELPER: LiteralHelper[FetcherCopyMode] = LiteralHelper(
FetcherCopyMode,
default_name="fetcher",
type_name="FetcherCopyMode",
)
ID_TRANSLATION_DISABLED = "ID_TRANSLATION_DISABLED"
_ON_EXISTING_CONFLICT_HINT = "Hint: Pass on_existing='append' to chain onto it, or 'overwrite' to replace it."
if TYPE_CHECKING:
from uuid import UUID
import pandas
ID_TRANSLATION_PANDAS_IS_TYPED: bool = False
[docs]
class Translator(Generic[NameType, SourceType, IdType], HasSources[SourceType]):
r"""End-user interface for all translation tasks.
See :meth:`Translator.translate <id_translation.Translator.translate>` for runtime configuration options. Any
argument chosen when the ``Translator`` is created can be overridden with
:meth:`Translator.copy <id_translation.Translator.copy>`. Use :meth:`~id_translation.Translator.go_offline` to store
translations in memory.
Args:
fetcher: A :class:`~id_translation.fetching.Fetcher` or ready-to-use translations.
fmt: String :class:`~id_translation.offline.Format` specification for translations.
mapper: A :class:`~id_translation.mapping.Mapper` instance for binding names to sources.
default_fmt: Alternative :class:`~id_translation.offline.Format` to use for fallback translation of unknown IDs.
default_fmt_placeholders: Shared and/or source-specific default placeholder values for unknown IDs. See
:meth:`InheritedKeysDict.make() <rics.collections.dicts.InheritedKeysDict.make>` for details.
enable_uuid_heuristics: Improves matching when :py:class:`~uuid.UUID`-like IDs are in use.
transformers: A dict ``{source: transformer}`` of initialized
:class:`~id_translation.transform.types.Transformer` instances. Values may be a sequence to chain per
source. Equivalent to calling :meth:`~id_translation.Translator.register_transformer` for each key, which
is the way to register once the :attr:`~id_translation.types.HasSources.sources` are known.
.. _translator-docstring-example:
Examples:
Basic usage. For a more complete use case, see the :ref:`dvdrental` example.
Assume that we have data for people and animals as in the tables below::
people: animals:
id | name id | name | is_nice
------+--------- ----+--------+---------
1991 | Richard 0 | Tarzan | false
1999 | Sofia 1 | Morris | true
1904 | Fred 2 | Simba | true
In most real cases we'd fetch this table from somewhere. In this case, however, there's so little data that we
can simply enumerate the components needed for translation ourselves.
>>> from id_translation import Translator
>>> translation_data = {
... "animals": {
... "id": [0, 1, 2],
... "name": ["Tarzan", "Morris", "Simba"],
... "is_nice": [False, True, True],
... },
... "people": {1999: "Sofia", 1991: "Richard", 1904: "Fred"},
... }
>>> fmt = "{id}:{name}[, nice={is_nice}]"
>>> translator = Translator(translation_data, fmt=fmt)
Since `people` only has columns `id` and `name`, we can use the simplified ``{id: name}`` data format. We're
using the full format for `animals` since we have an additional `is_nice` column in this table. We didn't define
a :class:`~id_translation.mapping.Mapper`, so the column names must match exactly.
>>> import pandas as pd
>>> df = pd.DataFrame({"animals": [0, 2], "people": [1991, 1999]})
>>> translator.translate(df) # Returns a copy
animals people
0 0:Tarzan, nice=False 1991:Richard
1 2:Simba, nice=True 1999:Sofia
Check out the :ref:`translation-primer` to learn how this is done "under the hood".
"""
def __init__(
self,
fetcher: FetcherTypes[NameType, SourceType, IdType] | None = None,
*,
fmt: FormatType = Format.DEFAULT,
mapper: Mapper[NameType, SourceType, None] | None = None,
default_fmt: FormatType = Format.DEFAULT_FAILED,
default_fmt_placeholders: MakeType[SourceType, str, Any] | None = None,
enable_uuid_heuristics: bool = False,
transformers: TransformersArg[SourceType, IdType] | None = None,
) -> None:
# The data source, assigned below. Declared first because `_register_transformer` reads both to decide
# whether the sources are known yet, and the loop below runs before either has an answer.
self._fetcher: Fetcher[SourceType, IdType] | None = None
self._cached_tmap: TranslationMap[NameType, SourceType, IdType] | None = None
self._transformers: Transformers[SourceType, IdType] = {}
# Registrations not yet checked against the known sources; see `_warn_on_transformers_for_unknown_sources`.
# The transitions that turn this instance into a snapshot discard what is queued instead of checking it,
# which is what exempts those registrations for good.
self._unverified_sources: list[SourceType] = []
# Whether the `Fetcher.get_transformer` pass has run; see `_apply_fetcher_transformers`. Nothing is
# cached from it, so a pass that raised simply runs again.
self._transformers_queried: bool = False
for source, transformer in (transformers or {}).items():
# Not the public method: an override may rely on attributes this constructor has not created yet.
self._register_transformer(source, transformer, on_existing="raise")
self._fmt = Format.parse(fmt)
self._default_fmt_placeholders, self._default_fmt = _handle_default(default_fmt, default_fmt_placeholders)
self._enable_uuid_heuristics = enable_uuid_heuristics
if fetcher is None:
self._fetcher = TestFetcher([]) # No explicit sources
if mapper: # pragma: no cover
emit_warning(
f"Mapper instance {mapper} given; consider creating a TestFetcher([sources..])-instance manually."
)
else:
mapper = TestMapper()
emit_warning("No fetcher given. Translation data will be automatically generated.")
elif isinstance(fetcher, Fetcher):
self._fetcher = fetcher
elif isinstance(fetcher, dict):
source_translations = {source: PlaceholderTranslations.make(source, pht) for source, pht in fetcher.items()}
self._cached_tmap = self._to_translation_map(source_translations)
elif isinstance(fetcher, TranslationMap):
# Extract raw translations and "overwrite" inherited properties, such as format specs and transformers.
source_translations = fetcher._extract_translations()
self._cached_tmap = self._to_translation_map(source_translations)
else:
raise TypeError(type(fetcher)) # pragma: no cover
self._mapper: Mapper[NameType, SourceType, None] = mapper or Mapper()
self._mapper.logger = LOGGER.getChild("map")
self._config_metadata: meta.ConfigMetadata | None = None
self._translated_names: tuple[int, NameToSource[NameType, SourceType]] | None = None
if isinstance(fetcher, dict):
# `initialize_sources()` never runs offline, so this is the only check inline data gets; see
# `register_transformer` for why a snapshot is exempt.
self._warn_on_transformers_for_unknown_sources()
elif isinstance(fetcher, TranslationMap):
# A snapshot may cover a subset of the original sources on purpose, so what the constructor registered is
# dropped unchecked rather than reported. Registrations made from here on are queued and checked as usual.
self._unverified_sources.clear()
def __getstate__(self) -> Any:
# `go_offline()` guarantees a serializable instance. Delegates, since a subclass may declare `__slots__`;
# `object.__getstate__` reports that state as the second element of a tuple, which the plain-`__dict__`
# case does not use.
state = super().__getstate__()
instance_dict, slots = state if isinstance(state, tuple) else (state, None)
instance_dict = dict(instance_dict or {})
return (instance_dict, slots) if slots else instance_dict
def __setstate__(self, state: Any) -> None:
slots: dict[str, Any] = {}
if isinstance(state, tuple): # A subclass with `__slots__`; see `__getstate__`.
state, slots = state[0] or {}, state[1] or {}
# Snapshots written by older versions predate fetcher-provided transformers; see `restore`. `False` is
# right for those: nothing was ever asked, and restored instances are offline, so the state is inert until the
# copy is given a fetcher. The empty queue is not inert: it is read while offline, and an old snapshot's
# registrations were exempt in all but name, as `go_offline` says.
state.setdefault("_transformers_queried", False)
state.setdefault("_unverified_sources", [])
# Offline snapshots used to record their state by omitting the fetcher rather than by nulling it.
state.setdefault("_fetcher", None)
self.__dict__.update(state)
for key, value in slots.items():
setattr(self, key, value)
[docs]
@classmethod
def from_config(
cls,
path: AnyPath,
extra_fetchers: Iterable[AnyPath] = (),
) -> Self:
"""Create a :class:`~id_translation.Translator` from TOML inputs.
See :ref:`translator-config` for help.
Args:
path: Path to the main TOML configuration file.
extra_fetchers: Paths to fetching configuration TOML files. If multiple fetchers are defined, they are
ranked by input order. A fetcher defined in the main configuration is prioritized (rank=0).
Returns:
A new :class:`~id_translation.Translator` instance with a :attr:`~id_translation.Translator.config_metadata`
attribute.
"""
# docs: https://id-translation.readthedocs.io/en/stable/documentation/translator-config.html
factory = TranslatorFactory(path, extra_fetchers, cls)
return factory.create() # type: ignore[return-value]
@property
def config_metadata(self) -> meta.ConfigMetadata:
"""Return :func:`~id_translation.Translator.from_config` initialization :class:`metadata <id_translation.toml.meta.ConfigMetadata>`.
Raises:
ValueError: If this instance was not created by :meth:`~id_translation.Translator.from_config`, or copied
from such an instance.
"""
if self._config_metadata is None:
raise ValueError("Not created using Translator.from_config()") # pragma: no cover
return self._config_metadata
[docs]
def initialize_sources(self, task_id: int | None = None) -> Self:
"""Perform source discovery (fetcher initialization).
This method does nothing if the ``Translator`` isn't :attr:`~id_translation.Translator.online`.
Args:
task_id: Used for logging.
Returns:
Self, for chained assignment.
See Also:
🧵 Call this before sharing a ``Translator`` between threads; see :ref:`thread-safety`.
"""
if self.online:
if task_id is None:
task_id = _logging.generate_task_id()
self.fetcher.initialize_sources(task_id)
self._apply_fetcher_transformers()
return self
[docs]
def copy(self, **overrides: Unpack[CopyParams[NameType, SourceType, IdType]]) -> Self:
"""Make a copy of this :class:`~id_translation.Translator`.
Args:
overrides: Keyword arguments to use when instantiating the copy. Options that aren't given will be taken
from the current instance. See the :class:`~id_translation.Translator` class documentation for possible
choices (except `fetcher`; see Notes).
Returns:
A copy of this :class:`~id_translation.Translator` with `overrides` applied.
Notes:
User types are copied using :func:`copy.deepcopy`. The `fetcher` override selects how the current
:attr:`~id_translation.Translator.fetcher` is carried over -- ``'keep'`` reuses it as-is, ``'copy'``
requires a successful :func:`~copy.deepcopy`, and the default, ``'auto'``, tries to clone and falls back
to ``'keep'`` with a warning if that fails; see
:attr:`~id_translation.translator_typing.FetcherCopyMode`. This only applies while
:attr:`~id_translation.Translator.online`; an offline copy always shares the cached translation records
with the original, which are never mutated in place.
Fetcher-provided transformers already derived by this instance are carried over via the `transformers`
argument, and are not derived again; see :meth:`Fetcher.get_transformer()
<id_translation.fetching.Fetcher.get_transformer>`. This holds for all three `fetcher` modes.
Passing `transformers` explicitly replaces that set wholesale -- including anything the fetcher provided
-- and the copy does not query the fetcher for more. Pass
:attr:`~id_translation.Translator.transformers` to keep the current set, or ``None`` to start empty and
derive on first use, as the constructor does.
.. deprecated:: 1.3.0
Passing anything but a :attr:`~id_translation.translator_typing.FetcherCopyMode` as `fetcher`. Use
``fetcher='keep'`` to reuse the current :attr:`~id_translation.Translator.fetcher`, which is what the
failed-clone hint used to recommend. Replacing the data source will raise in ``id-translation==2.0.0``;
construct a new :class:`~id_translation.Translator` instead.
"""
cls = type(self)
fetcher_mode = self._pop_fetcher_copy_mode(overrides)
kwargs: dict[str, Any] = {
"fmt": self.fmt,
"default_fmt": self.default_fmt,
"default_fmt_placeholders": self._default_fmt_placeholders,
"enable_uuid_heuristics": self.enable_uuid_heuristics,
**overrides,
}
if "mapper" not in kwargs:
kwargs["mapper"] = self.mapper.copy()
if "fetcher" not in kwargs: # TODO(2.0.0): Unconditional; `fetcher` is a FetcherCopyMode only.
kwargs["fetcher"] = self._copy_fetcher(fetcher_mode)
if "transformers" not in kwargs:
kwargs["transformers"] = self._copy_transformers()
# Derived results travel with the copies; a second pass over the same fetcher would self-conflict. Only a
# (deprecated) replacement is unqueried; `fetcher_mode` never substitutes the data source.
settled = "fetcher" not in overrides and self._transformers_queried
else:
# An explicit set is complete, so it settles the question either way -- without this, the same call
# would query or not depending on whether the caller had translated yet. `None` means "derive on first
# use", exactly as it does in the constructor.
settled = kwargs["transformers"] is not None
new = cls(**kwargs)
new._config_metadata = self._config_metadata
new._transformers_queried = settled
return new
def _pop_fetcher_copy_mode(self, overrides: CopyParams[NameType, SourceType, IdType]) -> FetcherCopyMode:
"""Resolve the `fetcher` override. A (deprecated) replacement data source is left in `overrides`."""
if "fetcher" not in overrides:
return "auto"
fetcher = overrides["fetcher"]
if isinstance(fetcher, str):
del overrides["fetcher"]
return FETCHER_COPY_MODE_HELPER.check(fetcher)
# TODO(2.0.0): Remove the rest; `fetcher` accepts only FetcherCopyMode.
cls = type(self).__name__
current = self._fetcher if self.online else self._cached_tmap
if fetcher is not None and fetcher is current:
del overrides["fetcher"]
emit_warning(
f"Passing the current data source to {cls}.copy(fetcher=...) is deprecated; use fetcher='keep'."
"\nWARNING: This will raise in `id-translation==2.0.0`.",
FutureWarning,
)
return "keep"
emit_warning(
f"{cls}.copy(fetcher=...) replaces the data source, which will be fixed for the lifetime of a {cls}."
f"\nWARNING: This will raise in `id-translation==2.0.0`. Construct a new {cls} instead.",
FutureWarning,
)
return "auto" # Unused; the replacement stays in `overrides`.
def _copy_fetcher(
self, mode: FetcherCopyMode
) -> Fetcher[SourceType, IdType] | TranslationMap[NameType, SourceType, IdType]:
if not self.online:
return self.cache # Never mutated in place, so every mode shares it.
if mode == "keep":
return self.fetcher
try:
return deepcopy(self.fetcher)
except TypeError as e:
hint = f"{type(self).__name__}.copy(fetcher='keep')"
if mode == "copy":
e.add_note(f"Hint: Could not clone {tname(self.fetcher, include_module=True)}. Use {hint} to reuse it.")
raise
emit_warning(
f"Failed to clone fetcher (TypeError: {e}). Caller instance will be reused."
f"\nHint: To suppress this warning: {hint}"
)
return self.fetcher
def _copy_transformers(self) -> Transformers[SourceType, IdType]:
try:
# One `deepcopy` call shares one memo internally, so sharing is preserved at any depth for free.
return deepcopy(self._transformers)
except TypeError:
pass # Retry per source below, to attribute the failure and reuse only what cannot be cloned.
retval: Transformers[SourceType, IdType] = {}
# Not a `deepcopy` memo: mapping a failed clone to its original there would splice the original into any
# later copy that embeds it -- silently, and only for the registration orders where the failure came first.
clones: dict[int, Transformer[IdType]] = {}
for source, transformer in self._transformers.items():
if (clone := clones.get(id(transformer))) is None:
try:
clone = deepcopy(transformer)
except TypeError as e:
cls = self.__class__.__name__
emit_warning(
f"Failed to clone the transformer for {source!r} (TypeError: {e})."
" Caller instance will be reused."
f"\nHint: {cls}.copy(transformers={cls}.transformers) also disables fetcher-provided"
" discovery on the copy; see Translator.copy().",
)
clone = transformer
clones[id(transformer)] = clone
retval[source] = clone
return retval
if TYPE_CHECKING:
@overload
def translate(
self,
translatable: InplaceTranslatable[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
# https://github.com/python/mypy/issues/7333#issuecomment-788255229
copy: Literal[False],
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> None: ...
@overload
def translate(
self,
translatable: CopyTranslatable[IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
# https://github.com/python/mypy/issues/7333#issuecomment-788255229
copy: Literal[False],
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> NoReturn: ...
@overload
def translate(
self,
translatable: IdTypes,
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> str: ...
@overload
def translate(
self,
translatable: list[IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> list[str]: ...
@overload
def translate(
self,
translatable: set[IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> set[str]: ...
@overload
def translate(
self,
# This is not correct, but using the TypeVar (which the user may bind to get proper typing) doesn't work
# at this time (python 3.11.3, mypy 1.5.1). Higher-Kinded TypeVars might solve this.
# TODO: Higher-Kinded TypeVars
translatable: DictToId[NameType, int] | DictToId[NameType, str] | DictToId[NameType, UUID],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToId[NameType, str]: ...
@overload
def translate(
self,
translatable: DictToSet[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToSet[NameType, str]: ...
@overload
def translate(
self,
translatable: DictToList[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToList[NameType, str]: ...
@overload
def translate( # Overlaps with DictToVarTuple
self,
translatable: DictToOneTuple[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToOneTuple[NameType, str]: ...
@overload
def translate( # Overlaps with DictToVarTuple
self,
translatable: DictToTwoTuple[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToTwoTuple[NameType, str]: ...
@overload
def translate( # Overlaps with DictToVarTuple
self,
translatable: DictToThreeTuple[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToThreeTuple[NameType, str]: ...
@overload
def translate(
self,
translatable: DictToVarTuple[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> DictToVarTuple[NameType, str]: ...
@overload
def translate(
self,
translatable: tuple[IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> tuple[str]: ...
@overload
def translate(
self,
translatable: tuple[IdType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> tuple[str, str]: ...
@overload
def translate(
self,
translatable: tuple[IdType, IdType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> tuple[str, str, str]: ...
@overload
def translate(
self,
translatable: tuple[IdType, ...],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> tuple[str, ...]: ...
if ID_TRANSLATION_PANDAS_IS_TYPED:
# pandas-stubs or similar
@overload
def translate(
self,
translatable: "pandas.DataFrame",
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> "pandas.DataFrame": ...
@overload
def translate(
self,
translatable: "pandas.Series[Any]",
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> "pandas.Series[str]": ...
@overload
def translate(
self,
translatable: "pandas.Index[Any]",
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[True] = True,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> "pandas.Index[str]": ...
@overload
def translate(
self,
translatable: "pandas.DataFrame" | "pandas.Series[Any]",
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
copy: Literal[False],
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> None: ...
@overload
def translate(
self,
translatable: "pandas.Index[Any]",
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
# https://github.com/python/mypy/issues/7333#issuecomment-788255229
copy: Literal[False],
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
# Translation specification
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> NoReturn: ...
@overload
def translate(
self,
translatable: Translatable[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
copy: Literal[True] = True,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
reverse: Literal[False] = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> Translatable[NameType, str]: ...
@overload
def translate(
self,
translatable: Translatable[NameType, str],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
copy: Literal[True] = True,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
reverse: Literal[True],
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> Translatable[NameType, IdType]: ...
[docs]
def translate(
self,
translatable: Translatable[NameType, IdType],
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
copy: bool = True,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
reverse: bool = False,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> Translatable[NameType, str] | None:
"""Translate IDs to human-readable strings.
Simplified process:
1. The :meth:`~id_translation.Translator.extract_names` method derives names of the `translatable` to translate (if needed).
2. The :meth:`~id_translation.Translator.map` method performs name-to-source mapping (see
:class:`~id_translation.mapping.DirectionalMapping`).
3. The :meth:`~id_translation.Translator.fetch` method extracts IDs to translate and retrieves data (see
:class:`~id_translation.offline.TranslationMap`).
4. Finally, the ``translate()`` method (i.e. this one) applies the translations and returns to the caller.
See the :ref:`translation-primer` page for a detailed process description.
Args:
translatable: A data structure to translate.
names: Explicit names to translate. Derive from `translatable` if ``None``. Alternatively, you may pass a
``dict`` on the form ``{name_in_translatable: source_to_use}``.
ignore_names: Names **not** to translate, or a predicate ``(NameType) -> bool``.
copy: If ``False``, translate in-place and return ``None``.
override_function: A callable ``(name, sources, context) -> Source | None``, where `context` is always
``None``. See :meth:`Mapper.apply <id_translation.mapping.Mapper.apply>` for details.
max_fails: The maximum fraction of IDs for which translation may fail. 1=disabled. Missing IDs (e.g.
``NaN``) in ``pandas`` types are pass-throughs and never count; a ``None`` in a builtin collection is
an ordinary -- unknown -- ID, and does.
reverse: If ``True``, perform translations back to IDs. Offline mode only.
fmt: A :class:`format string <id_translation.offline.Format>` such as **'{id}:{name}'** to use. Default is
:attr:`Translator.fmt <id_translation.Translator.fmt>`.
io_kwargs: Keyword arguments for the IO class (e.g.
:class:`~id_translation.dio.integration.pandas.PandasIO`).
Returns:
A translated copy of `translatable` if ``copy=True``, otherwise ``None``.
Examples:
Manual :ref:`name-to-source <name-to-source-mapping>` mapping with a temporary name-only
:class:`~id_translation.offline.Format`.
..
# Hidden setup code
>>> translator = Translator({"animals": {"id": [2], "name": ["Simba"]}})
>>> n2s = {"lions": "animals", "big_cats": "animals"}
>>> translator.translate({"lions": 2, "big_cats": 2}, names=n2s, fmt="{name}")
{'lions': 'Simba', 'big_cats': 'Simba'}
Name mappings must be complete; any name not present in the keys will be ignored (left as-is).
Raises:
~id_translation.dio.exceptions.UntranslatableTypeError: If ``type(translatable)`` cannot be translated.
~id_translation.exceptions.MissingNamesError: If `names` are not given and cannot be derived from
`translatable`.
~id_translation.mapping.exceptions.UnmappedExplicitNamesError: If any required (explicitly given) names fail
to map to a source.
~id_translation.mapping.exceptions.MappingError: If name-to-source mapping fails, e.g.
:class:`~id_translation.mapping.exceptions.ScoringDisabledError` for a name that is neither a source
nor overridden, when using the default :class:`~id_translation.mapping.Mapper`.
~id_translation.exceptions.MissingPlaceholderError: If a source lacks a placeholder that `fmt` requires.
ValueError: If `max_fails` is not a valid fraction.
~id_translation.exceptions.TooManyFailedTranslationsError: If translation fails for more than `max_fails` of
IDs.
~id_translation.exceptions.ConnectionStatusError: If ``reverse=True`` while the
:class:`~id_translation.Translator` is online.
~id_translation.mapping.exceptions.UserMappingError: If `override_function` returns a source which is not
known, and ``mapper.on_unknown_user_override='raise'``.
See Also:
🔑 This is a key event method. See :ref:`key-events` for details.
The :envvar:`ID_TRANSLATION_DISABLED` variable.
"""
if read_bool(ID_TRANSLATION_DISABLED):
message = f"Translation aborted; {ID_TRANSLATION_DISABLED} is set."
LOGGER.warning(message)
emit_warning(message, category=TranslationDisabledWarning, logged=True)
return translatable if copy else None # Return unchanged; breaks typing.
if self.online and reverse: # pragma: no cover
raise ConnectionStatusError("Reverse translation cannot be performed online.")
task_id = _logging.generate_task_id()
self.initialize_sources(task_id)
task = TranslationTask(
self,
translatable,
self._fmt if fmt is None else Format.parse(fmt),
names,
ignore_names=ignore_names,
override_function=override_function,
copy=copy,
max_fails=max_fails,
reverse=reverse,
enable_uuid_heuristics=self._enable_uuid_heuristics,
io_kwargs=io_kwargs,
event_key=f"{type(self).__name__}.{self.translate.__name__}",
task_id=task_id,
)
task.log_key_event_enter()
translation_map = self._get_updated_tmap(task)
if not translation_map:
# Return unchanged; this is technically against the API spec. If the user has required translation to
# success through configuration, exceptions will be raised elsewhere. I don't know how to express this using
# the Python type system.
return translatable if copy else None
task.verify(translation_map)
ans = task.insert(translation_map)
self._translated_names = (task_id, dict(task.name_to_source))
task.finished()
task.log_key_event_exit()
return ans
@overload
def translated_names(self, with_source: Literal[True]) -> NameToSource[NameType, SourceType]: ...
@overload
def translated_names(self, with_source: Literal[False] = False) -> list[NameType]: ...
[docs]
def translated_names(self, with_source: bool = True) -> NameToSource[NameType, SourceType] | list[NameType]:
"""Return the names that were translated by the most recent :meth:`~id_translation.Translator.translate`-call.
Args:
with_source: If ``True``, return a dict ``{name: source}`` instead of a list.
Returns:
Recent names translated by this :class:`~id_translation.Translator`, in **arbitrary** order.
Raises:
ValueError: If no names have been translated using this :class:`~id_translation.Translator`.
Notes:
🧵 This method is not thread safe. See :ref:`thread-safety` for details.
"""
if self._translated_names is None:
raise ValueError("No names have been translated using this Translator.")
task_id, name_to_source = self._translated_names
LOGGER.debug(f"Returning names translated by task with {task_id=}.", extra=dict(task_id=task_id))
return dict(name_to_source) if with_source else list(name_to_source)
@overload
def extract_names(
self,
translatable: Translatable[NameType, IdType],
*,
ignore_names: Names[NameType] | None = None,
io_kwargs: Mapping[str, Any] | None = None,
raising: Literal[False] = False,
) -> list[NameType] | None: ...
@overload
def extract_names(
self,
translatable: Translatable[NameType, IdType],
*,
ignore_names: Names[NameType] | None = None,
io_kwargs: Mapping[str, Any] | None = None,
raising: Literal[True],
) -> list[NameType]: ...
[docs]
def map(
self,
translatable: Translatable[NameType, IdType],
names: NameTypes[NameType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> NameToSource[NameType, SourceType]:
"""Map names to translation sources.
Args:
translatable: A data structure to map names for.
names: Explicit names to translate. Derive from `translatable` if ``None``.
ignore_names: Names **not** to translate, or a predicate ``(NameType) -> bool``.
override_function: A callable ``(name, sources, context) -> Source | None``, where `context` is always
``None``. See :meth:`Mapper.apply <id_translation.mapping.Mapper.apply>` for details.
io_kwargs: Keyword arguments for the IO class (e.g.
:class:`~id_translation.dio.integration.pandas.PandasIO`).
Returns:
A mapping of names to translation sources.
Raises:
~id_translation.exceptions.MissingNamesError: If `names` are not given and cannot be derived from
`translatable`.
~id_translation.mapping.exceptions.UnmappedExplicitNamesError: If any required (explicitly given) names fail
to map to a source.
~id_translation.mapping.exceptions.MappingError: If name-to-source mapping fails, e.g.
:class:`~id_translation.mapping.exceptions.ScoringDisabledError` for a name that is neither a source
nor overridden, when using the default :class:`~id_translation.mapping.Mapper`.
~id_translation.mapping.exceptions.UserMappingError: If `override_function` returns a source which is not
known, and ``mapper.on_unknown_user_override='raise'``.
See Also:
🔑 This is a key event method. See :ref:`key-events` for details.
The :meth:`~id_translation.Translator.extract_names` method.
"""
task_id = _logging.generate_task_id()
self.initialize_sources(task_id)
return MappingTask(
self,
translatable,
names,
ignore_names=ignore_names,
override_function=override_function,
io_kwargs=io_kwargs,
task_id=task_id,
).name_to_source
[docs]
def map_scores(
self,
translatable: Translatable[NameType, IdType],
names: NameTypes[NameType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> ScoreMatrix[NameType, SourceType]:
"""Returns raw match scores for name-to-source mapping. See :meth:`~id_translation.Translator.map` for details."""
return MappingTask(
self,
translatable,
names,
ignore_names=ignore_names,
override_function=override_function,
io_kwargs=io_kwargs,
).compute_scores()
@property
def sources(self) -> list[SourceType]:
"""A list of known source names.
Sources are determined either by the :attr:`~id_translation.Translator.fetcher` or the
:attr:`~id_translation.Translator.cache`.
"""
if not self.online and self._cached_tmap is not None:
# Not by way of `placeholders`: that property copies every per-source placeholder list.
return self._cached_tmap.sources
return list(self.placeholders)
@property
def placeholders(self) -> Mapping[SourceType, Sequence[str]]:
"""A mapping ``{source: [placeholders, ...]}``.
Placeholders shown here are the names as they appear **in the source**.
"""
if self.online:
return self.fetcher.placeholders
assert self._cached_tmap is not None, "bad internal state" # noqa: S101
return self._cached_tmap.placeholders
@property
def fmt(self) -> Format:
"""Main translation :class:`~id_translation.offline.Format` for this ``Translator`` instance."""
return self._fmt
@property
def default_fmt(self) -> Format:
"""Alternative translation :class:`~id_translation.offline.Format`, used for unknown IDs."""
return self._default_fmt
@property
def enable_uuid_heuristics(self) -> bool:
"""Improves matching when :py:class:`~uuid.UUID`-like IDs are in use."""
return self._enable_uuid_heuristics
@property
def online(self) -> bool:
"""Return connectivity status. If ``False``, no new translations may be fetched."""
return self._fetcher is not None
@property
def fetcher(self) -> Fetcher[SourceType, IdType]:
"""Return the :class:`~id_translation.fetching.Fetcher` instance used to retrieve translations."""
# Not `self.online`: narrowing the attribute here is what lets the rest of the class use this property
# without a guard of its own.
if self._fetcher is None:
raise ConnectionStatusError(
"Cannot fetch new translations.",
hints="Use the Translator.cache-property to access the data.",
)
return self._fetcher
@property
def mapper(self) -> Mapper[NameType, SourceType, None]:
"""Return the :class:`~id_translation.mapping.Mapper` instance used for name-to-source binding."""
return self._mapper
@property
def cache(self) -> TranslationMap[NameType, SourceType, IdType]:
"""Return a :class:`~id_translation.offline.TranslationMap` of cached translations."""
if self._cached_tmap is None:
assert self.online, "bad internal state" # noqa: S101
msg = f"No cache available. Use `{self.go_offline.__qualname__}` to cache translations."
raise RuntimeError(msg)
return self._cached_tmap
[docs]
@classmethod
def load_persistent_instance(
cls,
cache_dir: AnyPath,
config_path: AnyPath,
extra_fetchers: Iterable[AnyPath] = (),
max_age: str | timedelta | None = "12h",
on_config_changed: Literal["raise", "recreate"] = "recreate",
) -> Self:
"""Load or create a persistent, offline :class:`~id_translation.Translator`.
Instances are created, stored and loaded as determined by a metadata file located in the given `cache_dir`. A
new :class:`~id_translation.Translator` will be created if:
* There is no `'metadata'` file, or
* the original :class:`~id_translation.Translator` is too old (see `max_age`), or
* the current configuration, as defined by `config_path` and `extra_fetchers`, has changed in such a way
that it is no longer equivalent to the configuration used to create the original
:class:`~id_translation.Translator`. Equivalence also covers the ``Translator`` class, ``metaconf.toml``,
and the Python and package versions, so upgrading ``id-translation`` counts as a change. For details, see
:class:`~id_translation.toml.meta.ConfigMetadata`.
Args:
cache_dir: Root directory where the cached translator and associated metadata is stored.
config_path: Path to the main TOML configuration file.
extra_fetchers: Paths to fetching configuration TOML files. If multiple fetchers are defined, they are
ranked by input order. A fetcher defined in the main configuration is prioritized (rank=0).
max_age: The maximum age of the cached :class:`~id_translation.Translator` before it must be recreated. Pass
``'0s'`` or ``timedelta(0)`` to force recreation, or ``None`` to ignore.
on_config_changed: One of ``raise|recreate``. If ``'raise'``, crash instead of creating a new instance if
the configuration (as determined by `config_path` and `extra_fetchers`) has changed, including after a
version upgrade.
Returns:
A new or cached :class:`~id_translation.Translator` instance with a
:attr:`~id_translation.Translator.config_metadata` attribute.
Raises:
~id_translation.exceptions.ConfigurationChangedError: If the configuration has changed and
``on_config_changed='raise'``.
Notes:
🧵 This method is not thread safe. See :ref:`thread-safety` for details.
See Also:
The :meth:`~id_translation.Translator.from_config` method, which will read the `config_path`.
"""
path = any_path_to_path(config_path)
cache_dir = any_path_to_path(cache_dir).expanduser().absolute()
cache_dir.mkdir(parents=True, exist_ok=True)
metadata_path = cache_dir / "metadata.json"
cache_path = cache_dir / "translator.pkl"
extra_fetcher_paths = list(map(str, extra_fetchers))
metadata = meta.ConfigMetadata.from_toml_paths(
str(path),
extra_fetcher_paths,
clazz=cls,
)
use_cached, reason, reason_type = metadata.use_cached(metadata_path, max_age)
if use_cached:
LOGGER.info(f"Reuse existing Translator; {reason}. Cache dir: '{cache_dir}'.")
return cls.restore(cache_path)
if reason_type == "metadata-changed" and on_config_changed.lower() == "raise":
raise ConfigurationChangedError(reason)
LOGGER.info(f"Create new Translator; {reason}. Cache dir: '{cache_dir}'.")
translator = cls.from_config(path, extra_fetcher_paths)
translator.go_offline(path=cache_path)
metadata_path.write_text(translator.config_metadata.to_json(), encoding="utf-8")
return translator
[docs]
@classmethod
def restore(cls, path: AnyPath) -> Self:
"""Restore a serialized :class:`~id_translation.Translator`.
Args:
path: Path to a serialized :class:`~id_translation.Translator`.
Returns:
A :class:`~id_translation.Translator`.
Raises:
TypeError: If the object at `path` is not a :class:`~id_translation.Translator` or a subtype thereof.
See Also:
The :meth:`~id_translation.Translator.go_offline` method.
"""
full_path = any_path_to_path(path).expanduser()
with full_path.open("rb") as f:
ans = pickle.load(f) # noqa: S301
if type(ans) is not cls: # pragma: no cover
raise TypeError(f"Serialized object at '{full_path}' is a {type(ans)}, not {cls}.")
if LOGGER.isEnabledFor(logging.DEBUG):
extra = "" if ans._config_metadata is None else f" with {ans.config_metadata}"
LOGGER.debug(f"Deserialized {ans}{extra}.")
return ans
[docs]
def go_offline(
self,
translatable: Translatable[NameType, IdType] | None = None,
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
path: AnyPath | None = None,
) -> Self:
"""Retrieve and :attr:`~id_translation.Translator.cache` translations in memory.
.. warning::
The :attr:`~id_translation.Translator.fetcher` will be destroyed.
Args:
translatable: Data from which IDs to fetch will be extracted. Fetch all IDs if ``None``.
names: Explicit names to translate. Derive from `translatable` if ``None``.
ignore_names: Names **not** to translate, or a predicate ``(NameType) -> bool``.
override_function: A callable ``(name, sources, context) -> Source | None``, where `context` is always
``None``. See :meth:`Mapper.apply <id_translation.mapping.Mapper.apply>` for details. Ignored when
`translatable` is given.
max_fails: The maximum fraction of IDs for which translation may fail. 1=disabled. Missing IDs (e.g.
``NaN``) in ``pandas`` types are pass-throughs and never count; a ``None`` in a builtin collection is
an ordinary (unknown) ID, and does.
fmt: A :class:`format string <id_translation.offline.Format>` such as **'{id}:{name}'** to use. Default is
:attr:`Translator.fmt <id_translation.Translator.fmt>`. The snapshot is only guaranteed to hold the
placeholders of `fmt` when `translatable` is given, and those of `Translator.fmt` otherwise; list any
others that a later `translate()` needs there as optional blocks, e.g. ``'{id}:{name}[ {team}]'``.
io_kwargs: Keyword arguments for the IO class (e.g.
:class:`~id_translation.dio.integration.pandas.PandasIO`).
path: If given, serialize the :class:`~id_translation.Translator` to disk after retrieving data.
Returns:
Self, for chained assignment.
Raises:
~id_translation.exceptions.ConnectionStatusError: If already disconnected from the fetcher, i.e. not
:attr:`~id_translation.Translator.online`.
~id_translation.fetching.exceptions.ForbiddenOperationError: If
:meth:`Fetcher.fetch_all <id_translation.fetching.Fetcher.fetch_all>` is disabled and
``translatable=None``.
ValueError: If `io_kwargs` is given without `translatable`.
~id_translation.mapping.exceptions.MappingError: If :meth:`~id_translation.Translator.map` fails (only when
`translatable` is given).
Notes:
🧵 This method is not thread safe. See :ref:`thread-safety` for details.
The :class:`~id_translation.Translator` is guaranteed to be :func:`~rics.misc.serializable` once offline.
Fetchers often aren't as they require things like database connections to function.
See Also:
🔑 This is a key event method. See :ref:`key-events` for details.
The :meth:`~id_translation.Translator.restore` method (when `path` is set).
"""
if not self.online:
raise ConnectionStatusError("Cannot fetch new translations.")
start = perf_counter()
task_id = _logging.generate_task_id(start)
self.initialize_sources(task_id)
if LOGGER.isEnabledFor(logging.DEBUG):
LOGGER.debug(
msg=f"Begin going offline with {len(self.sources)} sources provided by: {self.fetcher}",
extra=dict(
task_id=task_id,
event_key=_logging.get_event_key(self.go_offline, "enter"),
# Task-specific
path=None if path is None else str(path),
translatable_type=None if translatable is None else tname(translatable, include_module=True),
),
)
translation_map = self._user_fetch(
translatable,
names,
ignore_names=ignore_names,
override_function=override_function,
max_fails=max_fails,
fmt=fmt,
io_kwargs=io_kwargs,
func=self.go_offline.__qualname__,
task_id=task_id,
)
self.fetcher.close()
self._fetcher = None
self._cached_tmap = translation_map
# This produces the same kind of snapshot as building from one directly; see the `TranslationMap` branch in
# `__init__`. The registrations up to this point may cover only a subset of sources on purpose.
self._unverified_sources.clear()
if path:
path = any_path_to_path(path).expanduser()
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("wb") as f:
pickle.dump(self, f)
cls = tname(self, include_module=True)
LOGGER.info(
f"Serialized '{cls}' of size {path.stat().st_size / 2**20:.2g} MiB at path='{path}'.",
extra=dict(
task_id=task_id,
event_key=_logging.get_event_key(self.go_offline, "serialize"),
# Task-specific
path=str(path),
translatable_type=None if translatable is None else tname(translatable, include_module=True),
),
)
if LOGGER.isEnabledFor(logging.INFO):
seconds = perf_counter() - start
LOGGER.info(
f"Went offline with {len(translation_map.sources)} sources in {fmt_sec(seconds)}: {translation_map}.",
extra=dict(
task_id=task_id,
event_key=_logging.get_event_key(self.go_offline, "exit"),
seconds=seconds,
# Task-specific
path=None if path is None else str(path),
translatable_type=None if translatable is None else tname(translatable, include_module=True),
),
)
return self
[docs]
def fetch(
self,
translatable: Translatable[NameType, IdType] | None = None,
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
) -> TranslationMap[NameType, SourceType, IdType]:
"""Fetch translations.
Calling ``fetch`` without arguments will perform a
:meth:`Fetcher.fetch_all <id_translation.fetching.Fetcher.fetch_all>` -operation, without going offline.
The returned :class:`~id_translation.offline.TranslationMap` may be converted to native types with
:meth:`TranslationMap.to_dicts <id_translation.offline.TranslationMap.to_dicts>`.
Args:
translatable: A data structure to translate.
names: Explicit names to translate. Derive from `translatable` if ``None``. Alternatively, you may pass a
``dict`` on the form ``{name_in_translatable: source_to_use}``.
ignore_names: Names **not** to translate, or a predicate ``(NameType) -> bool``.
override_function: A callable ``(name, sources, context) -> Source | None``, where `context` is always
``None``. See :meth:`Mapper.apply <id_translation.mapping.Mapper.apply>` for details. Ignored when
`translatable` is given.
max_fails: The maximum fraction of IDs for which translation may fail. 1=disabled. Missing IDs (e.g.
``NaN``) in ``pandas`` types are pass-throughs and never count; a ``None`` in a builtin collection is
an ordinary (unknown) ID, and does.
fmt: A :class:`format string <id_translation.offline.Format>` such as **'{id}:{name}'** to use. Default is
:attr:`Translator.fmt <id_translation.Translator.fmt>`.
io_kwargs: Keyword arguments for the IO class (e.g.
:class:`~id_translation.dio.integration.pandas.PandasIO`).
Returns:
A :class:`~id_translation.offline.TranslationMap`.
Raises:
~id_translation.exceptions.ConnectionStatusError: If disconnected from the fetcher, i.e. not
:attr:`~id_translation.Translator.online`.
ValueError: If `io_kwargs` is given without `translatable`.
Examples:
Using the returned :class:`~id_translation.offline.TranslationMap` class.
..
# Hidden setup code
>>> from id_translation.fetching import MemoryFetcher
>>> translation_data = {
... "animals": {"id": [0, 1, 2], "name": ["Tarzan", "Morris", "Simba"]},
... "people": {"id": [1999, 1991], "name": ["Sofia", "Richard"]},
... }
>>> translator = Translator(MemoryFetcher(translation_data))
>>> translation_map = translator.fetch()
>>> translation_map
TranslationMap('animals': 3 IDs, 'people': 2 IDs)
**Convert to finished translations.**
* :meth:`TranslationMap.to_translations <id_translation.offline.TranslationMap.to_translations>` → ``{source: MagicDict}``, where a
:class:`~id_translation.offline.MagicDict` is similar to a regular ``dict[IdType, str]``-type dict.
>>> people = translation_map.to_translations()["people"]
>>> people
{1999: '1999:Sofia', 1991: '1991:Richard'}
.. warning::
The :class:`~id_translation.offline.MagicDict` class is used internally and has a few important
differences from the built-in type. Please refer to the :class:`~id_translation.offline.MagicDict` class
documentation for details.
To convert a :class:`~id_translation.offline.MagicDict` to a regular ``dict``, simply use the dict
constructor:
>>> dict(people)
{1999: '1999:Sofia', 1991: '1991:Richard'}
**Convert to raw translation data.**
* :meth:`TranslationMap.to_pandas <id_translation.offline.TranslationMap.to_pandas>` → ``{source: DataFrame}``
* :meth:`TranslationMap.to_dicts <id_translation.offline.TranslationMap.to_dicts>` → ``{source: {placeholder: [values...]}}``
>>> translation_map.to_dicts()["people"]
{'id': [1999, 1991], 'name': ['Sofia', 'Richard']}
"""
task_id = _logging.generate_task_id()
self.initialize_sources(task_id)
return self._user_fetch(
translatable,
names,
ignore_names=ignore_names,
override_function=override_function,
max_fails=max_fails,
fmt=fmt,
io_kwargs=io_kwargs,
func=self.fetch.__qualname__,
task_id=task_id,
)
def _user_fetch(
self,
translatable: Translatable[NameType, IdType] | None = None,
names: NameTypes[NameType] | NameToSource[NameType, SourceType] | None = None,
*,
ignore_names: Names[NameType] | None = None,
override_function: UserOverrideFunction[NameType, SourceType, None] | None = None,
max_fails: float = 1.0,
fmt: FormatType | None = None,
io_kwargs: Mapping[str, Any] | None = None,
func: str,
task_id: int | None = None,
) -> TranslationMap[NameType, SourceType, IdType]:
if io_kwargs and translatable is None:
raise ValueError(f"Got {io_kwargs=}, but {translatable=}; io_kwargs requires a translatable.")
fmt = self._fmt if fmt is None else Format.parse(fmt)
if translatable is None:
if all(p is None for p in (names, ignore_names, override_function)):
source_translations = self._fetch(None, task_id=task_id)
else:
dummy = {source: None for source in self.sources}
sources = self.map(dummy, names, ignore_names=ignore_names, override_function=override_function)
source_translations = self._fetch(set(sources.values()), task_id=task_id)
translation_map = self._to_translation_map(source_translations, fmt=fmt)
if names is None:
pass # Callers must perform mapping unless name=source.
elif isinstance(names, dict):
translation_map.name_to_source = names
else:
translation_map.name_to_source = self.mapper.apply(
as_list(names),
translation_map.sources,
override_function=override_function,
task_id=task_id,
).flatten()
else:
task = TranslationTask(
self,
translatable,
fmt,
names,
ignore_names=ignore_names,
max_fails=max_fails,
enable_uuid_heuristics=self._enable_uuid_heuristics,
io_kwargs=io_kwargs,
event_key=func,
task_id=task_id,
)
translation_map = self._get_updated_tmap(task, force_fetch=True)
task.verify(translation_map)
return translation_map
def _get_updated_tmap(
self,
task: TranslationTask[NameType, SourceType, IdType],
force_fetch: bool = False,
) -> TranslationMap[NameType, SourceType, IdType]:
"""Get an updated translation map."""
if not task.name_to_source:
source_translations = {} # Nothing to translate.
elif force_fetch or self._cached_tmap is None:
source_translations = self._execute_fetch(task)
else:
source_translations = self._cached_tmap._extract_translations()
return self._to_translation_map(source_translations, task.fmt, task.name_to_source)
@property
def transformers(self) -> Transformers[SourceType, IdType]:
"""Get a dict ``{source: transformer}``.
The ``Translator`` is the sole owner of transformers; see
:meth:`~id_translation.Translator.register_transformer` to add them.
.. note::
Mutating the returned dict bypasses chain normalization and is **deprecated**; this will become a read-only
view in ``2.0.0``. Use :meth:`~id_translation.Translator.register_transformer` instead.
Returns:
A dict with one transformer per source. Chains are a single
:class:`~id_translation.transform.TransformerStack`.
"""
# TODO(2.0.0): Return MappingProxyType(self._transformers); returning the live dict is a 1.x compatibility
# concession, since callers may hold it or mutate it.
return self._transformers
def _register_transformer(
self,
source: SourceType,
transformer: Transformer[IdType] | Sequence[Transformer[IdType]],
on_existing: OnExistingTransformer,
) -> None:
# The implementation of `register_transformer`, which subclasses may override. `__init__` calls this instead,
# since an override is entitled to touch attributes that do not exist until construction finishes.
on_existing = ON_EXISTING_HELPER.check(on_existing)
new = as_transformer(transformer)
registered = self._transformers.get(source)
if registered is not None:
if on_existing == "append":
new = TransformerStack(registered, new) # May warn about a duplicate -- before anything is committed.
elif on_existing == "raise":
exc = TransformerConflictError(f"Transformer {registered!r} is already registered for {source!r}.")
exc.add_note(_ON_EXISTING_CONFLICT_HINT)
raise exc
self._transformers[source] = new
self._unverified_sources.append(source)
if self._cached_tmap is not None and not self.online:
# Offline sources are fixed and already known, so verify immediately; online instances are verified by
# initialize_sources(). Both are None while __init__ registers, where the sources are not yet reachable --
# inline-data registrations are verified at the end of __init__ instead, and a snapshot's never are.
try:
self._warn_on_transformers_for_unknown_sources()
except Exception:
# Promoted to an error (e.g. under `-W error`): the caller sees a raise, so the registration must not
# stick -- retrying the apparently-failed call would otherwise conflict with it.
if registered is None:
del self._transformers[source]
else:
self._transformers[source] = registered
self._unverified_sources.pop() # Ours is last: a raised check never drains the queue.
raise
# The cache's appliers captured their transformers when it was built; keep them in sync so that
# `Translator.cache` and `translate()` agree about the registration.
self._cached_tmap._set_transformer(source, new)
def _apply_fetcher_transformers(self) -> None:
if not self._transformers_queried:
# Every pass is derived from current truth (fetcher answer, registrations, declared intent); a failed
# pass mutates nothing and the next one simply asks again. No cached answer exists to go stale.
self._register_transformers_from_fetcher(self._collect_provided())
self._transformers_queried = True
self._warn_on_transformers_for_unknown_sources()
def _collect_provided(self) -> Transformers[SourceType, IdType]:
"""Query every source, applying nothing: a raising fetcher must leave the instance untouched."""
provided: Transformers[SourceType, IdType] = {}
for source in self.sources:
try:
transformer = self.fetcher.get_transformer(source)
if transformer is not None:
provided[source] = as_transformer(transformer) # A bad answer raises before anything registers.
except Exception as e:
e.add_note(f" - Raised by Fetcher.get_transformer({source!r}) on fetcher={self.fetcher}.")
raise
return provided
def _register_transformers_from_fetcher(self, transformers: Transformers[SourceType, IdType]) -> None:
backup = self._transformers.copy()
mark = len(self._unverified_sources) # Append-only below, so this rewinds the queue to where it started.
try:
for source, provided in transformers.items():
registered = self._transformers.get(source)
if registered is None:
chain = provided
else:
# Member-wise: the answer may be a sequence, of which only part is already registered.
members = provided.transformers if isinstance(provided, TransformerStack) else (provided,)
fresh = [it for it in members if not has_member(registered, it)]
if not fresh:
continue
chain = as_transformer([*fresh, registered])
self.register_transformer(source, chain, on_existing="overwrite")
except Exception as e:
# An overridden `register_transformer`, or a promoted duplicate warning out of `as_transformer`.
# Roll back either way, preserving the dict identity `Translator.transformers` hands out.
self._transformers.clear()
self._transformers.update(backup)
del self._unverified_sources[mark:]
e.add_note(f" - Raised while registering a transformer provided by fetcher={self.fetcher}.")
raise
def _warn_on_transformers_for_unknown_sources(self) -> None:
# This runs per translate, so do no work unless something has been registered since the last check.
if not self._unverified_sources:
return
# Nothing below may drain the queue early: neither a discovery that raises nor a warning that `-W error`
# promotes may pass a registration off as checked, since the misconfiguration would then vanish on the
# next call.
sources = self.sources
# The TestFetcher installed when no fetcher is given reports none, but serves every source. A fetcher that
# reports none because it has none is the case where every registration is unreachable.
serves_everything = not sources and isinstance(self._fetcher, TestFetcher)
if not serves_everything:
known = set(sources) # Hoisted: the `if` below is evaluated once per queued registration.
# Deduplicated in registration order: `on_existing='append'` queues the same source once per call.
if unknown := list(dict.fromkeys(s for s in self._unverified_sources if s not in known)):
msg = (
f"Transformers registered for unknown sources: {unknown}. These will never be applied."
f"\nHint: Known sources: {sources}."
)
if unresolved := self._unresolved_children():
# A `[transform]`-section outlives a fetcher discarded during discovery, so these may be its
# sources rather than a typo. Cannot be told apart: a child that raised never reported what it
# serves.
msg += (
f"\nNote: source discovery raised for {len(unresolved)} optional fetcher(s), so what they"
f" would have served is unknown: {unresolved}."
)
emit_warning(msg)
self._unverified_sources.clear()
def _unresolved_children(self) -> list[str]:
return self._fetcher._unresolved_children() if isinstance(self._fetcher, MultiFetcher) else []
def _execute_fetch(
self,
task: TranslationTask[NameType, SourceType, IdType],
) -> SourcePlaceholderTranslations[SourceType]:
start = perf_counter()
ids_to_fetch = [IdsToFetch(source, ids=ids) for source, ids in task.extract_ids().items()]
for itf in ids_to_fetch:
transformer = self._transformers.get(itf.source)
if transformer is not None:
transformer.update_ids(itf.ids)
source_translations = self._fetch(ids_to_fetch, fmt=task.fmt, task_id=task.task_id)
task.add_timing("fetch", perf_counter() - start)
return source_translations
def _fetch(
self,
ids_or_sources: list[IdsToFetch[SourceType, IdType]] | set[SourceType] | None,
fmt: Format | None = None,
task_id: int | None = None,
) -> SourcePlaceholderTranslations[SourceType]:
fmt = fmt or self._fmt
placeholders = fmt.placeholders
placeholder_attributes = fmt.placeholder_attributes
required = fmt.required_placeholders
if self._default_fmt and ID in self._default_fmt.placeholders and ID not in placeholders:
# Ensure that default translations can always use the ID
placeholders = (*placeholders, ID)
required = (*required, ID)
if task_id is None:
task_id = _logging.generate_task_id()
if ids_or_sources is None or isinstance(ids_or_sources, set):
return self.fetcher.fetch_all(
placeholders,
required=required,
placeholder_attributes=placeholder_attributes,
sources=ids_or_sources,
task_id=task_id,
enable_uuid_heuristics=self._enable_uuid_heuristics,
)
else:
return self.fetcher.fetch(
ids_or_sources,
placeholders,
required=required,
placeholder_attributes=placeholder_attributes,
task_id=task_id,
enable_uuid_heuristics=self._enable_uuid_heuristics,
)
def _to_translation_map(
self,
source_translations: SourcePlaceholderTranslations[SourceType],
fmt: Format | None = None,
name_to_source: NameToSource[NameType, SourceType] | None = None,
) -> TranslationMap[NameType, SourceType, IdType]:
return TranslationMap(
source_translations,
fmt=fmt or self._fmt,
default_fmt=self._default_fmt,
name_to_source=name_to_source,
default_fmt_placeholders=self._default_fmt_placeholders,
enable_uuid_heuristics=self._enable_uuid_heuristics,
transformers=self._transformers,
)
def __repr__(self) -> str:
more = f"fetcher={self.fetcher}" if self.online else f"cache={self.cache}"
online = self.online
return f"{tname(self)}({online=}: {more})"
def _handle_default(
default_fmt: FormatType,
default_fmt_placeholders: MakeType[SourceType, str, Any] | None,
) -> tuple[InheritedKeysDict[SourceType, str, Any], Format]:
default_fmt = Format.parse(default_fmt)
if not default_fmt_placeholders:
return InheritedKeysDict(), default_fmt
if isinstance(default_fmt_placeholders, InheritedKeysDict):
default_placeholders = default_fmt_placeholders
else:
default_placeholders = InheritedKeysDict.make(default_fmt_placeholders)
return default_placeholders, default_fmt
