We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Straighten out commit_veto situation
In https://github.com/Kinto/kinto-signer/issues/591, we discovered that failing requests do not necessarily abort the transaction. (Exceptions do, but non-exceptional responses with e.g. 503 status codes don't.) There seems to be some confusion about this throughout the codebase; for example see https://github.com/Kinto/kinto-signer/blob/master/kinto_signer/listeners.py#L23-L24. At some point it must have seemed like Pyramid's [commit_veto hook](https://docs.pylonsproject.org/projects/pyramid_tm/en/latest/#adding-a-commit-veto-hook) would solve this automatically for us, but actually rereading this I see that "By default, `pyramid_tm` does not configure a commit veto into the system; you must do it explicitly."

However, the docstring for our [notify_resource_events_before](https://github.com/Kinto/kinto/blob/master/kinto/core/events.py#L152) *claims* to be a commit veto hook, muddying the waters somewhat. This code was added in https://github.com/Kinto/kinto/pull/1263, when we discovered that running "before resource changed" events as a "before commit" hook was invalid and we had to figure out a better approach.

- It seems like at a very minimum, the docstring here is invalid. I believe this is not really a veto but rather just part of the events machinery. 
- Should there be a `commit_veto` hook? Personally I would find it surprising that 4xx/5xx responses always automatically abort the transaction, even if they were not driven by exceptions.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `kinto/core/events.py`
Content:
```
1 import logging
2 import warnings
3 from collections import OrderedDict
4 
5 import transaction
6 from pyramid.events import NewRequest
7 import pyramid.tweens
8 from enum import Enum
9 
10 from kinto.core.utils import strip_uri_prefix
11 
12 
13 logger = logging.getLogger(__name__)
14 
15 
16 class ACTIONS(Enum):
17     CREATE = "create"
18     DELETE = "delete"
19     READ = "read"
20     UPDATE = "update"
21 
22     @staticmethod
23     def from_string_list(elements):
24         return tuple(ACTIONS(el) for el in elements)
25 
26 
27 class _ResourceEvent:
28     def __init__(self, payload, request):
29         self.payload = payload
30         self.request = request
31 
32     def __repr__(self):
33         return f"<{self.__class__.__name__} action={self.payload['action']} uri={self.payload['uri']}>"
34 
35     @property
36     def read_records(self):
37         message = "`read_records` is deprecated, use `read_objects` instead."
38         warnings.warn(message, DeprecationWarning)
39         return self.read_objects
40 
41     @property
42     def impacted_records(self):
43         message = "`impacted_records` is deprecated, use `impacted_objects` instead."
44         warnings.warn(message, DeprecationWarning)
45         return self.impacted_objects
46 
47 
48 class ResourceRead(_ResourceEvent):
49     """Triggered when a resource is being read.
50     """
51 
52     def __init__(self, payload, read_objects, request):
53         super().__init__(payload, request)
54         self.read_objects = read_objects
55 
56 
57 class ResourceChanged(_ResourceEvent):
58     """Triggered when a resource is being changed.
59     """
60 
61     def __init__(self, payload, impacted_objects, request):
62         super().__init__(payload, request)
63         self.impacted_objects = impacted_objects
64 
65 
66 class AfterResourceRead(_ResourceEvent):
67     """Triggered after a resource was successfully read.
68     """
69 
70     def __init__(self, payload, read_objects, request):
71         super().__init__(payload, request)
72         self.read_objects = read_objects
73 
74 
75 class AfterResourceChanged(_ResourceEvent):
76     """Triggered after a resource was successfully changed.
77     """
78 
79     def __init__(self, payload, impacted_objects, request):
80         super().__init__(payload, request)
81         self.impacted_objects = impacted_objects
82 
83 
84 class EventCollector(object):
85     """A collection to gather events emitted over the course of a request.
86 
87     Events are gathered by parent id, resource type, and event
88     type. This serves as a primitive normalization so that we can emit
89     fewer events.
90     """
91 
92     def __init__(self):
93         self.event_dict = OrderedDict()
94         """The events as collected so far.
95 
96         The key of the event_dict is a triple (resource_name,
97         parent_id, action). The value is a triple (impacted, request,
98         payload). If the same (resource_name, parent_id, action) is
99         encountered, we just extend the existing impacted with the new
100         impacted. N.B. this means all values in the payload must not
101         be specific to a single impacted_object. See
102         https://github.com/Kinto/kinto/issues/945 and
103         https://github.com/Kinto/kinto/issues/1731.
104         """
105 
106     def add_event(self, resource_name, parent_id, action, payload, impacted, request):
107         key = (resource_name, parent_id, action)
108         if key not in self.event_dict:
109             value = (payload, impacted, request)
110             self.event_dict[key] = value
111         else:
112             old_value = self.event_dict[key]
113             (old_payload, old_impacted, old_request) = old_value
114             # May be a good idea to assert that old_payload == payload here.
115             self.event_dict[key] = (old_payload, old_impacted + impacted, old_request)
116 
117     def drain(self):
118         """Return an iterator that removes elements from this EventCollector.
119 
120         This can be used to process events while still allowing events
121         to be added (for instance, as part of a cascade where events
122         add other events).
123 
124         Items yielded will be of a tuple suitable for using as
125         arguments to EventCollector.add_event.
126         """
127         return EventCollectorDrain(self)
128 
129 
130 class EventCollectorDrain(object):
131     """An iterator that drains an EventCollector.
132 
133     Get one using EventCollector.drain()."""
134 
135     def __init__(self, event_collector):
136         self.event_collector = event_collector
137 
138     def __iter__(self):
139         return self
140 
141     def __next__(self):
142         if self.event_collector.event_dict:
143             # Get the "first" key in insertion order, so as to process
144             # events in the same order they were queued.
145             key = next(iter(self.event_collector.event_dict.keys()))
146             value = self.event_collector.event_dict.pop(key)
147             return key + value
148         else:
149             raise StopIteration
150 
151 
152 def notify_resource_events_before(handler, registry):
153     """pyramid_tm "commit veto" hook to run ResourceChanged events.
154 
155     This hook being a "commit veto" let us tell pyramid_tm to abort
156     the transaction if the ResourceChanged listeners raise.
157     """
158 
159     def tween(request):
160         response = handler(request)
161         for event in request.get_resource_events():
162             request.registry.notify(event)
163 
164         return response
165 
166     return tween
167 
168 
169 def setup_transaction_hook(config):
170     """
171     Resource events are plugged with the transactions of ``pyramid_tm``.
172 
173     Once a transaction is committed, ``AfterResourceRead`` and
174     ``AfterResourceChanged`` events are sent.
175     """
176 
177     def _notify_resource_events_after(success, request):
178         """Notify the accumulated resource events if transaction succeeds.
179         """
180         if not success:  # pragma: no cover
181             return
182 
183         for event in request.get_resource_events(after_commit=True):
184             try:
185                 request.registry.notify(event)
186             except Exception:
187                 logger.error("Unable to notify", exc_info=True)
188 
189     def on_new_request(event):
190         """When a new request comes in, hook on transaction commit.
191         """
192         # Since there is one transaction per batch, ignore subrequests.
193         if hasattr(event.request, "parent"):
194             return
195         current = transaction.get()
196         current.addAfterCommitHook(_notify_resource_events_after, args=(event.request,))
197 
198     config.add_subscriber(on_new_request, NewRequest)
199     config.add_tween(
200         "kinto.core.events.notify_resource_events_before", under=pyramid.tweens.EXCVIEW
201     )
202 
203 
204 def get_resource_events(request, after_commit=False):
205     """Generator to iterate the list of events triggered on resources.
206 
207     The list is sorted chronologically (see OrderedDict).
208 
209     This drains the resource_events currently in the request, which
210     allows us to process new events as they are added by current
211     events. However, once the iteration is over, we merge all the
212     events we've emitted into a new resource_events, which we store on
213     the request so we can reprocess the same events in an after-commit
214     tween.
215 
216     This generator must be completely consumed!
217     """
218     by_resource = request.bound_data.get("resource_events", EventCollector())
219     afterwards = EventCollector()
220 
221     for event_call in by_resource.drain():
222         afterwards.add_event(*event_call)
223         (_, _, action, payload, impacted, request) = event_call
224 
225         if after_commit:
226             if action == ACTIONS.READ:
227                 event_cls = AfterResourceRead
228             else:
229                 event_cls = AfterResourceChanged
230         else:
231             if action == ACTIONS.READ:
232                 event_cls = ResourceRead
233             else:
234                 event_cls = ResourceChanged
235 
236         yield event_cls(payload, impacted, request)
237 
238     request.bound_data["resource_events"] = afterwards
239 
240 
241 def notify_resource_event(
242     request, parent_id, timestamp, data, action, old=None, resource_name=None, resource_data=None
243 ):
244     """Request helper to stack a resource event.
245 
246     If a similar event (same resource, same action) already occured during the
247     current transaction (e.g. batch) then just extend the impacted objects of
248     the previous one.
249 
250     :param resource_name: The name of the resource on which the event
251         happened (taken from the request if not provided).
252     :param resource_data: Information about the resource on which the
253         event is being emitted. Usually contains information about how
254         to find this object in the hierarchy (for instance,
255         ``bucket_id`` and ``collection_id`` for a record). Taken from
256         the request matchdict if absent.
257     :type resource_data: dict
258 
259     """
260     if action == ACTIONS.READ:
261         if not isinstance(data, list):
262             data = [data]
263         impacted = data
264     elif action == ACTIONS.CREATE:
265         impacted = [{"new": data}]
266     elif action == ACTIONS.DELETE:
267         if not isinstance(data, list):
268             impacted = [{"new": data, "old": old}]
269         else:
270             impacted = []
271             for i, new in enumerate(data):
272                 impacted.append({"new": new, "old": old[i]})
273     else:  # ACTIONS.UPDATE:
274         impacted = [{"new": data, "old": old}]
275 
276     # Get previously triggered events.
277     events = request.bound_data.setdefault("resource_events", EventCollector())
278 
279     resource_name = resource_name or request.current_resource_name
280     matchdict = resource_data or dict(request.matchdict)
281 
282     payload = {
283         "timestamp": timestamp,
284         "action": action.value,
285         # Deprecated: don't actually use URI (see #945).
286         "uri": strip_uri_prefix(request.path),
287         "user_id": request.prefixed_userid,
288         "resource_name": resource_name,
289     }
290 
291     # Deprecated: don't actually use `resource_name_id` either (see #945).
292     if "id" in request.matchdict:
293         matchdict[resource_name + "_id"] = matchdict.pop("id")
294 
295     payload.update(**matchdict)
296 
297     events.add_event(resource_name, parent_id, action, payload, impacted, request)
298
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/kinto/core/events.py b/kinto/core/events.py
--- a/kinto/core/events.py
+++ b/kinto/core/events.py
@@ -150,10 +150,12 @@
 
 
 def notify_resource_events_before(handler, registry):
-    """pyramid_tm "commit veto" hook to run ResourceChanged events.
+    """Tween that runs ResourceChanged events.
+
+    This tween runs under EXCVIEW, so exceptions raised by
+    ResourceChanged events are caught and dealt the same as any other
+    exceptions.
 
-    This hook being a "commit veto" let us tell pyramid_tm to abort
-    the transaction if the ResourceChanged listeners raise.
     """
 
     def tween(request):

{"golden_diff": "diff --git a/kinto/core/events.py b/kinto/core/events.py\n--- a/kinto/core/events.py\n+++ b/kinto/core/events.py\n@@ -150,10 +150,12 @@\n \n \n def notify_resource_events_before(handler, registry):\n-    \"\"\"pyramid_tm \"commit veto\" hook to run ResourceChanged events.\n+    \"\"\"Tween that runs ResourceChanged events.\n+\n+    This tween runs under EXCVIEW, so exceptions raised by\n+    ResourceChanged events are caught and dealt the same as any other\n+    exceptions.\n \n-    This hook being a \"commit veto\" let us tell pyramid_tm to abort\n-    the transaction if the ResourceChanged listeners raise.\n     \"\"\"\n \n     def tween(request):\n", "issue": "Straighten out commit_veto situation\nIn https://github.com/Kinto/kinto-signer/issues/591, we discovered that failing requests do not necessarily abort the transaction. (Exceptions do, but non-exceptional responses with e.g. 503 status codes don't.) There seems to be some confusion about this throughout the codebase; for example see https://github.com/Kinto/kinto-signer/blob/master/kinto_signer/listeners.py#L23-L24. At some point it must have seemed like Pyramid's [commit_veto hook](https://docs.pylonsproject.org/projects/pyramid_tm/en/latest/#adding-a-commit-veto-hook) would solve this automatically for us, but actually rereading this I see that \"By default, `pyramid_tm` does not configure a commit veto into the system; you must do it explicitly.\"\r\n\r\nHowever, the docstring for our [notify_resource_events_before](https://github.com/Kinto/kinto/blob/master/kinto/core/events.py#L152) *claims* to be a commit veto hook, muddying the waters somewhat. This code was added in https://github.com/Kinto/kinto/pull/1263, when we discovered that running \"before resource changed\" events as a \"before commit\" hook was invalid and we had to figure out a better approach.\r\n\r\n- It seems like at a very minimum, the docstring here is invalid. I believe this is not really a veto but rather just part of the events machinery. \r\n- Should there be a `commit_veto` hook? Personally I would find it surprising that 4xx/5xx responses always automatically abort the transaction, even if they were not driven by exceptions.\r\n\n", "before_files": [{"content": "import logging\nimport warnings\nfrom collections import OrderedDict\n\nimport transaction\nfrom pyramid.events import NewRequest\nimport pyramid.tweens\nfrom enum import Enum\n\nfrom kinto.core.utils import strip_uri_prefix\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass ACTIONS(Enum):\n    CREATE = \"create\"\n    DELETE = \"delete\"\n    READ = \"read\"\n    UPDATE = \"update\"\n\n    @staticmethod\n    def from_string_list(elements):\n        return tuple(ACTIONS(el) for el in elements)\n\n\nclass _ResourceEvent:\n    def __init__(self, payload, request):\n        self.payload = payload\n        self.request = request\n\n    def __repr__(self):\n        return f\"<{self.__class__.__name__} action={self.payload['action']} uri={self.payload['uri']}>\"\n\n    @property\n    def read_records(self):\n        message = \"`read_records` is deprecated, use `read_objects` instead.\"\n        warnings.warn(message, DeprecationWarning)\n        return self.read_objects\n\n    @property\n    def impacted_records(self):\n        message = \"`impacted_records` is deprecated, use `impacted_objects` instead.\"\n        warnings.warn(message, DeprecationWarning)\n        return self.impacted_objects\n\n\nclass ResourceRead(_ResourceEvent):\n    \"\"\"Triggered when a resource is being read.\n    \"\"\"\n\n    def __init__(self, payload, read_objects, request):\n        super().__init__(payload, request)\n        self.read_objects = read_objects\n\n\nclass ResourceChanged(_ResourceEvent):\n    \"\"\"Triggered when a resource is being changed.\n    \"\"\"\n\n    def __init__(self, payload, impacted_objects, request):\n        super().__init__(payload, request)\n        self.impacted_objects = impacted_objects\n\n\nclass AfterResourceRead(_ResourceEvent):\n    \"\"\"Triggered after a resource was successfully read.\n    \"\"\"\n\n    def __init__(self, payload, read_objects, request):\n        super().__init__(payload, request)\n        self.read_objects = read_objects\n\n\nclass AfterResourceChanged(_ResourceEvent):\n    \"\"\"Triggered after a resource was successfully changed.\n    \"\"\"\n\n    def __init__(self, payload, impacted_objects, request):\n        super().__init__(payload, request)\n        self.impacted_objects = impacted_objects\n\n\nclass EventCollector(object):\n    \"\"\"A collection to gather events emitted over the course of a request.\n\n    Events are gathered by parent id, resource type, and event\n    type. This serves as a primitive normalization so that we can emit\n    fewer events.\n    \"\"\"\n\n    def __init__(self):\n        self.event_dict = OrderedDict()\n        \"\"\"The events as collected so far.\n\n        The key of the event_dict is a triple (resource_name,\n        parent_id, action). The value is a triple (impacted, request,\n        payload). If the same (resource_name, parent_id, action) is\n        encountered, we just extend the existing impacted with the new\n        impacted. N.B. this means all values in the payload must not\n        be specific to a single impacted_object. See\n        https://github.com/Kinto/kinto/issues/945 and\n        https://github.com/Kinto/kinto/issues/1731.\n        \"\"\"\n\n    def add_event(self, resource_name, parent_id, action, payload, impacted, request):\n        key = (resource_name, parent_id, action)\n        if key not in self.event_dict:\n            value = (payload, impacted, request)\n            self.event_dict[key] = value\n        else:\n            old_value = self.event_dict[key]\n            (old_payload, old_impacted, old_request) = old_value\n            # May be a good idea to assert that old_payload == payload here.\n            self.event_dict[key] = (old_payload, old_impacted + impacted, old_request)\n\n    def drain(self):\n        \"\"\"Return an iterator that removes elements from this EventCollector.\n\n        This can be used to process events while still allowing events\n        to be added (for instance, as part of a cascade where events\n        add other events).\n\n        Items yielded will be of a tuple suitable for using as\n        arguments to EventCollector.add_event.\n        \"\"\"\n        return EventCollectorDrain(self)\n\n\nclass EventCollectorDrain(object):\n    \"\"\"An iterator that drains an EventCollector.\n\n    Get one using EventCollector.drain().\"\"\"\n\n    def __init__(self, event_collector):\n        self.event_collector = event_collector\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        if self.event_collector.event_dict:\n            # Get the \"first\" key in insertion order, so as to process\n            # events in the same order they were queued.\n            key = next(iter(self.event_collector.event_dict.keys()))\n            value = self.event_collector.event_dict.pop(key)\n            return key + value\n        else:\n            raise StopIteration\n\n\ndef notify_resource_events_before(handler, registry):\n    \"\"\"pyramid_tm \"commit veto\" hook to run ResourceChanged events.\n\n    This hook being a \"commit veto\" let us tell pyramid_tm to abort\n    the transaction if the ResourceChanged listeners raise.\n    \"\"\"\n\n    def tween(request):\n        response = handler(request)\n        for event in request.get_resource_events():\n            request.registry.notify(event)\n\n        return response\n\n    return tween\n\n\ndef setup_transaction_hook(config):\n    \"\"\"\n    Resource events are plugged with the transactions of ``pyramid_tm``.\n\n    Once a transaction is committed, ``AfterResourceRead`` and\n    ``AfterResourceChanged`` events are sent.\n    \"\"\"\n\n    def _notify_resource_events_after(success, request):\n        \"\"\"Notify the accumulated resource events if transaction succeeds.\n        \"\"\"\n        if not success:  # pragma: no cover\n            return\n\n        for event in request.get_resource_events(after_commit=True):\n            try:\n                request.registry.notify(event)\n            except Exception:\n                logger.error(\"Unable to notify\", exc_info=True)\n\n    def on_new_request(event):\n        \"\"\"When a new request comes in, hook on transaction commit.\n        \"\"\"\n        # Since there is one transaction per batch, ignore subrequests.\n        if hasattr(event.request, \"parent\"):\n            return\n        current = transaction.get()\n        current.addAfterCommitHook(_notify_resource_events_after, args=(event.request,))\n\n    config.add_subscriber(on_new_request, NewRequest)\n    config.add_tween(\n        \"kinto.core.events.notify_resource_events_before\", under=pyramid.tweens.EXCVIEW\n    )\n\n\ndef get_resource_events(request, after_commit=False):\n    \"\"\"Generator to iterate the list of events triggered on resources.\n\n    The list is sorted chronologically (see OrderedDict).\n\n    This drains the resource_events currently in the request, which\n    allows us to process new events as they are added by current\n    events. However, once the iteration is over, we merge all the\n    events we've emitted into a new resource_events, which we store on\n    the request so we can reprocess the same events in an after-commit\n    tween.\n\n    This generator must be completely consumed!\n    \"\"\"\n    by_resource = request.bound_data.get(\"resource_events\", EventCollector())\n    afterwards = EventCollector()\n\n    for event_call in by_resource.drain():\n        afterwards.add_event(*event_call)\n        (_, _, action, payload, impacted, request) = event_call\n\n        if after_commit:\n            if action == ACTIONS.READ:\n                event_cls = AfterResourceRead\n            else:\n                event_cls = AfterResourceChanged\n        else:\n            if action == ACTIONS.READ:\n                event_cls = ResourceRead\n            else:\n                event_cls = ResourceChanged\n\n        yield event_cls(payload, impacted, request)\n\n    request.bound_data[\"resource_events\"] = afterwards\n\n\ndef notify_resource_event(\n    request, parent_id, timestamp, data, action, old=None, resource_name=None, resource_data=None\n):\n    \"\"\"Request helper to stack a resource event.\n\n    If a similar event (same resource, same action) already occured during the\n    current transaction (e.g. batch) then just extend the impacted objects of\n    the previous one.\n\n    :param resource_name: The name of the resource on which the event\n        happened (taken from the request if not provided).\n    :param resource_data: Information about the resource on which the\n        event is being emitted. Usually contains information about how\n        to find this object in the hierarchy (for instance,\n        ``bucket_id`` and ``collection_id`` for a record). Taken from\n        the request matchdict if absent.\n    :type resource_data: dict\n\n    \"\"\"\n    if action == ACTIONS.READ:\n        if not isinstance(data, list):\n            data = [data]\n        impacted = data\n    elif action == ACTIONS.CREATE:\n        impacted = [{\"new\": data}]\n    elif action == ACTIONS.DELETE:\n        if not isinstance(data, list):\n            impacted = [{\"new\": data, \"old\": old}]\n        else:\n            impacted = []\n            for i, new in enumerate(data):\n                impacted.append({\"new\": new, \"old\": old[i]})\n    else:  # ACTIONS.UPDATE:\n        impacted = [{\"new\": data, \"old\": old}]\n\n    # Get previously triggered events.\n    events = request.bound_data.setdefault(\"resource_events\", EventCollector())\n\n    resource_name = resource_name or request.current_resource_name\n    matchdict = resource_data or dict(request.matchdict)\n\n    payload = {\n        \"timestamp\": timestamp,\n        \"action\": action.value,\n        # Deprecated: don't actually use URI (see #945).\n        \"uri\": strip_uri_prefix(request.path),\n        \"user_id\": request.prefixed_userid,\n        \"resource_name\": resource_name,\n    }\n\n    # Deprecated: don't actually use `resource_name_id` either (see #945).\n    if \"id\" in request.matchdict:\n        matchdict[resource_name + \"_id\"] = matchdict.pop(\"id\")\n\n    payload.update(**matchdict)\n\n    events.add_event(resource_name, parent_id, action, payload, impacted, request)\n", "path": "kinto/core/events.py"}], "after_files": [{"content": "import logging\nimport warnings\nfrom collections import OrderedDict\n\nimport transaction\nfrom pyramid.events import NewRequest\nimport pyramid.tweens\nfrom enum import Enum\n\nfrom kinto.core.utils import strip_uri_prefix\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass ACTIONS(Enum):\n    CREATE = \"create\"\n    DELETE = \"delete\"\n    READ = \"read\"\n    UPDATE = \"update\"\n\n    @staticmethod\n    def from_string_list(elements):\n        return tuple(ACTIONS(el) for el in elements)\n\n\nclass _ResourceEvent:\n    def __init__(self, payload, request):\n        self.payload = payload\n        self.request = request\n\n    def __repr__(self):\n        return f\"<{self.__class__.__name__} action={self.payload['action']} uri={self.payload['uri']}>\"\n\n    @property\n    def read_records(self):\n        message = \"`read_records` is deprecated, use `read_objects` instead.\"\n        warnings.warn(message, DeprecationWarning)\n        return self.read_objects\n\n    @property\n    def impacted_records(self):\n        message = \"`impacted_records` is deprecated, use `impacted_objects` instead.\"\n        warnings.warn(message, DeprecationWarning)\n        return self.impacted_objects\n\n\nclass ResourceRead(_ResourceEvent):\n    \"\"\"Triggered when a resource is being read.\n    \"\"\"\n\n    def __init__(self, payload, read_objects, request):\n        super().__init__(payload, request)\n        self.read_objects = read_objects\n\n\nclass ResourceChanged(_ResourceEvent):\n    \"\"\"Triggered when a resource is being changed.\n    \"\"\"\n\n    def __init__(self, payload, impacted_objects, request):\n        super().__init__(payload, request)\n        self.impacted_objects = impacted_objects\n\n\nclass AfterResourceRead(_ResourceEvent):\n    \"\"\"Triggered after a resource was successfully read.\n    \"\"\"\n\n    def __init__(self, payload, read_objects, request):\n        super().__init__(payload, request)\n        self.read_objects = read_objects\n\n\nclass AfterResourceChanged(_ResourceEvent):\n    \"\"\"Triggered after a resource was successfully changed.\n    \"\"\"\n\n    def __init__(self, payload, impacted_objects, request):\n        super().__init__(payload, request)\n        self.impacted_objects = impacted_objects\n\n\nclass EventCollector(object):\n    \"\"\"A collection to gather events emitted over the course of a request.\n\n    Events are gathered by parent id, resource type, and event\n    type. This serves as a primitive normalization so that we can emit\n    fewer events.\n    \"\"\"\n\n    def __init__(self):\n        self.event_dict = OrderedDict()\n        \"\"\"The events as collected so far.\n\n        The key of the event_dict is a triple (resource_name,\n        parent_id, action). The value is a triple (impacted, request,\n        payload). If the same (resource_name, parent_id, action) is\n        encountered, we just extend the existing impacted with the new\n        impacted. N.B. this means all values in the payload must not\n        be specific to a single impacted_object. See\n        https://github.com/Kinto/kinto/issues/945 and\n        https://github.com/Kinto/kinto/issues/1731.\n        \"\"\"\n\n    def add_event(self, resource_name, parent_id, action, payload, impacted, request):\n        key = (resource_name, parent_id, action)\n        if key not in self.event_dict:\n            value = (payload, impacted, request)\n            self.event_dict[key] = value\n        else:\n            old_value = self.event_dict[key]\n            (old_payload, old_impacted, old_request) = old_value\n            # May be a good idea to assert that old_payload == payload here.\n            self.event_dict[key] = (old_payload, old_impacted + impacted, old_request)\n\n    def drain(self):\n        \"\"\"Return an iterator that removes elements from this EventCollector.\n\n        This can be used to process events while still allowing events\n        to be added (for instance, as part of a cascade where events\n        add other events).\n\n        Items yielded will be of a tuple suitable for using as\n        arguments to EventCollector.add_event.\n        \"\"\"\n        return EventCollectorDrain(self)\n\n\nclass EventCollectorDrain(object):\n    \"\"\"An iterator that drains an EventCollector.\n\n    Get one using EventCollector.drain().\"\"\"\n\n    def __init__(self, event_collector):\n        self.event_collector = event_collector\n\n    def __iter__(self):\n        return self\n\n    def __next__(self):\n        if self.event_collector.event_dict:\n            # Get the \"first\" key in insertion order, so as to process\n            # events in the same order they were queued.\n            key = next(iter(self.event_collector.event_dict.keys()))\n            value = self.event_collector.event_dict.pop(key)\n            return key + value\n        else:\n            raise StopIteration\n\n\ndef notify_resource_events_before(handler, registry):\n    \"\"\"Tween that runs ResourceChanged events.\n\n    This tween runs under EXCVIEW, so exceptions raised by\n    ResourceChanged events are caught and dealt the same as any other\n    exceptions.\n\n    \"\"\"\n\n    def tween(request):\n        response = handler(request)\n        for event in request.get_resource_events():\n            request.registry.notify(event)\n\n        return response\n\n    return tween\n\n\ndef setup_transaction_hook(config):\n    \"\"\"\n    Resource events are plugged with the transactions of ``pyramid_tm``.\n\n    Once a transaction is committed, ``AfterResourceRead`` and\n    ``AfterResourceChanged`` events are sent.\n    \"\"\"\n\n    def _notify_resource_events_after(success, request):\n        \"\"\"Notify the accumulated resource events if transaction succeeds.\n        \"\"\"\n        if not success:  # pragma: no cover\n            return\n\n        for event in request.get_resource_events(after_commit=True):\n            try:\n                request.registry.notify(event)\n            except Exception:\n                logger.error(\"Unable to notify\", exc_info=True)\n\n    def on_new_request(event):\n        \"\"\"When a new request comes in, hook on transaction commit.\n        \"\"\"\n        # Since there is one transaction per batch, ignore subrequests.\n        if hasattr(event.request, \"parent\"):\n            return\n        current = transaction.get()\n        current.addAfterCommitHook(_notify_resource_events_after, args=(event.request,))\n\n    config.add_subscriber(on_new_request, NewRequest)\n    config.add_tween(\n        \"kinto.core.events.notify_resource_events_before\", under=pyramid.tweens.EXCVIEW\n    )\n\n\ndef get_resource_events(request, after_commit=False):\n    \"\"\"Generator to iterate the list of events triggered on resources.\n\n    The list is sorted chronologically (see OrderedDict).\n\n    This drains the resource_events currently in the request, which\n    allows us to process new events as they are added by current\n    events. However, once the iteration is over, we merge all the\n    events we've emitted into a new resource_events, which we store on\n    the request so we can reprocess the same events in an after-commit\n    tween.\n\n    This generator must be completely consumed!\n    \"\"\"\n    by_resource = request.bound_data.get(\"resource_events\", EventCollector())\n    afterwards = EventCollector()\n\n    for event_call in by_resource.drain():\n        afterwards.add_event(*event_call)\n        (_, _, action, payload, impacted, request) = event_call\n\n        if after_commit:\n            if action == ACTIONS.READ:\n                event_cls = AfterResourceRead\n            else:\n                event_cls = AfterResourceChanged\n        else:\n            if action == ACTIONS.READ:\n                event_cls = ResourceRead\n            else:\n                event_cls = ResourceChanged\n\n        yield event_cls(payload, impacted, request)\n\n    request.bound_data[\"resource_events\"] = afterwards\n\n\ndef notify_resource_event(\n    request, parent_id, timestamp, data, action, old=None, resource_name=None, resource_data=None\n):\n    \"\"\"Request helper to stack a resource event.\n\n    If a similar event (same resource, same action) already occured during the\n    current transaction (e.g. batch) then just extend the impacted objects of\n    the previous one.\n\n    :param resource_name: The name of the resource on which the event\n        happened (taken from the request if not provided).\n    :param resource_data: Information about the resource on which the\n        event is being emitted. Usually contains information about how\n        to find this object in the hierarchy (for instance,\n        ``bucket_id`` and ``collection_id`` for a record). Taken from\n        the request matchdict if absent.\n    :type resource_data: dict\n\n    \"\"\"\n    if action == ACTIONS.READ:\n        if not isinstance(data, list):\n            data = [data]\n        impacted = data\n    elif action == ACTIONS.CREATE:\n        impacted = [{\"new\": data}]\n    elif action == ACTIONS.DELETE:\n        if not isinstance(data, list):\n            impacted = [{\"new\": data, \"old\": old}]\n        else:\n            impacted = []\n            for i, new in enumerate(data):\n                impacted.append({\"new\": new, \"old\": old[i]})\n    else:  # ACTIONS.UPDATE:\n        impacted = [{\"new\": data, \"old\": old}]\n\n    # Get previously triggered events.\n    events = request.bound_data.setdefault(\"resource_events\", EventCollector())\n\n    resource_name = resource_name or request.current_resource_name\n    matchdict = resource_data or dict(request.matchdict)\n\n    payload = {\n        \"timestamp\": timestamp,\n        \"action\": action.value,\n        # Deprecated: don't actually use URI (see #945).\n        \"uri\": strip_uri_prefix(request.path),\n        \"user_id\": request.prefixed_userid,\n        \"resource_name\": resource_name,\n    }\n\n    # Deprecated: don't actually use `resource_name_id` either (see #945).\n    if \"id\" in request.matchdict:\n        matchdict[resource_name + \"_id\"] = matchdict.pop(\"id\")\n\n    payload.update(**matchdict)\n\n    events.add_event(resource_name, parent_id, action, payload, impacted, request)\n", "path": "kinto/core/events.py"}]}