Python werkzeug.datastructures 模块，ImmutableDict() 实例源码

def __setitem__(self, key, value):
        """store key-value to collection (lazy if value isn't changed)"""
        assert PRIMARY_KEY not in value or value[PRIMARY_KEY] == key
        # copy before write
        value = dict(value)
        value[PRIMARY_KEY] = key
        # do nothing on no changes
        if key in self._store.keys() and value == self._store[key]:
            return
        with self._context:
            self._collection.find_one_and_replace(
                {PRIMARY_KEY: key},
                value,
                upsert=True
            )
        self._store[key] = datastruct.ImmutableDict(value)

def update_context(self, context, app=None):
        """Replace the component's context with a new one.

        Args:
            context (dict): The new context to set this component's context to.

        Keyword Args:
            app (flask.Flask, optional): The app to update this context for. If
                not provided, the result of ``Component.app`` will be used.
        """
        if (app is None and self._context is _CONTEXT_MISSING
                and not in_app_context()):
            raise RuntimeError("Attempted to update component context without"
                               " a bound app context or eager app set! Please"
                               " pass the related app you want to update the"
                               " context for!")

        if self._context is not _CONTEXT_MISSING:
            self._context = ImmutableDict(context)
        else:
            key = self._get_context_name(app=app)
            setattr(_CONTEXT_LOCALS, key, ImmutableDict(context))

def web_config(self) -> typing.Mapping[str, object]:
        """(:class:`typing.Mapping`) The configuration maping for
        web that will go to :attr:`flask.Flask.config <Flask.config>`.

        """
        web_config = self.config.get('web', {})
        if not isinstance(web_config, collections.abc.Mapping):
            web_config = {}
        return ImmutableDict((k.upper(), v) for k, v in web_config.items())

def test_default_converters(self):
        class MyMap(r.Map):
            default_converters = r.Map.default_converters.copy()
            default_converters['foo'] = r.UnicodeConverter
        assert isinstance(r.Map.default_converters, ImmutableDict)
        m = MyMap([
            r.Rule('/a/<foo:a>', endpoint='a'),
            r.Rule('/b/<foo:b>', endpoint='b'),
            r.Rule('/c/<c>', endpoint='c')
        ], converters={'bar': r.UnicodeConverter})
        a = m.bind('example.org', '/')
        assert a.match('/a/1') == ('a', {'a': '1'})
        assert a.match('/b/2') == ('b', {'b': '2'})
        assert a.match('/c/3') == ('c', {'c': '3'})
        assert 'foo' not in r.Map.default_converters

def reload(self):
        """read collection to dictionary"""
        with self._context:
            rows = tuple(self._collection.find({}))
            self._store = {
                row[PRIMARY_KEY]: datastruct.ImmutableDict(row) for row in rows
            }

def context(self):
        """Return the current context for the component.

        Returns:
            werkzeug.datastructures.ImmutableDict: The current ``context`` that
                this component is being used within.
        """
        if self._context is not _CONTEXT_MISSING:
            return self._context

        return _CONTEXT_LOCALS.context

def _run_post_configure_callbacks(self, configure_args):
        """Run all post configure callbacks we have stored.

        Functions are passed the configuration that resulted from the call to
        :meth:`configure` as the first argument, in an immutable form; and are
        given the arguments passed to :meth:`configure` for the second
        argument.

        Returns from callbacks are ignored in all fashion.

        Args:
            configure_args (list[object]):
                The full list of arguments passed to :meth:`configure`.

        Returns:
            None:
                Does not return anything.
        """
        resulting_configuration = ImmutableDict(self.config)

        # copy callbacks in case people edit them while running
        multiple_callbacks = copy.copy(
            self._post_configure_callbacks['multiple']
        )
        single_callbacks = copy.copy(self._post_configure_callbacks['single'])
        # clear out the singles
        self._post_configure_callbacks['single'] = []

        for callback in multiple_callbacks:
            callback(resulting_configuration, configure_args)

        # now do the single run callbacks
        for callback in single_callbacks:
            callback(resulting_configuration, configure_args)

def test_config_post_configure_callbacks():
    """Ensure post configure callbacks work."""
    app = MultiStageConfigurableApp.create_app('tests')
    app.config['CANARY'] = True

    configure_args = {
        'FOO': 'bar',
    }

    def evil_cb(cfg, cfg_args):
        """This fella tries to change the config so we can make sure we pass
        around a frozen config.
        """
        # this is the only way you can change the config from here
        app.config['BAD_VAL'] = True

    def small_cb(cfg, cfg_args):
        """Ensure that we get the right arguments to our callbacks."""
        assert cfg['CANARY']
        assert cfg['FOO'] == 'bar'
        assert cfg_args == (configure_args,)
        assert 'BAD_VAL' not in cfg

    # we need the proper arguments to the call assertions below, so construct
    # them.
    expected_config = app.config.copy()
    expected_config['FOO'] = 'bar'
    expected_config = ImmutableDict(expected_config)

    # make sure we can count calls and call order
    small_cb = mock.MagicMock(wraps=small_cb)
    evil_cb = mock.MagicMock(wraps=evil_cb)
    parent_mock = mock.Mock()
    parent_mock.m1, parent_mock.m2 = small_cb, evil_cb

    assert not app._post_configure_callbacks['single']
    assert not app._post_configure_callbacks['multiple']

    app.add_post_configure_callback(evil_cb)
    app.add_post_configure_callback(small_cb)

    assert len(app._post_configure_callbacks) == 2

    app.configure(configure_args)

    # ensure we called the right number of times in the right order
    assert small_cb.call_count == 1
    assert evil_cb.call_count == 1
    parent_mock.assert_has_calls([
        mock.call.m2(expected_config, (configure_args,)),
        mock.call.m1(expected_config, (configure_args,)),
    ])

def add_post_configure_callback(self, callback, run_once=False):
        """Add a new callback to be run after every call to :meth:`configure`.

        Functions run at the end of :meth:`configure` are given the
        application's resulting configuration and the arguments passed to
        :meth:`configure`, in that order. As a note, this first argument will
        be an immutable dictionary.

        The return value of all registered callbacks is entirely ignored.

        Callbacks are run in the order they are registered, but you should
        never depend on another callback.

        .. admonition:: The "Resulting" Configuration

            The first argument to the callback is always the "resulting"
            configuration from the call to :meth:`configure`. What this means
            is you will get the Application's FROZEN configuration after the
            call to :meth:`configure` finished. Moreover, this resulting
            configuration will be an
            :class:`~werkzeug.datastructures.ImmutableDict`.

            The purpose of a Post Configure callback is not to futher alter the
            configuration, but rather to do lazy initialization for anything
            that absolutely requires the configuration, so any attempt to alter
            the configuration of the app has been made intentionally difficult!

        Args:
            callback (function):
                The function you wish to run after :meth:`configure`. Will
                receive the application's current configuration as the first
                arugment, and the same arguments passed to :meth:`configure` as
                the second.

        Keyword Args:
            run_once (bool):
                Should this callback run every time configure is called? Or
                just once and be deregistered? Pass ``True`` to only run it
                once.

        Returns:
            fleaker.base.BaseApplication:
                Returns itself for a fluent interface.
        """
        if run_once:
            self._post_configure_callbacks['single'].append(callback)
        else:
            self._post_configure_callbacks['multiple'].append(callback)

        return self