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)
python类ImmutableDict()的实例源码
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
