How to use the typing.Union function in typing

To help you get started, we’ve selected a few typing examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github python / mypy / lib-typing / 3.2 / test_typing.py View on Github external
def test_cannot_instantiate(self):
        with self.assertRaises(TypeError):
            Union()
        with self.assertRaises(TypeError):
            type(Union)()
        u = Union[int, float]
        with self.assertRaises(TypeError):
            u()
        with self.assertRaises(TypeError):
            type(u)()
github ibis-project / ibis / ibis / expr / window.py View on Github external
import numpy as np
import pandas as pd

import ibis.common.exceptions as com
import ibis.expr.operations as ops
import ibis.expr.types as ir
import ibis.util as util


def _sequence_to_tuple(x):
    return tuple(x) if util.is_iterable(x) else x


RowsWithMaxLookback = NamedTuple(
    'RowsWithMaxLookback',
    [('rows', Union[int, np.integer]), ('max_lookback', ir.IntervalValue)],
)


def _choose_non_empty_val(first, second):
    if isinstance(first, (int, np.integer)) and first:
        non_empty_value = first
    elif not isinstance(first, (int, np.integer)) and first is not None:
        non_empty_value = first
    else:
        non_empty_value = second
    return non_empty_value


def _determine_how(preceding):
    offset_type = type(_get_preceding_value(preceding))
    if issubclass(offset_type, (int, np.integer)):
github prompt-toolkit / python-prompt-toolkit / prompt_toolkit / eventloop / async_generator.py View on Github external
async def generator_to_async_generator(
    get_iterable: Callable[[], Iterable[_T]]
) -> AsyncGenerator[_T, None]:
    """
    Turn a generator or iterable into an async generator.

    This works by running the generator in a background thread.

    :param get_iterable: Function that returns a generator or iterable when
        called.
    """
    quitting = False
    _done = _Done()
    q: Queue[Union[_T, _Done]] = Queue()
    loop = get_event_loop()

    def runner() -> None:
        """
        Consume the generator in background thread.
        When items are received, they'll be pushed to the queue.
        """
        try:
            for item in get_iterable():
                loop.call_soon_threadsafe(q.put_nowait, item)

                # When this async generator was cancelled (closed), stop this
                # thread.
                if quitting:
                    break
github schlegelp / navis / navis / plotting / colors.py View on Github external
from typing import Union, List, Tuple, Optional, Dict, Any, Sequence, overload
from typing_extensions import Literal

from .. import core, config, utils

__all__ = ['generate_colors', 'prepare_connector_cmap', 'prepare_colormap',
           'eval_color', 'hex_to_rgb', 'vary_colors']

logger = config.logger

# Some definitions for mypy
RGB_color = Tuple[float, float, float]
RGBA_color = Tuple[float, float, float, float]
Str_color = str
ColorList = Sequence[Union[RGB_color, RGBA_color, Str_color]]
AnyColor = Union[RGB_color, RGBA_color, Str_color, ColorList]

def generate_colors(N: int,
                    color_space: Union[Literal['RGB'],
                                       Literal['Grayscale']] = 'RGB',
                    color_range: Union[Literal[1],
                                       Literal[255]] = 1
                    ) -> List[Tuple[float, float, float]]:
    """Divide colorspace into N evenly distributed colors.

    Returns
    -------
    colormap :  list
                [(r, g, b), (r, g, b), ...]

    """
    if N == 1:
github rosswhitfield / ase / ase / io / formats.py View on Github external
else:
        fd = open(filename, mode)

    return fd


def wrap_read_function(read, filename, index=None, **kwargs):
    """Convert read-function to generator."""
    if index is None:
        yield read(filename, **kwargs)
    else:
        for atoms in read(filename, index, **kwargs):
            yield atoms


NameOrFile = Union[str, PurePath, IO]


def write(
        filename: NameOrFile,
        images: Union[Atoms, Sequence[Atoms]],
        format: str = None,
        parallel: bool = True,
        append: bool = False,
        **kwargs: dict
) -> None:
    """Write Atoms object(s) to file.

    filename: str or file
        Name of the file to write to or a file descriptor.  The name '-'
        means standard output.
    images: Atoms object or list of Atoms objects
github qutech / qupulse / qupulse / _program / tabor.py View on Github external
else:
                    raise TaborException('The algorithm is not smart enough to make this sequence table longer')
            elif _check_partial_unroll(program, i, min_seq_len=min_seq_len):
                i += 1
            else:
                raise TaborException('The algorithm is not smart enough to make this sequence table longer')
        else:
            i += 1


ParsedProgram = NamedTuple('ParsedProgram', [('advanced_sequencer_table', Sequence[TableEntry]),
                                             ('sequencer_tables', Sequence[Sequence[
                                                     Tuple[TableDescription, Optional[VolatileProperty]]]]),
                                             ('waveforms', Tuple[Waveform, ...]),
                                             ('volatile_parameter_positions', Dict[Union[int, Tuple[int, int]],
                                                                                   VolatileRepetitionCount])])


def parse_aseq_program(program: Loop, used_channels: FrozenSet[ChannelID]) -> ParsedProgram:
    volatile_parameter_positions = {}

    advanced_sequencer_table = []
    # we use an ordered dict here to avoid O(n**2) behaviour while looking for duplicates
    sequencer_tables = OrderedDict()
    waveforms = OrderedDict()
    for adv_position, sequencer_table_loop in enumerate(program):
        current_sequencer_table = []
        for position, (waveform, repetition_definition, volatile_repetition) in enumerate(
                (waveform_loop.waveform.get_subset_for_channels(used_channels),
                 waveform_loop.repetition_definition, waveform_loop.volatile_repetition)
                for waveform_loop in cast(Sequence[Loop], sequencer_table_loop)):
github ethereum / py-trie / trie / typing.py View on Github external
from typing_extensions import (
    Literal,
    Protocol,
)

from eth_utils import (
    is_list_like,
)


# The RLP-decoded node is either blank, or a list, full of bytes or recursive nodes
# Recursive definitions don't seem supported at the moment, follow:
#   https://github.com/python/mypy/issues/731
# Another option is to manually declare a few levels of the type. It should be possible
#   to determine the maximum number of embeds with single-nibble keys and single byte values.
RawHexaryNode = Union[
    # Blank node
    Literal[b''],

    # Leaf or extension node are length 2
    # Branch node is length 17
    List[Union[
        # keys, hashes to next nodes, and values
        bytes,

        # embedded subnodes
        "RawHexaryNode",
    ]],
]


class Nibble(enum.IntEnum):
github IEMLdev / ieml / ieml / lexicon / grammar / usl.py View on Github external
return IEMLSyntaxType.__ge__(self, other) and self != other

    def __le__(self, other):
        return IEMLSyntaxType.__lt__(self, other) and self == other

    def __ge__(self, other):
        return not IEMLSyntaxType.__lt__(self, other)

    def __lt__(self, other):
        return self.__rank < other.__rank

    def syntax_rank(self):
        return self.__rank


Literal = Union[List[str], str]


class Usl(metaclass=IEMLSyntaxType):
    def __init__(self, literals: Literal=None):
        super().__init__()
        self._paths = None

        _literals = []
        if literals is not None:
            if isinstance(literals, str):
                _literals = [literals]
            else:
                try:
                    _literals = tuple(literals)
                except TypeError:
                    raise InvalidIEMLObjectArgument(self.__class__, "The literals argument %s must be an iterable of "
github prompt-toolkit / python-prompt-toolkit / prompt_toolkit / key_binding / key_bindings.py View on Github external
"""
        return []

    @abstractproperty
    def bindings(self) -> List[Binding]:
        """
        List of `Binding` objects.
        (These need to be exposed, so that `KeyBindings` objects can be merged
        together.)
        """
        return []

    # `add` and `remove` don't have to be part of this interface.


T = TypeVar("T", bound=Union[KeyHandlerCallable, Binding])


class KeyBindings(KeyBindingsBase):
    """
    A container for a set of key bindings.

    Example usage::

        kb = KeyBindings()

        @kb.add('c-t')
        def _(event):
            print('Control-T pressed')

        @kb.add('c-a', 'c-b')
        def _(event):
github emilkarlen / exactly / src / exactly_lib / util / textformat / utils.py View on Github external
from typing import List, Union, Tuple

from exactly_lib.util.textformat.structure import document as doc
from exactly_lib.util.textformat.structure import lists, table, structures as docs
from exactly_lib.util.textformat.structure.core import ParagraphItem
from exactly_lib.util.textformat.structure.paragraph import Paragraph
from exactly_lib.util.textformat.structure.structures import cell

InitialParagraphsOrSectionContents = Union[doc.SectionContents, List[ParagraphItem]]


def append_section_if_contents_is_non_empty(
        output_list_of_sections: List[doc.SectionItem],
        section_title_str_or_text,
        initial_paragraphs_or_section_contents: InitialParagraphsOrSectionContents):
    section_contents = initial_paragraphs_or_section_contents
    if isinstance(initial_paragraphs_or_section_contents, list):
        section_contents = doc.SectionContents(initial_paragraphs_or_section_contents)
    if not section_contents.is_empty:
        output_list_of_sections.append(doc.Section(docs.text_from_unknown(section_title_str_or_text),
                                                   section_contents))


def append_sections_if_contents_is_non_empty(
        output_list_of_sections: List[doc.SectionItem],