Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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'))
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):
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
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"])
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.
#!/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')
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]
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]
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]
'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)