我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用types.MappingProxyType()。
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
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 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 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)
def with_order(self) -> 'typing.Mapping[int, Guild]': """ :return: A mapping of the guilds with the order specified in the ready packet. """ if not self.order: return self.view() o = collections.OrderedDict() for guild in map(int, self.order): o[guild] = self.guilds[guild] return MappingProxyType(o) # abc methods
def __init__(self, parameters=None, *, return_annotation=_empty, __validate_parameters__=True): '''Constructs Signature from the given list of Parameter objects and 'return_annotation'. All arguments are optional. ''' if parameters is None: params = OrderedDict() else: if __validate_parameters__: params = OrderedDict() top_kind = _POSITIONAL_ONLY for idx, param in enumerate(parameters): kind = param.kind if kind < top_kind: msg = 'wrong parameter order: {} before {}' msg = msg.format(top_kind, param.kind) raise ValueError(msg) else: top_kind = kind name = param.name if name is None: name = str(idx) param = param.replace(name=name) if name in params: msg = 'duplicate parameter name: {!r}'.format(name) raise ValueError(msg) params[name] = param else: params = OrderedDict(((param.name, param) for param in parameters)) self._parameters = types.MappingProxyType(params) self._return_annotation = return_annotation
def isNonPrimitiveInstance(x): return not isinstance(x,(float,int,type,tuple,list,dict,str,bytes,complex,bool,slice,_rl_NoneType, types.FunctionType,types.LambdaType,types.CodeType, types.MappingProxyType,types.SimpleNamespace, types.GeneratorType,types.MethodType,types.BuiltinFunctionType, types.BuiltinMethodType,types.ModuleType,types.TracebackType, types.FrameType,types.GetSetDescriptorType,types.MemberDescriptorType))
def params(self): return MappingProxyType(self._params)
def get_global_outputs(): import vapoursynth if hasattr(vapoursynth, "get_outputs"): return vapoursynth.get_outputs() return types.MappingProxyType(vapoursynth._get_output_dict("OutputManager.get_outputs"))
def _sp_to_field(mcs, fields): """ Returns an ordered dictionary, which maps the source pointer of a field to the field. Nested fields are listed before the parent. """ from aiohttp_json_api.fields.base import Relationship result = OrderedDict() for field in fields: if isinstance(field, Relationship): result.update(mcs._sp_to_field(field.links.values())) result[field.sp] = field return MappingProxyType(result)
def named_resources(self): return MappingProxyType(self._named_resources)
def cookies(self): """Return request cookies. A read-only dictionary-like object. """ raw = self.headers.get(hdrs.COOKIE, '') parsed = SimpleCookie(raw) return MappingProxyType( {key: val.value for key, val in parsed.items()})
def cached_hosts(self): """Read-only dict of cached DNS record.""" return MappingProxyType(self._cached_hosts)