How to use the typing.NamedTuple 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 / typing / src / test_typing.py View on Github external
def test_namedtuple_errors(self):
        with self.assertRaises(TypeError):
            NamedTuple.__new__()
        with self.assertRaises(TypeError):
            NamedTuple()
        with self.assertRaises(TypeError):
            NamedTuple('Emp', [('name', str)], None)
        with self.assertRaises(ValueError):
            NamedTuple('Emp', [('_name', str)])

        with self.assertWarns(DeprecationWarning):
            Emp = NamedTuple(typename='Emp', name=str, id=int)
        self.assertEqual(Emp.__name__, 'Emp')
        self.assertEqual(Emp._fields, ('name', 'id'))

        with self.assertWarns(DeprecationWarning):
            Emp = NamedTuple('Emp', fields=[('name', str), ('id', int)])
        self.assertEqual(Emp.__name__, 'Emp')
        self.assertEqual(Emp._fields, ('name', 'id'))
github Azkae / caribou / caribou / models.py View on Github external
from typing import NamedTuple, Callable, Union, List as TList


class Choice(NamedTuple):
    options: TList[str]

    def process_value(self, value):
        return value


class List(NamedTuple):
    separator: str = ','

    def process_value(self, value):
        if value.strip() == '':
            return []
        return value.split(',')


class Request(NamedTuple):
github facebookresearch / ReAgent / ml / rl / json_serialize.py View on Github external
def prepare_for_json(o: Any) -> Any:
    if isinstance(o, NamedTuple):
        d = {}
        for field_name in o._fields:
            d[field_name] = prepare_for_json(getattr(o, field_name))
        return d
    elif is_dataclass(o):
        return asdict(o)
    else:
        return o
github CoinAlpha / hummingbot / hummingbot / core / event / events.py View on Github external
gas_used: int
    eth_amount: float
    eth_amount_raw: int


class TokenApprovedEvent(NamedTuple):
    timestamp: float
    tx_hash: str
    owner_address: str
    spender_address: str
    asset_name: str
    amount: float
    raw_amount: int


class TradeFee(NamedTuple):
    percent: float # 0.1 = 10%
    flat_fees: List[Tuple[str, float]] = [] # list of (symbol, amount) ie: ("ETH", 0.05)

    @classmethod
    def to_json(cls, trade_fee: "TradeFee") -> Dict[str, any]:
        return {
            "percent": trade_fee.percent,
            "flat_fees": [{"symbol": symbol, "amount": amount}
                          for symbol, amount in trade_fee.flat_fees]
        }

    @classmethod
    def from_json(cls, data: Dict[str, any]) -> "TradeFee":
        return TradeFee(
            data["percent"],
            [(fee_entry["symbol"], fee_entry["amount"])
github magma / magma / lte / gateway / python / magma / pipelined / ng_manager / session_state_manager.py View on Github external
CauseIE,
    OffendingIE,
    PdrState,
    UPFSessionContextState,
)
from lte.protos.session_manager_pb2 import (
    UPFSessionConfigState,
    UPFSessionState,
)
from magma.pipelined.ng_manager.session_state_manager_util import (
    pdr_create_rule_entry,
)
from magma.pipelined.set_interface_client import send_periodic_session_update

# Help to build failure report
MsgParseOutput = NamedTuple(
                   'MsgParseOutput',
                   [('offending_ie', OffendingIE),
                    ('cause_info', int)])

class SessionMessageType(Enum):
    MSG_TYPE_CONTEXT_STATE = 1  # In response to session configuraiton from SMF
    MSG_TYPE_CONFIG_STATE  = 2  # Periodic messages to update session information to SMF
    MSG_TYPE_CONFIG_REPORT = 3  # Event or Periodic Report from session to SMF

class SessionStateManager:
    send_message_offset = 0
    periodic_config_msg_count = 0

    """
    This controller manages session state information
    and reports session config to SMF.
github karlicoss / my / my / media / imdb.py View on Github external
#!/usr/bin/env python3
import csv
import json
from datetime import datetime
from typing import Iterator, List, NamedTuple

from ..common import get_files

from my.config import imdb as config

def _get_last():
    return max(get_files(config.export_path, glob='*.csv'))


class Movie(NamedTuple):
    created: datetime
    title: str
    rating: int


def iter_movies() -> Iterator[Movie]:
    last = _get_last()

    with last.open() as fo:
        reader = csv.DictReader(fo)
        for i, line in enumerate(reader):
            # TODO extract directors??
            title = line['Title']
            rating = int(line['You rated'])
            createds = line['created']
            created = datetime.strptime(createds, '%a %b %d %H:%M:%S %Y')
github alliefitter / boto3_type_annotations / build_scripts / structures.py View on Github external
methods: List[Method]
    attributes: List[Attribute]
    collections: List[Collection]


class Waiter(NamedTuple):
    name: str
    methods: List[Method]


class Paginator(NamedTuple):
    name: str
    methods: List[Method]


class ServiceResource(NamedTuple):
    name: str
    methods: List[Method]
    attributes: List[Attribute]
    collections: List[Collection]
    sub_resources: List[Resource]


class Client(NamedTuple):
    name: str
    methods: List[Method]


class ServiceWaiter(NamedTuple):
    name: str
    waiters: List[Waiter]
github magma / magma / symphony / cli / pyinventory / consts.py View on Github external
class EquipmentPort(NamedTuple):
    """
    Attributes:
        id (str): equipment port ID
        properties (List[Dict[str, PropertyValue]]): list of equipment port properties
        definition (pyinventory.consts.EquipmentPortDefinition): port definition
        link (Optional[pyinventory.consts.Link]): link
    """

    id: str
    properties: Sequence[PropertyFragment]
    definition: EquipmentPortDefinition
    link: Optional[Link]


class SiteSurvey(NamedTuple):
    name: str
    id: str
    completionTime: datetime
    sourceFileId: Optional[str]
    sourceFileName: Optional[str]
    sourceFileKey: Optional[str]
    forms: Dict[str, Dict[str, Any]]


class ServiceType(NamedTuple):
    name: str
    id: str
    hasCustomer: bool
    property_types: Sequence[PropertyTypeFragment]
github josiah-wolf-oberholtzer / supriya / supriya / clock.py View on Github external
SECONDS = 1
    MEASURES = 2


class CallbackCommand(NamedTuple):
    args: Optional[Tuple]
    event_id: int
    event_type: int
    kwargs: Optional[Dict]
    procedure: Callable
    quantization: Optional[str]
    schedule_at: float
    time_unit: Optional[int]


class CallbackEvent(NamedTuple):
    seconds: float
    event_type: int
    event_id: int
    measure: Optional[int]
    offset: Optional[float]
    procedure: Callable
    args: Optional[Tuple]
    kwargs: Optional[Dict]
    invocations: int

    def __hash__(self):
        return hash((type(self), self.event_id))


class ChangeCommand(NamedTuple):
    beats_per_minute: Optional[float]
github nosix / PyMineHub / src / pyminehub / mcpe / value.py View on Github external
'Block',
    'PlacedBlock',
]


PlayerID = UUID
EntityUniqueID = int
EntityRuntimeID = int

GameRule = NamedTuple('GameRule', [
    ('name', str),
    ('type', GameRuleType),
    ('value', Union[bool, int, float])
])

Attribute = NamedTuple('Attribute', [
    ('min', float),
    ('max', float),
    ('current', float),
    ('default', float),
    ('name', str)
])


class AdventureSettings(NamedTuple('AdventureSettings', [
    ('flags', int),
    ('flags2', int)
])):
    @staticmethod
    def create(**kwargs: bool) -> 'AdventureSettings':
        """
        >>> AdventureSettings.create(world_immutable=True, attack_players=True, flying=True, teleport=True)