Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"bob": {"name": "Bob", "age": 10},
"clarice": {"name": "Clarice", "age": 10},
"jeremy": None,
},
}
assert ParentWithOptionalChildren.loads(nancy.dumps()) == nancy
@dataclass
class ChildWithFriends(Child):
friends: List[Optional[Child]] = mutable_field(list)
@dataclass
class ParentWithOptionalChildrenWithFriends(Serializable):
name: str = "Consuela"
children: Mapping[str, Optional[ChildWithFriends]] = mutable_field(OrderedDict)
def test_lists(silent):
bob = ChildWithFriends("Bob")
clarice = Child("Clarice")
bob.friends.append(clarice)
bob.friends.append(None)
nancy = ParentWithOptionalChildrenWithFriends("Nancy", children=dict(bob=bob))
nancy.children["jeremy"] = None
assert nancy.to_dict() == {
"name": "Nancy",
"children": {
"bob": {"name": "Bob", "age": 10, "friends": [
def test_weakref_all(self):
T = TypeVar('T')
things = [Any, Union[T, int], Callable[..., T], Tuple[Any, Any],
Optional[List[int]], typing.Mapping[int, str],
typing.re.Match[bytes], typing.Iterable['whatever']]
for t in things:
self.assertEqual(weakref.ref(t)(), t)
def from_subtypes(
cls: typing.Type[BaseTuple_T],
schema: s_schema.Schema,
subtypes: Union[Iterable[Type], Mapping[str, Type]],
typemods: Any = None,
*,
name: Optional[s_name.Name] = None,
id: Union[uuid.UUID, so.NoDefaultT] = so.NoDefault,
**kwargs,
) -> BaseTuple_T:
named = False
if typemods is not None:
named = typemods.get('named', False)
types: Mapping[str, Type]
if isinstance(subtypes, collections.abc.Mapping):
types = subtypes
else:
types = {str(i): type for i, type in enumerate(subtypes)}
return cls.create(schema, element_types=types, named=named,
name=name, id=id, **kwargs)
return res
def _condemned_reason_group_of_bool_condemned_reason(
b: BoolCondemnedReason
) -> _CondemnedReasonGroup:
res = _CondemnedReasonGroup(enum_value(b) + enum_count(GenCondemnedReason))
assert res.name == b.name
return res
@with_slots
@dataclass(frozen=True)
class CondemnedReasonsForHeap:
# Only has non-Gen0 entries
gen_reasons: Mapping[GenCondemnedReason, Gens]
bool_reasons: FrozenSet[BoolCondemnedReason]
def is_empty(self) -> bool:
return is_empty(self.gen_reasons) and is_empty(self.bool_reasons)
def get_gen(self) -> Gens:
return self.gen_reasons.get(GenCondemnedReason.final_generation, Gens.Gen0)
@property
def initial_generation(self) -> Optional[Gens]:
return self.gen_reasons.get(GenCondemnedReason.initial_generation)
@property
def final_generation(self) -> Optional[Gens]:
return self.gen_reasons[GenCondemnedReason.final_generation]
def traverse_all(obj: Any) -> Iterable[Any]:
"""Iterate over all nested mappings and sequences
Parameters
----------
obj: Any
Returns
-------
Iterable[Any]
Iterable of child values to obj
"""
if isinstance(obj, Mapping):
for key, value in obj.items():
yield from traverse_all(value)
elif isinstance(obj, Iterable) and not isinstance(obj, str):
for value in obj:
yield from traverse_all(value)
else:
yield obj
def iter(self, value):
# type: (Iterable) -> Generator[Any]
"""
Iterator version of :py:meth:`apply`.
"""
if value is not None:
if isinstance(value, Mapping):
for k, v in iteritems(value):
u_key = self.unicodify_key(k)
if (
(self.restrict_keys is None)
or (k in self.restrict_keys)
):
yield k, self._apply_item(u_key, v, self._filter_chain)
else:
# For consistency with FilterMapper, invalid
# keys are not included in the filtered
# value (hence this statement does not
# ``yield``).
self._invalid_value(
value = v,
reason = self.CODE_EXTRA_KEY,
'RegistryKeyError',
'SortedClassRegistry',
]
class RegistryKeyError(KeyError):
"""
Used to differentiate a registry lookup from a standard KeyError.
This is especially useful when a registry class expects to extract
values from dicts to generate keys.
"""
pass
class BaseRegistry(typing.Mapping, metaclass=ABCMeta):
"""
Base functionality for registries.
"""
def __contains__(self, key):
"""
Returns whether the specified key is registered.
"""
try:
# Use :py:meth:`get_class` instead of :py:meth:`__getitem__`
# in case the registered class' initializer requires
# arguments.
self.get_class(key)
except RegistryKeyError:
return False
else:
self._feature = _
def register(self, feature: 'Feature'):
feature.services.append(self)
def setup_env(self, feature: 'Feature') -> None:
pass
def exit_env(self):
pass
def __call__(self, elem):
return self.get_dispatch(elem)
class CompilingTimeContext(typing.Mapping):
def __len__(self) -> int:
raise NotImplemented
def __iter__(self):
raise NotImplemented
def __init__(self, globals: dict, closure: dict):
"""
:param locals: a dictionary made from outside closure.
"""
self.locals = {}
self.closure = closure
self.globals = globals
def __contains__(self, item):
return item in self.closure or item in self.locals or item in self.globals
async def _fetch_headers(self, url: str, headers: Mapping[str, str]) \
-> Optional[Mapping[str, str]]:
"""Do a HEAD/GET to get resources headers."""
requester = self.device.requester
# try a HEAD first
status, headers, _ = await requester.async_http_request('HEAD',
url,
headers=headers,
body_type='ignore')
if 200 <= status < 300:
return headers
# then try a GET
status, headers, _ = await requester.async_http_request('GET',
url,
headers=headers,
body_type='ignore')
# something that another item `provides()`.
#
# By having this phase be last, we also allow removing files added by
# RPM_INSTALL. The downside is that this is a footgun. The upside is
# that e.g. cleaning up yum log & caches can be done as an
# `image_feature` instead of being code. We might also use this to
# remove e.g. unnecessary parts of excessively monolithic RPMs.
REMOVE_PATHS = enum.auto()
class LayerOpts(NamedTuple):
artifacts_may_require_repo: bool
build_appliance: str
layer_target: str
yum_from_snapshot: str
target_to_path: Mapping[str, str]
subvolumes_dir: str
preserve_yum_cache: bool
class ImageItem(type):
'A metaclass for the types of items that can be installed into images.'
def __new__(metacls, classname, bases, dct):
# Future: `deepfrozen` has a less clunky way of doing this
def customize_fields(kwargs):
fn = dct.get('customize_fields')
if fn:
fn(kwargs)
return kwargs
# Some items, like RPM actions, are not sorted by dependencies, but