Skip to content

core

Bad

Bases: Generic[B, G]

The bad side of the disjoint union.

Source code in m/core/fp.py
class Bad(Generic[B, G]):
    """The bad side of the disjoint union."""

    value: B
    is_bad = True

    def __init__(self, value: B):
        """Initialize a `Bad` instance.

        Args:
            value: The "bad" value to store in the instance.
        """
        self.value = value

    def __iter__(self) -> Iterator[G]:
        """Iterate over values of an instance.

        The intention is to raise a `StopBadIteration` exception to
        be able to break out of for loop comprehensions.

        Raises:
            StopBadIteration: If the instance has a "Bad" value.
        """
        raise StopBadIteration(self)

    def iter(self) -> Iterator[G]:
        """Shortcut to transform to a list.

        Can be used as `list(x.iter())`. It will either contain a value or be
        an empty list.

        Yields:
            The value if the instance is a "Good" value.
        """
        empty = ()
        yield from empty

    def get_or_else(self, default: LazyArg[G]) -> G:
        """Return the value if its Good or the given argument if its a Bad.

        Args:
            default: The default value in case the instance is "Bad".

        Returns:
            Either the value or the default specified by "default".
        """
        return lazy_arg(default)

    def map(self, _fct: Callable[[G], K]) -> 'Bad[B, K]':
        """Apply the function to its value if this is a `Good` instance.

        Args:
            _fct: The function to apply to the "Good" value.

        Returns:
            Itself if its a `Bad` otherwise another instance of `Good`.
        """
        return cast(Bad[B, K], self)

    def flat_map_bad(self, fct: Callable[[B], OneOf[B, G]]) -> OneOf[B, G]:
        """Apply the input function if this is a `Bad` value.

        Args:
            fct: The function to apply to the "Bad" value.

        Returns:
            Itself if its a `Good` otherwise another instance of `Bad`.
        """
        return fct(self.value)

__init__(value)

Initialize a Bad instance.

Parameters:

Name Type Description Default
value B

The "bad" value to store in the instance.

required
Source code in m/core/fp.py
def __init__(self, value: B):
    """Initialize a `Bad` instance.

    Args:
        value: The "bad" value to store in the instance.
    """
    self.value = value

__iter__()

Iterate over values of an instance.

The intention is to raise a StopBadIteration exception to be able to break out of for loop comprehensions.

Raises:

Type Description
StopBadIteration

If the instance has a "Bad" value.

Source code in m/core/fp.py
def __iter__(self) -> Iterator[G]:
    """Iterate over values of an instance.

    The intention is to raise a `StopBadIteration` exception to
    be able to break out of for loop comprehensions.

    Raises:
        StopBadIteration: If the instance has a "Bad" value.
    """
    raise StopBadIteration(self)

flat_map_bad(fct)

Apply the input function if this is a Bad value.

Parameters:

Name Type Description Default
fct Callable[[B], OneOf[B, G]]

The function to apply to the "Bad" value.

required

Returns:

Type Description
OneOf[B, G]

Itself if its a Good otherwise another instance of Bad.

Source code in m/core/fp.py
def flat_map_bad(self, fct: Callable[[B], OneOf[B, G]]) -> OneOf[B, G]:
    """Apply the input function if this is a `Bad` value.

    Args:
        fct: The function to apply to the "Bad" value.

    Returns:
        Itself if its a `Good` otherwise another instance of `Bad`.
    """
    return fct(self.value)

get_or_else(default)

Return the value if its Good or the given argument if its a Bad.

Parameters:

Name Type Description Default
default LazyArg[G]

The default value in case the instance is "Bad".

required

Returns:

Type Description
G

Either the value or the default specified by "default".

Source code in m/core/fp.py
def get_or_else(self, default: LazyArg[G]) -> G:
    """Return the value if its Good or the given argument if its a Bad.

    Args:
        default: The default value in case the instance is "Bad".

    Returns:
        Either the value or the default specified by "default".
    """
    return lazy_arg(default)

iter()

Shortcut to transform to a list.

Can be used as list(x.iter()). It will either contain a value or be an empty list.

Yields:

Type Description
G

The value if the instance is a "Good" value.

Source code in m/core/fp.py
def iter(self) -> Iterator[G]:
    """Shortcut to transform to a list.

    Can be used as `list(x.iter())`. It will either contain a value or be
    an empty list.

    Yields:
        The value if the instance is a "Good" value.
    """
    empty = ()
    yield from empty

map(_fct)

Apply the function to its value if this is a Good instance.

Parameters:

Name Type Description Default
_fct Callable[[G], K]

The function to apply to the "Good" value.

required

Returns:

Type Description
'Bad[B, K]'

Itself if its a Bad otherwise another instance of Good.

Source code in m/core/fp.py
def map(self, _fct: Callable[[G], K]) -> 'Bad[B, K]':
    """Apply the function to its value if this is a `Good` instance.

    Args:
        _fct: The function to apply to the "Good" value.

    Returns:
        Itself if its a `Bad` otherwise another instance of `Good`.
    """
    return cast(Bad[B, K], self)

Good

Bases: Generic[B, G]

The good side of the disjoint union.

Source code in m/core/fp.py
class Good(Generic[B, G]):
    """The good side of the disjoint union."""

    value: G
    is_bad = False

    def __init__(self, value: G):
        """Initialize a `Good` instance.

        Args:
            value: The "good" value to store in the instance.
        """
        self.value = value

    def __iter__(self) -> Iterator[G]:
        """Iterate over the value of the instance.

        Yields:
            The value of the instance.
        """
        yield self.value

    def iter(self) -> Iterator[G]:
        """Shortcut to transform to a list.

        Can be used as `list(x.iter())`. It will either contain a value or be
        an empty list.

        Yields:
            The value if the instance is a "Good" value.
        """
        if not self.is_bad:
            yield self.value

    def get_or_else(self, _default: LazyArg[G]) -> G:
        """Return the value.

        Args:
            _default: The default value in case the instance is "Bad".

        Returns:
            Either the value or the default specified by "default".
        """
        return self.value

    def map(self, fct: Callable[[G], K]) -> 'Good[B, K]':
        """Apply the function to its value if this is a `Good` instance.

        Args:
            fct: The function to apply to the "Good" value.

        Returns:
            Itself if its a `Bad` otherwise another instance of `Good`.
        """
        return Good(fct(self.value))

    def flat_map_bad(self, _fct: Callable[[B], OneOf[B, G]]) -> OneOf[B, G]:
        """Apply the input function if this is a `Bad` value.

        Args:
            _fct: The function to apply to the "Bad" value.

        Returns:
            Itself if its a `Good` otherwise another instance of `Bad`.
        """
        return self

__init__(value)

Initialize a Good instance.

Parameters:

Name Type Description Default
value G

The "good" value to store in the instance.

required
Source code in m/core/fp.py
def __init__(self, value: G):
    """Initialize a `Good` instance.

    Args:
        value: The "good" value to store in the instance.
    """
    self.value = value

__iter__()

Iterate over the value of the instance.

Yields:

Type Description
G

The value of the instance.

Source code in m/core/fp.py
def __iter__(self) -> Iterator[G]:
    """Iterate over the value of the instance.

    Yields:
        The value of the instance.
    """
    yield self.value

flat_map_bad(_fct)

Apply the input function if this is a Bad value.

Parameters:

Name Type Description Default
_fct Callable[[B], OneOf[B, G]]

The function to apply to the "Bad" value.

required

Returns:

Type Description
OneOf[B, G]

Itself if its a Good otherwise another instance of Bad.

Source code in m/core/fp.py
def flat_map_bad(self, _fct: Callable[[B], OneOf[B, G]]) -> OneOf[B, G]:
    """Apply the input function if this is a `Bad` value.

    Args:
        _fct: The function to apply to the "Bad" value.

    Returns:
        Itself if its a `Good` otherwise another instance of `Bad`.
    """
    return self

get_or_else(_default)

Return the value.

Parameters:

Name Type Description Default
_default LazyArg[G]

The default value in case the instance is "Bad".

required

Returns:

Type Description
G

Either the value or the default specified by "default".

Source code in m/core/fp.py
def get_or_else(self, _default: LazyArg[G]) -> G:
    """Return the value.

    Args:
        _default: The default value in case the instance is "Bad".

    Returns:
        Either the value or the default specified by "default".
    """
    return self.value

iter()

Shortcut to transform to a list.

Can be used as list(x.iter()). It will either contain a value or be an empty list.

Yields:

Type Description
G

The value if the instance is a "Good" value.

Source code in m/core/fp.py
def iter(self) -> Iterator[G]:
    """Shortcut to transform to a list.

    Can be used as `list(x.iter())`. It will either contain a value or be
    an empty list.

    Yields:
        The value if the instance is a "Good" value.
    """
    if not self.is_bad:
        yield self.value

map(fct)

Apply the function to its value if this is a Good instance.

Parameters:

Name Type Description Default
fct Callable[[G], K]

The function to apply to the "Good" value.

required

Returns:

Type Description
'Good[B, K]'

Itself if its a Bad otherwise another instance of Good.

Source code in m/core/fp.py
def map(self, fct: Callable[[G], K]) -> 'Good[B, K]':
    """Apply the function to its value if this is a `Good` instance.

    Args:
        fct: The function to apply to the "Good" value.

    Returns:
        Itself if its a `Bad` otherwise another instance of `Good`.
    """
    return Good(fct(self.value))

Issue

Bases: Exception

Wrapper to keep track of all exceptions.

It provides a 'cause' field so that we may know why an issue was triggered.

Source code in m/core/issue.py
class Issue(Exception):  # noqa: N818, WPS230 - Intention is not to raise
    """Wrapper to keep track of all exceptions.

    It provides a 'cause' field so that we may know why an issue was
    triggered.
    """

    show_traceback = True
    yaml_traceback = True

    message: str
    description: str | None
    cause: Exception | None
    context: object | None
    include_traceback: bool
    cause_tb: list[str] | None
    traceback: list[str] | None

    def __init__(  # noqa: WPS211 - need to initialize the attributes
        self,
        message: str,
        description: str | None = None,
        cause: Exception | None = None,
        context: object | None = None,
        include_traceback: bool = True,
    ):
        """Create an Issue.

        Args:
            message: Simple description of the issue.
            description: More in depth detail on the issue
            cause: The exception/Issue that is responsible for this instance.
            context: Dictionary with any useful data related to the issue.
            include_traceback: If False, it skips computing the traceback.
        """
        super().__init__()
        self.message = message
        self.description = description
        self.cause = cause
        if cause and not isinstance(cause, Issue):
            # https://stackoverflow.com/a/12539332/788553
            with suppress(BaseException):
                exc_info = sys.exc_info()
                fmt_exception = (
                    traceback.format_exception_only(exc_info[0], exc_info[1])
                    if exc_info[0] is not None and exc_info[1] is not None
                    else []
                )
                exception_list = [
                    *traceback.format_tb(exc_info[2]),
                    *fmt_exception,
                ]
                self.cause_tb = [
                    y
                    for x in exception_list
                    for y in x.splitlines()
                ]
        self.context = context
        self.include_traceback = include_traceback
        self.traceback = None
        if self.include_traceback:
            frame = inspect.currentframe()
            self.traceback = [
                y
                for x in traceback.format_stack(frame)[:-1]
                for y in x.splitlines()
            ]

    def only_context(self) -> bool:
        """Return true if the issue only offers context.

        In some cases we may create an issue with only a message and a
        context. This function will let us know of such case so that a
        log formatter may be able to unwrap the context.

        Returns:
            True if it is safe to only display the context value on a log.
        """
        has_context = self.context is not None
        all_props = (self.description, self.traceback, self.cause)
        return has_context and not [_ for _ in all_props if _]

    def to_dict(self, show_traceback: bool = True) -> IssueDict:
        """Convert to a ordered dictionary.

        This is done so that each of the properties are written in an expected
        order.

        Args:
            show_traceback: If False, it will remove all stacktraces.

        Returns:
            An `IssueDict` instance.
        """
        issue_dict = cast(IssueDict, OrderedDict())
        issue_dict['message'] = self.message
        if self.description:
            issue_dict['description'] = self.description
        if self.context:
            issue_dict['context'] = self.context
        if self.include_traceback and self.traceback:
            issue_dict['traceback'] = self.traceback
            if Issue.yaml_traceback and isinstance(self.traceback, list):
                issue_dict['traceback'] = _traceback_to_str(self.traceback)
        if self.cause:
            self._handle_cause(issue_dict)
        if not show_traceback:
            remove_traceback(issue_dict)
        return issue_dict

    def to_str(self, show_traceback: bool) -> str:
        """Convert the instance to string.

        Args:
            show_traceback: If false, it will remove the error traceback.

        Returns:
            A string representation of the Issue instance.
        """
        issue_dict = self.to_dict(show_traceback=show_traceback)
        if Issue.yaml_traceback:
            return yaml.dumps(issue_dict)
        return json.dumps(issue_dict, indent=2)

    def __str__(self) -> str:
        """Convert the instance to a string.

        Returns:
            A string representation of the Issue instance.
        """
        issue_str = self.to_str(Issue.show_traceback)
        if Issue.yaml_traceback:
            return highlight_yaml(issue_str)
        return highlight_json(issue_str)

    def _handle_cause(self, issue_dict: IssueDict) -> None:
        if isinstance(self.cause, Issue):
            issue_dict['cause'] = self.cause.to_dict()
        else:
            issue_dict['cause'] = {
                'message': str(self.cause),
                'traceback': self.cause_tb,
            }
            # I'm clearly assigning to a dict in the above statement...
            # but it is an object so a dict can be assigned to it but that
            # does not mean that we can use `pop` on an object. So we have to
            # help mypy know that know we are dealing with a dict.
            cause = cast(dict, issue_dict['cause'])
            if not self.cause_tb:
                cause.pop('traceback', None)
            if Issue.yaml_traceback and isinstance(self.cause_tb, list):
                cause['traceback'] = _traceback_to_str(self.cause_tb)

__init__(message, description=None, cause=None, context=None, include_traceback=True)

Create an Issue.

Parameters:

Name Type Description Default
message str

Simple description of the issue.

required
description str | None

More in depth detail on the issue

None
cause Exception | None

The exception/Issue that is responsible for this instance.

None
context object | None

Dictionary with any useful data related to the issue.

None
include_traceback bool

If False, it skips computing the traceback.

True
Source code in m/core/issue.py
def __init__(  # noqa: WPS211 - need to initialize the attributes
    self,
    message: str,
    description: str | None = None,
    cause: Exception | None = None,
    context: object | None = None,
    include_traceback: bool = True,
):
    """Create an Issue.

    Args:
        message: Simple description of the issue.
        description: More in depth detail on the issue
        cause: The exception/Issue that is responsible for this instance.
        context: Dictionary with any useful data related to the issue.
        include_traceback: If False, it skips computing the traceback.
    """
    super().__init__()
    self.message = message
    self.description = description
    self.cause = cause
    if cause and not isinstance(cause, Issue):
        # https://stackoverflow.com/a/12539332/788553
        with suppress(BaseException):
            exc_info = sys.exc_info()
            fmt_exception = (
                traceback.format_exception_only(exc_info[0], exc_info[1])
                if exc_info[0] is not None and exc_info[1] is not None
                else []
            )
            exception_list = [
                *traceback.format_tb(exc_info[2]),
                *fmt_exception,
            ]
            self.cause_tb = [
                y
                for x in exception_list
                for y in x.splitlines()
            ]
    self.context = context
    self.include_traceback = include_traceback
    self.traceback = None
    if self.include_traceback:
        frame = inspect.currentframe()
        self.traceback = [
            y
            for x in traceback.format_stack(frame)[:-1]
            for y in x.splitlines()
        ]

__str__()

Convert the instance to a string.

Returns:

Type Description
str

A string representation of the Issue instance.

Source code in m/core/issue.py
def __str__(self) -> str:
    """Convert the instance to a string.

    Returns:
        A string representation of the Issue instance.
    """
    issue_str = self.to_str(Issue.show_traceback)
    if Issue.yaml_traceback:
        return highlight_yaml(issue_str)
    return highlight_json(issue_str)

only_context()

Return true if the issue only offers context.

In some cases we may create an issue with only a message and a context. This function will let us know of such case so that a log formatter may be able to unwrap the context.

Returns:

Type Description
bool

True if it is safe to only display the context value on a log.

Source code in m/core/issue.py
def only_context(self) -> bool:
    """Return true if the issue only offers context.

    In some cases we may create an issue with only a message and a
    context. This function will let us know of such case so that a
    log formatter may be able to unwrap the context.

    Returns:
        True if it is safe to only display the context value on a log.
    """
    has_context = self.context is not None
    all_props = (self.description, self.traceback, self.cause)
    return has_context and not [_ for _ in all_props if _]

to_dict(show_traceback=True)

Convert to a ordered dictionary.

This is done so that each of the properties are written in an expected order.

Parameters:

Name Type Description Default
show_traceback bool

If False, it will remove all stacktraces.

True

Returns:

Type Description
IssueDict

An IssueDict instance.

Source code in m/core/issue.py
def to_dict(self, show_traceback: bool = True) -> IssueDict:
    """Convert to a ordered dictionary.

    This is done so that each of the properties are written in an expected
    order.

    Args:
        show_traceback: If False, it will remove all stacktraces.

    Returns:
        An `IssueDict` instance.
    """
    issue_dict = cast(IssueDict, OrderedDict())
    issue_dict['message'] = self.message
    if self.description:
        issue_dict['description'] = self.description
    if self.context:
        issue_dict['context'] = self.context
    if self.include_traceback and self.traceback:
        issue_dict['traceback'] = self.traceback
        if Issue.yaml_traceback and isinstance(self.traceback, list):
            issue_dict['traceback'] = _traceback_to_str(self.traceback)
    if self.cause:
        self._handle_cause(issue_dict)
    if not show_traceback:
        remove_traceback(issue_dict)
    return issue_dict

to_str(show_traceback)

Convert the instance to string.

Parameters:

Name Type Description Default
show_traceback bool

If false, it will remove the error traceback.

required

Returns:

Type Description
str

A string representation of the Issue instance.

Source code in m/core/issue.py
def to_str(self, show_traceback: bool) -> str:
    """Convert the instance to string.

    Args:
        show_traceback: If false, it will remove the error traceback.

    Returns:
        A string representation of the Issue instance.
    """
    issue_dict = self.to_dict(show_traceback=show_traceback)
    if Issue.yaml_traceback:
        return yaml.dumps(issue_dict)
    return json.dumps(issue_dict, indent=2)

hone(msg, context=None, description=None, include_traceback=True)

Create a function to repackage the issue with a new message and context.

Parameters:

Name Type Description Default
msg str

The new message.

required
context object | None

The new context.

None
description str | None

The new description.

None
include_traceback bool

Whether to include the traceback.

True

Returns:

Type Description
Callable[[Issue], OneOf[Issue, Any]]

A function that takes an issue and returns a new issue.

Source code in m/core/one_of.py
def hone(
    msg: str,
    context: object | None = None,
    description: str | None = None,
    include_traceback: bool = True,
) -> Callable[[Issue], OneOf[Issue, Any]]:
    """Create a function to repackage the issue with a new message and context.

    Args:
        msg: The new message.
        context: The new context.
        description: The new description.
        include_traceback: Whether to include the traceback.

    Returns:
        A function that takes an issue and returns a new issue.
    """
    return partial(_hone, msg, context, description, include_traceback)

is_bad(inst)

Assert that a OneOf instance is a Bad.

Parameters:

Name Type Description Default
inst OneOf[B, G]

The OneOf instance.

required

Returns:

Type Description
TypeGuard[Bad[B, G]]

True if the instance is a Bad.

Source code in m/core/fp.py
@typing_extensions.deprecated(
    'The `is_bad` type guard is deprecated; use `isinstance(inst, Bad)` instead.',
)
def is_bad(inst: OneOf[B, G]) -> TypeGuard[Bad[B, G]]:
    """Assert that a OneOf instance is a `Bad`.

    Args:
        inst: The `OneOf` instance.

    Returns:
        True if the instance is a `Bad`.
    """
    # `m` does not reference this function anymore, excluding from coverage
    return isinstance(inst, Bad)  # pragma: no cover

non_issue(inst)

Obtain the value of the OneOf as if it was a Good value.

Warning: This should only be used provided that we know for sure that we are not dealing with a Bad value.

Parameters:

Name Type Description Default
inst OneOf[Issue, G]

A OneOf.

required

Returns:

Type Description
G

The value stored in the OneOf.

Source code in m/core/one_of.py
def non_issue(inst: OneOf[Issue, G]) -> G:
    """Obtain the value of the `OneOf` as if it was a Good value.

    Warning: This should only be used provided that we know for sure
    that we are not dealing with a `Bad` value.

    Args:
        inst: A OneOf.

    Returns:
        The value stored in the OneOf.
    """
    # The assert statement can go away with the -O flag.
    assert not inst.is_bad  # noqa: S101
    return cast(G, inst.value)