def __init__(
self, *, cache_sketches: bool=True,
source_encoding: str="utf-8",
custom_escape_fns: Mapping[str, Callable[[Any], str]]={}) -> None:
self._source_encoding = source_encoding
escape_fns = escaping.builtin_escape_fns.copy()
if custom_escape_fns:
escape_fns.update(custom_escape_fns)
self._escape_fns = types.MappingProxyType(escape_fns)
self._stmt_classes = list(statements.builtin_stmt_classes)
class OutputStmt(statements.BaseOutput):
_filter_fn_names = list(self.escape_fns.keys())
self._stmt_classes.append(OutputStmt)
self._cache_sketches = cache_sketches
python类MappingProxyType()的实例源码
def __init__(self, name, mapping):
if type(mapping) is not dict: # we really only want dict literals
raise ValueError()
if type(name) is not str:
raise ValueError()
for key in mapping:
self._validate_member_(key)
for value in mapping.values():
self._validate_member_(value)
# Read only proxy of mapping, mutation to `mapping` will be reflected
# but there isn't much we can do about that. Good use of this object
# would involve a dict literal anyway.
self.mapping = types.MappingProxyType(mapping)
super().__init__(name)
def get_config(filename):
try:
config_dir = os.path.dirname(filename)
with NamedTemporaryFile(suffix='.py', dir=config_dir, delete=True) as tf:
shutil.copyfile(filename, tf.name)
spec = importlib.util.spec_from_file_location('_config.settings', tf.name)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
if hasattr(module, '__all__'):
settings = {k: getattr(module, k) for k in module.__all__}
else:
settings = {k: v for k, v in vars(module).items() if not k.startswith('_')}
return MappingProxyType(settings)
except Exception:
sys.stderr.write('Failed to read config file: %s' % filename)
sys.stderr.flush()
raise
def validate_query(query, possible_columns):
q = validate_query_structure(query)
sort_field = q.get('_sortField')
filters = q.get('_filters', [])
columns = [field_name for field_name in filters]
if sort_field is not None:
columns.append(sort_field)
not_valid = set(columns).difference(
possible_columns + [MULTI_FIELD_TEXT_QUERY])
if not_valid:
column_list = ', '.join(not_valid)
msg = 'Columns: {} do not present in resource'.format(column_list)
raise JsonValidaitonError(msg)
return MappingProxyType(q)
def json_decoder_hook(dct, str_decoders=STRING_DECODERS,
converters=MappingProxyType(dict())) -> dict:
"""Decoder for parsing typical objects like uuid's and dates."""
for k, v in dct.items():
if k in converters:
parse_func = converters[k]
dct[k] = parse_func(v)
elif isinstance(v, str):
for decode_func in str_decoders:
v = decode_func(v)
if not isinstance(v, str):
break
dct[k] = v
elif isinstance(v, collections.Mapping):
dct[k] = json_decoder_hook(v, str_decoders, converters)
return dct
def make_json_decoder_hook(str_decoders=STRING_DECODERS,
extra_str_decoders=tuple(),
converters=MappingProxyType(dict())) -> Callable:
"""Customize JSON string decoder hooks.
Object hook for typical deserialization scenarios.
Notes
-----
Specifying a field in converters will ensure custom decoding/passthrough.
Parameters
----------
str_decoders: functions for decoding strings to objects.
extra_str_decoders: appends additional string decoders to str_decoders.
converters: field / parser function mapping.
"""
str_decoders = tuple(chain(str_decoders, extra_str_decoders))
object_hook = partial(json_decoder_hook, str_decoders=str_decoders,
converters=converters)
return object_hook
def _make_context_immutable(context):
"""Best effort attempt at turning a properly formatted context
(either a string, dict, or array of strings and dicts) into an
immutable data structure.
If we get an array, make it immutable by creating a tuple; if we get
a dict, copy it into a MappingProxyType. Otherwise, return as-is.
"""
def make_immutable(val):
if isinstance(val, Mapping):
return MappingProxyType(val)
else:
return val
if not isinstance(context, (str, Mapping)):
try:
return tuple([make_immutable(val) for val in context])
except TypeError:
pass
return make_immutable(context)
def __call__(self, inst, attr, value):
if not isinstance(value, dict) and \
not isinstance(value, MappingProxyType):
raise TypeError(
"""'{name}' must be a dict object or
instance of MappingProxyType.""".format(
name=attr.name,
),
)
if not self.required_keypaths:
return
for keypath in self.required_keypaths:
current = value
components = keypath.split('.')
for component in components[:-1]:
current = current.get(component, {})
if components[-1] not in current:
raise DictValidatorKeypathDoesNotExistError(
"'{keypath}' not found".format(
keypath=keypath,
)
)
def cspec_from_feature_json(feature_json):
"""Create a CSpec instance from feature_json string.
:param feature_json: JSON string of Caniuse feature json
"""
raw = json.loads(feature_json)
cspec_kwargs = {}
direct_mapping_names = [
'bugs',
'categories',
'description',
'spec',
'stats',
'title',
]
for key in direct_mapping_names:
cspec_kwargs[key] = raw[key]
# stats is a relatively complex dict, we should prevent it
# from being mutated
cspec_kwargs['stats'] = MappingProxyType(cspec_kwargs['stats'])
return CSpec(**cspec_kwargs)
def __members__(cls):
"""Returns a mapping of member name->value.
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.
"""
return MappingProxyType(cls._member_map_)
def parameters(self):
try:
return types.MappingProxyType(self._parameters)
except AttributeError:
return OrderedDict(self._parameters.items())
def repr_helper(inp: Any) -> str:
"""Help creating a more readable string representation of objects."""
if isinstance(inp, (dict, MappingProxyType)):
return ", ".join(
repr_helper(key)+"="+repr_helper(item) for key, item
in inp.items())
elif isinstance(inp, datetime):
return as_local(inp).isoformat()
else:
return str(inp)
def __members__(cls):
"""Returns a mapping of member name->value.
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.
"""
return MappingProxyType(cls._member_map_)
def parameters(self):
try:
return types.MappingProxyType(self._parameters)
except AttributeError:
return OrderedDict(self._parameters.items())
def parameters(self):
try:
return types.MappingProxyType(self._parameters)
except AttributeError:
return OrderedDict(self._parameters.items())
def actions(self):
# TODO this doesn't handle method/visualizer name collisions. The
# auto-generated `qiime2.plugins.<plugin-name>.actions` API has the
# same problem. This should be solved at method/visualizer registration
# time, which will solve the problem for both APIs.
actions = {}
actions.update(self.methods)
actions.update(self.visualizers)
actions.update(self.pipelines)
return types.MappingProxyType(actions)
def test_actions(self):
actions = self.plugin.actions
self.assertIsInstance(actions, types.MappingProxyType)
self.assertEqual(actions.keys(),
{'merge_mappings', 'concatenate_ints', 'split_ints',
'most_common_viz', 'mapping_viz',
'identity_with_metadata',
'identity_with_metadata_category',
'identity_with_optional_metadata',
'identity_with_optional_metadata_category',
'params_only_method', 'no_input_method',
'optional_artifacts_method', 'variadic_input_method',
'params_only_viz', 'no_input_viz',
'long_description_method', 'parameter_only_pipeline',
'typical_pipeline', 'optional_artifact_pipeline',
'pointless_pipeline', 'visualizer_only_pipeline',
'pipelines_in_pipeline', 'failing_pipeline'})
for action in actions.values():
self.assertIsInstance(action, qiime2.sdk.Action)
# Read-only dict.
with self.assertRaises(TypeError):
actions["i-shouldn't-do-this"] = "my-action"
with self.assertRaises(TypeError):
actions["merge_mappings"] = "my-action"
def __init__(self, name, field_names, field_members, variant_of):
self.field_names = field_names
self.field = types.MappingProxyType({
f: VariantField(name, f, field_members[f])
for f in self.field_names})
self.variant_of = variant_of
super().__init__(name, field_names)
def __members__(cls):
"""Returns a mapping of member name->value.
This mapping lists all enum members, including aliases. Note that this
is a read-only view of the internal mapping.
"""
return MappingProxyType(cls._member_map_)
def parameters(self):
try:
return types.MappingProxyType(self._parameters)
except AttributeError:
return OrderedDict(self._parameters.items())
def view(self) -> 'typing.Mapping[int, channel.Channel]':
"""
:return: A read-only view into the channels for this wrapper.
"""
return MappingProxyType(self._channels)
def view(self) -> 'typing.Mapping[int, role.Role]':
"""
:return: A read-only view into the channels for this wrapper.
"""
return MappingProxyType(self._roles)
def view(self) -> 'typing.Mapping[int, dt_emoji.Emoji]':
"""
:return: A read-only view into the channels for this wrapper.
"""
return MappingProxyType(self._emojis)
def members(self) -> 'typing.Mapping[int, dt_member.Member]':
"""
:return: A mapping of :class:`~.Member` that represent members on this guild.
"""
return MappingProxyType(self._members)
def recipients(self) -> '_typing.Mapping[int, dt_user.User]':
"""
:return: A mapping of int -> :class:`~.User` for the recipients of this private chat.
"""
return MappingProxyType(self._recipients)
def channels(self) -> 'typing.Mapping[int, WidgetChannel]':
"""
:return: A read-only mapping of :class:`~.WidgetChannel` representing the channels for \
this guild.
"""
return MappingProxyType(self._channels)
def members(self) -> 'typing.Mapping[int, WidgetMember]':
"""
:return: A read-only mapping of :class:`~.WidgetMember` representing the channels for \
this guild.
"""
return MappingProxyType(self._members)
def friends(self) -> typing.Mapping[int, 'RelationshipUser']:
"""
:return: A mapping of :class:`~.RelationshipUser` that represents the friends for this user.
"""
if self.bot:
raise CuriousError("Bots cannot have friends")
return MappingProxyType(self._bot.state._friends)
def blocks(self) -> typing.Mapping[int, 'RelationshipUser']:
"""
:return: A mapping of :class:`~.RelationshipUser` that represents the blocked users for \
this user.
"""
if self.bot:
raise CuriousError("Bots cannot have friends")
return MappingProxyType(self._bot.state._blocked)
def gateways(self):
"""
:return: A read-only view of the current gateways for this client.
"""
return MappingProxyType(self._gateways)
