CGM_Public/pretix_cgo
1
0
Fork 0
forked from CGM_Public/pretix_original
Code Pull Requests Actions Activity

LogEntryType registry (#4235)

Browse Source 
Move display of LogEntry details from the `logentry_display` and 
`logentry_object_link` signals to a class hierarchy based approach. 
For each action_type, an instance of a subclass of `LogEntryType` 
is registered in the `log_entry_types` registry.

Analogous to EventPluginSignal, this registry is an `EventPluginRegistry`, 
so it keeps track of the plugin the LogEntryType is defined in.

---------

Co-authored-by: Raphael Michel <michel@rami.io>
Co-authored-by: Richard Schreiber <schreiber@rami.io>
This commit is contained in:
Mira
2025-01-16 13:05:57 +01:00
committed by GitHub
parent c25d6988a7
commit c8d4815c9e
16 changed files with 1298 additions and 665 deletions
Download Patch File Download Diff File Expand all files Collapse all files

253
src/pretix/base/logentrytypes.py Normal file
View File

@@ -0,0 +1,253 @@
#
# This file is part of pretix (Community Edition).
#
# Copyright (C) 2014-2020 Raphael Michel and contributors
# Copyright (C) 2020-2021 rami.io GmbH and contributors
#
# This program is free software: you can redistribute it and/or modify it under the terms of the GNU Affero General
# Public License as published by the Free Software Foundation in version 3 of the License.
#
# ADDITIONAL TERMS APPLY: Pursuant to Section 7 of the GNU Affero General Public License, additional terms are
# applicable granting you additional permissions and placing additional restrictions on your usage of this software.
# Please refer to the pretix LICENSE file to obtain the full terms applicable to this work. If you did not receive
# this file, see <https://pretix.eu/about/en/license>.
#
# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
# details.
#
# You should have received a copy of the GNU Affero General Public License along with this program. If not, see
# <https://www.gnu.org/licenses/>.
#
from collections import defaultdict
from django.urls import reverse
from django.utils.html import escape
from django.utils.translation import gettext_lazy as _, pgettext_lazy
from pretix.base.signals import EventPluginRegistry
def make_link(a_map, wrapper, is_active=True, event=None, plugin_name=None):
if a_map:
if is_active:
a_map['val'] = '<a href="{href}">{val}</a>'.format_map(a_map)
elif event and plugin_name:
a_map['val'] = (
'<i>{val}</i> <a href="{plugin_href}">'
'<span data-toggle="tooltip" title="{errmes}" class="fa fa-warning fa-fw"></span></a>'
).format_map({
**a_map,
"errmes": _("The relevant plugin is currently not active. To activate it, click here to go to the plugin settings."),
"plugin_href": reverse('control:event.settings.plugins', kwargs={
'organizer': event.organizer.slug,
'event': event.slug,
}) + '#plugin_' + plugin_name,
})
else:
a_map['val'] = '<i>{val}</i> <span data-toggle="tooltip" title="{errmes}" class="fa fa-warning fa-fw"></span>'.format_map({
**a_map,
"errmes": _("The relevant plugin is currently not active."),
})
return wrapper.format_map(a_map)
class LogEntryTypeRegistry(EventPluginRegistry):
def __init__(self):
super().__init__({'action_type': lambda o: getattr(o, 'action_type')})
def register(self, *objs):
for obj in objs:
if not isinstance(obj, LogEntryType):
raise TypeError('Entries must be derived from LogEntryType')
if obj.__module__ == LogEntryType.__module__:
raise TypeError('Must not register base classes, only derived ones')
return super().register(*objs)
def new_from_dict(self, data):
"""
Register multiple instance of a `LogEntryType` class with different `action_type`
and plain text strings, as given by the items of the specified data dictionary.
This method is designed to be used as a decorator as follows:
.. code-block:: python
@log_entry_types.new_from_dict({
'pretix.event.item.added': _('The product has been created.'),
'pretix.event.item.changed': _('The product has been changed.'),
# ...
})
class CoreItemLogEntryType(ItemLogEntryType):
# ...
:param data: action types and descriptions
``{"some_action_type": "Plain text description", ...}``
"""
def reg(clz):
for action_type, plain in data.items():
self.register(clz(action_type=action_type, plain=plain))
return clz
return reg
"""
Registry for LogEntry types.
Each entry in this registry should be an instance of a subclass of ``LogEntryType``.
They are annotated with their ``action_type`` and the defining ``plugin``.
"""
log_entry_types = LogEntryTypeRegistry()
class LogEntryType:
"""
Base class for a type of LogEntry, identified by its action_type.
"""
def __init__(self, action_type=None, plain=None):
if action_type:
self.action_type = action_type
if plain:
self.plain = plain
def display(self, logentry):
"""
Returns the message to be displayed for a given logentry of this type.
:return: `str` or `LazyI18nString`
"""
if hasattr(self, 'plain'):
plain = str(self.plain)
if '{' in plain:
data = defaultdict(lambda: '?', logentry.parsed_data)
return plain.format_map(data)
else:
return plain
def get_object_link_info(self, logentry) -> dict:
"""
Return information to generate a link to the `content_object` of a given log entry.
Not implemented in the base class, causing the object link to be omitted.
:return: Dictionary with the keys ``href`` (containing a URL to view/edit the object) and ``val`` (containing the
escaped text for the anchor element)
"""
pass
def get_object_link(self, logentry):
a_map = self.get_object_link_info(logentry)
return make_link(a_map, self.object_link_wrapper)
object_link_wrapper = '{val}'
def shred_pii(self, logentry):
"""
To be used for shredding personally identified information contained in the data field of a LogEntry of this
type.
"""
raise NotImplementedError
class EventLogEntryType(LogEntryType):
"""
Base class for any `LogEntry` type whose `content_object` is either an `Event` itself or belongs to a specific `Event`.
"""
def get_object_link_info(self, logentry) -> dict:
if hasattr(self, 'object_link_viewname') and logentry.content_object:
return {
'href': reverse(self.object_link_viewname, kwargs={
'event': logentry.event.slug,
'organizer': logentry.event.organizer.slug,
**self.object_link_args(logentry.content_object),
}),
'val': escape(self.object_link_display_name(logentry.content_object)),
}
def object_link_args(self, content_object):
"""Return the kwargs for the url used in a link to content_object."""
if hasattr(self, 'object_link_argname'):
return {self.object_link_argname: content_object.pk}
return {}
def object_link_display_name(self, content_object):
"""Return the display name to refer to content_object in the user interface."""
return str(content_object)
class OrderLogEntryType(EventLogEntryType):
object_link_wrapper = _('Order {val}')
object_link_viewname = 'control:event.order'
def object_link_args(self, order):
return {'code': order.code}
def object_link_display_name(self, order):
return order.code
class VoucherLogEntryType(EventLogEntryType):
object_link_wrapper = _('Voucher {val}…')
object_link_viewname = 'control:event.voucher'
object_link_argname = 'voucher'
def object_link_display_name(self, voucher):
if len(voucher.code) > 6:
return voucher.code[:6] + "…"
return voucher.code
class ItemLogEntryType(EventLogEntryType):
object_link_wrapper = _('Product {val}')
object_link_viewname = 'control:event.item'
object_link_argname = 'item'
class SubEventLogEntryType(EventLogEntryType):
object_link_wrapper = pgettext_lazy('subevent', 'Date {val}')
object_link_viewname = 'control:event.subevent'
object_link_argname = 'subevent'
class QuotaLogEntryType(EventLogEntryType):
object_link_wrapper = _('Quota {val}')
object_link_viewname = 'control:event.items.quotas.show'
object_link_argname = 'quota'
class DiscountLogEntryType(EventLogEntryType):
object_link_wrapper = _('Discount {val}')
object_link_viewname = 'control:event.items.discounts.edit'
object_link_argname = 'discount'
class ItemCategoryLogEntryType(EventLogEntryType):
object_link_wrapper = _('Category {val}')
object_link_viewname = 'control:event.items.categories.edit'
object_link_argname = 'category'
class QuestionLogEntryType(EventLogEntryType):
object_link_wrapper = _('Question {val}')
object_link_viewname = 'control:event.items.questions.show'
object_link_argname = 'question'
class TaxRuleLogEntryType(EventLogEntryType):
object_link_wrapper = _('Tax rule {val}')
object_link_viewname = 'control:event.settings.tax.edit'
object_link_argname = 'rule'
class NoOpShredderMixin:
def shred_pii(self, logentry):
pass
class ClearDataShredderMixin:
def shred_pii(self, logentry):
logentry.data = None

132
src/pretix/base/models/log.py
View File

@@ -33,16 +33,15 @@
# License for the specific language governing permissions and limitations under the License.
import json
import logging
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.db import models
from django.urls import reverse
from django.utils.functional import cached_property
from django.utils.html import escape
from django.utils.translation import gettext_lazy as _, pgettext_lazy
from pretix.base.signals import logentry_object_link
from pretix.base.logentrytypes import log_entry_types, make_link
from pretix.base.signals import is_app_active, logentry_object_link
class VisibleOnlyManager(models.Manager):
@@ -92,6 +91,10 @@ class LogEntry(models.Model):
indexes = [models.Index(fields=["datetime", "id"])]
def display(self):
log_entry_type, meta = log_entry_types.get(action_type=self.action_type)
if log_entry_type:
return log_entry_type.display(self)
from ..signals import logentry_display
for receiver, response in logentry_display.send(self.event, logentry=self):
@@ -126,10 +129,18 @@ class LogEntry(models.Model):
@cached_property
def display_object(self):
from . import (
Discount, Event, Item, ItemCategory, Order, Question, Quota,
SubEvent, TaxRule, Voucher,
Discount, Event, Item, Order, Question, Quota, SubEvent, Voucher,
)
log_entry_type, meta = log_entry_types.get(action_type=self.action_type)
if log_entry_type:
link_info = log_entry_type.get_object_link_info(self)
if is_app_active(self.event, meta['plugin']):
return make_link(link_info, log_entry_type.object_link_wrapper)
else:
return make_link(link_info, log_entry_type.object_link_wrapper, is_active=False,
event=self.event, plugin_name=meta['plugin'] and getattr(meta['plugin'], 'name'))
try:
if self.content_type.model_class() is Event:
return ''
@@ -137,110 +148,15 @@ class LogEntry(models.Model):
co = self.content_object
except:
return ''
a_map = None
a_text = None
if isinstance(co, Order):
a_text = _('Order {val}')
a_map = {
'href': reverse('control:event.order', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'code': co.code
}),
'val': escape(co.code),
}
elif isinstance(co, Voucher):
a_text = _('Voucher {val}…')
a_map = {
'href': reverse('control:event.voucher', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'voucher': co.id
}),
'val': escape(co.code[:6]),
}
elif isinstance(co, Item):
a_text = _('Product {val}')
a_map = {
'href': reverse('control:event.item', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'item': co.id
}),
'val': escape(co.name),
}
elif isinstance(co, SubEvent):
a_text = pgettext_lazy('subevent', 'Date {val}')
a_map = {
'href': reverse('control:event.subevent', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'subevent': co.id
}),
'val': escape(str(co))
}
elif isinstance(co, Quota):
a_text = _('Quota {val}')
a_map = {
'href': reverse('control:event.items.quotas.show', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'quota': co.id
}),
'val': escape(co.name),
}
elif isinstance(co, Discount):
a_text = _('Discount {val}')
a_map = {
'href': reverse('control:event.items.discounts.edit', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'discount': co.id
}),
'val': escape(co.internal_name),
}
elif isinstance(co, ItemCategory):
a_text = _('Category {val}')
a_map = {
'href': reverse('control:event.items.categories.edit', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'category': co.id
}),
'val': escape(co.name),
}
elif isinstance(co, Question):
a_text = _('Question {val}')
a_map = {
'href': reverse('control:event.items.questions.show', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'question': co.id
}),
'val': escape(co.question),
}
elif isinstance(co, TaxRule):
a_text = _('Tax rule {val}')
a_map = {
'href': reverse('control:event.settings.tax.edit', kwargs={
'event': self.event.slug,
'organizer': self.event.organizer.slug,
'rule': co.id
}),
'val': escape(co.name),
}
for receiver, response in logentry_object_link.send(self.event, logentry=self):
if response:
return response
if a_text and a_map:
a_map['val'] = '<a href="{href}">{val}</a>'.format_map(a_map)
return a_text.format_map(a_map)
elif a_text:
return a_text
else:
for receiver, response in logentry_object_link.send(self.event, logentry=self):
if response:
return response
return ''
if isinstance(co, (Order, Voucher, Item, SubEvent, Quota, Discount, Question)):
logging.warning("LogEntryType missing or ill-defined: %s", self.action_type)
return ''
@cached_property
def parsed_data(self):

226
src/pretix/base/signals.py
View File

@@ -52,6 +52,50 @@ def _populate_app_cache():
app_cache[ac.name] = ac
def get_defining_app(o):
# If sentry packed this in a wrapper, unpack that
if "sentry" in o.__module__:
o = o.__wrapped__
# Find the Django application this belongs to
searchpath = o.__module__
# Core modules are always active
if any(searchpath.startswith(cm) for cm in settings.CORE_MODULES):
return 'CORE'
if not app_cache:
_populate_app_cache()
while True:
app = app_cache.get(searchpath)
if "." not in searchpath or app:
break
searchpath, _ = searchpath.rsplit(".", 1)
return app
def is_app_active(sender, app):
if app == 'CORE':
return True
excluded = settings.PRETIX_PLUGINS_EXCLUDE
if sender and app and app.name in sender.get_plugins() and app.name not in excluded:
if not hasattr(app, 'compatibility_errors') or not app.compatibility_errors:
return True
return False
def is_receiver_active(sender, receiver):
if sender is None:
# Send to all events!
return True
app = get_defining_app(receiver)
return is_app_active(sender, app)
class EventPluginSignal(django.dispatch.Signal):
"""
This is an extension to Django's built-in signals which differs in a way that it sends
@@ -59,33 +103,6 @@ class EventPluginSignal(django.dispatch.Signal):
Event.
"""
def _is_active(self, sender, receiver):
if sender is None:
# Send to all events!
return True
# If sentry packed this in a wrapper, unpack that
if "sentry" in receiver.__module__:
receiver = receiver.__wrapped__
# Find the Django application this belongs to
searchpath = receiver.__module__
core_module = any([searchpath.startswith(cm) for cm in settings.CORE_MODULES])
app = None
if not core_module:
while True:
app = app_cache.get(searchpath)
if "." not in searchpath or app:
break
searchpath, _ = searchpath.rsplit(".", 1)
# Only fire receivers from active plugins and core modules
excluded = settings.PRETIX_PLUGINS_EXCLUDE
if core_module or (sender and app and app.name in sender.get_plugins() and app.name not in excluded):
if not hasattr(app, 'compatibility_errors') or not app.compatibility_errors:
return True
return False
def send(self, sender: Event, **named) -> List[Tuple[Callable, Any]]:
"""
Send signal from sender to all connected receivers that belong to
@@ -104,7 +121,7 @@ class EventPluginSignal(django.dispatch.Signal):
_populate_app_cache()
for receiver in self._sorted_receivers(sender):
if self._is_active(sender, receiver):
if is_receiver_active(sender, receiver):
response = receiver(signal=self, sender=sender, **named)
responses.append((receiver, response))
return responses
@@ -128,7 +145,7 @@ class EventPluginSignal(django.dispatch.Signal):
_populate_app_cache()
for receiver in self._sorted_receivers(sender):
if self._is_active(sender, receiver):
if is_receiver_active(sender, receiver):
named[chain_kwarg_name] = response
response = receiver(signal=self, sender=sender, **named)
return response
@@ -155,7 +172,7 @@ class EventPluginSignal(django.dispatch.Signal):
_populate_app_cache()
for receiver in self._sorted_receivers(sender):
if self._is_active(sender, receiver):
if is_receiver_active(sender, receiver):
try:
response = receiver(signal=self, sender=sender, **named)
except Exception as err:
@@ -202,6 +219,122 @@ class DeprecatedSignal(django.dispatch.Signal):
super().connect(receiver, sender=None, weak=True, dispatch_uid=None)
class Registry:
"""
A Registry is a collection of objects (entries), annotated with metadata. Entries can be searched and filtered by
metadata keys, and metadata is returned as part of the result.
Entry metadata is generated during registration using to the accessor functions given to the Registry
constructor.
Example:
.. code-block:: python
animal_sound_registry = Registry({"animal": lambda s: s.animal})
@animal_sound_registry.new("dog", "woof")
@animal_sound_registry.new("cricket", "chirp")
class AnimalSound:
def __init__(self, animal, sound):
self.animal = animal
self.sound = sound
def make_sound(self):
return self.sound
@animal_sound_registry.new()
class CatSound(AnimalSound):
def __init__(self):
super().__init__(animal="cat", sound=["meow", "meww", "miaou"])
def make_sound(self):
return random.choice(self.sound)
"""
def __init__(self, keys):
"""
:param keys: Dictionary with `{key: accessor_function}`
When a new entry is registered, all accessor functions are called with the new entry as parameter.
Their return value is stored as the metadata value for that key.
"""
self.registered_entries = dict()
self.keys = keys
self.by_key = {key: {} for key in self.keys.keys()}
def register(self, *objs):
"""
Register one or more entries in this registry.
Usable as a regular method or as decorator on a class or function. If used on a class, the class type object
itself is registered, not an instance of the class. To register an instance, use the ``new`` method.
.. code-block:: python
@some_registry.register
def my_new_entry(foo):
# ...
"""
for obj in objs:
if obj in self.registered_entries:
raise RuntimeError('Object already registered: {}'.format(obj))
meta = {k: accessor(obj) for k, accessor in self.keys.items()}
tup = (obj, meta)
for key, value in meta.items():
self.by_key[key][value] = tup
self.registered_entries[obj] = meta
if len(objs) == 1:
return objs[0]
def new(self, *args, **kwargs):
"""
Instantiate the decorated class with the given `*args` and `**kwargs`, and register the instance in this registry.
May be used multiple times.
.. code-block:: python
@animal_sound_registry.new("meow")
@animal_sound_registry.new("woof")
class AnimalSound:
def __init__(self, sound):
# ...
"""
def reg(clz):
obj = clz(*args, **kwargs)
self.register(obj)
return clz
return reg
def get(self, **kwargs):
(key, value), = kwargs.items()
return self.by_key.get(key).get(value, (None, None))
def filter(self, **kwargs):
return (
(entry, meta)
for entry, meta in self.registered_entries.items()
if all(value == meta[key] for key, value in kwargs.items())
)
class EventPluginRegistry(Registry):
"""
A Registry which automatically annotates entries with a "plugin" key, specifying which plugin
the entry is defined in. This allows the consumer of entries to determine whether an entry is
enabled for a given event, or filter only for entries defined by enabled plugins.
.. code-block:: python
logtype, meta = my_registry.find(action_type="foo.bar.baz")
# meta["plugin"] contains the django app name of the defining plugin
"""
def __init__(self, keys):
super().__init__({"plugin": lambda o: get_defining_app(o), **keys})
event_live_issues = EventPluginSignal()
"""
This signal is sent out to determine whether an event can be taken live. If you want to
@@ -507,41 +640,16 @@ logentry_display = EventPluginSignal()
"""
Arguments: ``logentry``
To display an instance of the ``LogEntry`` model to a human user,
``pretix.base.signals.logentry_display`` will be sent out with a ``logentry`` argument.
The first received response that is not ``None`` will be used to display the log entry
to the user. The receivers are expected to return plain text.
As with all event-plugin signals, the ``sender`` keyword argument will contain the event.
**DEPRECTATION:** Please do not use this signal for new LogEntry types. Use the log_entry_types
registry instead, as described in https://docs.pretix.eu/en/latest/development/implementation/logging.html
"""
logentry_object_link = EventPluginSignal()
"""
Arguments: ``logentry``
To display the relationship of an instance of the ``LogEntry`` model to another model
to a human user, ``pretix.base.signals.logentry_object_link`` will be sent out with a
``logentry`` argument.
The first received response that is not ``None`` will be used to display the related object
to the user. The receivers are expected to return a HTML link. The internal implementation
builds the links like this::
a_text = _('Tax rule {val}')
a_map = {
'href': reverse('control:event.settings.tax.edit', kwargs={
'event': sender.slug,
'organizer': sender.organizer.slug,
'rule': logentry.content_object.id
}),
'val': escape(logentry.content_object.name),
}
a_map['val'] = '<a href="{href}">{val}</a>'.format_map(a_map)
return a_text.format_map(a_map)
Make sure that any user content in the HTML code you return is properly escaped!
As with all event-plugin signals, the ``sender`` keyword argument will contain the event.
**DEPRECTATION:** Please do not use this signal for new LogEntry types. Use the log_entry_types
registry instead, as described in https://docs.pretix.eu/en/latest/development/implementation/logging.html
"""
requiredaction_display = EventPluginSignal()