CGM_Public/pretix_original
2
0
Fork 1
mirror of https://github.com/pretix/pretix.git synced 2026-08-09 10:37:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Review notes

Browse Source
This commit is contained in:
Raphael Michel
2026-03-09 14:57:31 +01:00
parent 714aecddb2
commit a29261a67c
2 changed files with 27 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pretix/base/permissions.py
+22 -34
View File
@@ -19,11 +19,11 @@
# 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/>.
#
import functools
import logging
import warnings
from collections import OrderedDict
from typing import Dict, List, NamedTuple, Set, Tuple
from typing import Callable, Dict, List, NamedTuple, Set, Tuple
from django.apps import apps
from django.dispatch import receiver
@@ -35,11 +35,22 @@ from pretix.base.signals import (
)
logger = logging.getLogger(__name__)
_ALL_EVENT_PERMISSION_GROUPS = None
_ALL_ORGANIZER_PERMISSION_GROUPS = None
_ALL_EVENT_PERMISSIONS = None
_ALL_ORGANIZER_PERMISSIONS = None
_CACHE_TIME_APPS_READY = None # hack: we need to clear the cache after plugins are loaded during startup
def cache_until_change(input_value: Callable):
def decorator(func):
old_input_value = None
cached_result = None
@functools.wraps(func)
def wrapper():
nonlocal cached_result, old_input_value
if cached_result is None or old_input_value != input_value():
cached_result = func()
old_input_value = input_value()
return cached_result
return wrapper
return decorator
class PermissionOption(NamedTuple):
@@ -56,12 +67,8 @@ class PermissionGroup(NamedTuple):
help_text: str | Promise = None
@cache_until_change(input_value=lambda: apps.ready)
def get_all_event_permission_groups() -> Dict[str, PermissionGroup]:
global _ALL_EVENT_PERMISSION_GROUPS, _CACHE_TIME_APPS_READY
if _ALL_EVENT_PERMISSION_GROUPS and apps.ready == _CACHE_TIME_APPS_READY:
return _ALL_EVENT_PERMISSION_GROUPS
types = OrderedDict()
for recv, ret in register_event_permission_groups.send(None):
if isinstance(ret, (list, tuple)):
@@ -69,17 +76,11 @@ def get_all_event_permission_groups() -> Dict[str, PermissionGroup]:
types[r.name] = r
else:
types[ret.name] = ret
_ALL_EVENT_PERMISSION_GROUPS = types
_CACHE_TIME_APPS_READY = apps.ready
return types
@cache_until_change(input_value=lambda: apps.ready)
def get_all_organizer_permission_groups() -> Dict[str, PermissionGroup]:
global _ALL_ORGANIZER_PERMISSION_GROUPS, _CACHE_TIME_APPS_READY
if _ALL_ORGANIZER_PERMISSION_GROUPS and apps.ready == _CACHE_TIME_APPS_READY:
return _ALL_ORGANIZER_PERMISSION_GROUPS
types = OrderedDict()
for recv, ret in register_organizer_permission_groups.send(None):
if isinstance(ret, (list, tuple)):
@@ -87,42 +88,29 @@ def get_all_organizer_permission_groups() -> Dict[str, PermissionGroup]:
types[r.name] = r
else:
types[ret.name] = ret
_ALL_ORGANIZER_PERMISSION_GROUPS = types
_CACHE_TIME_APPS_READY = apps.ready
return types
@cache_until_change(input_value=lambda: apps.ready)
def get_all_event_permissions() -> Set[str]:
from pretix.helpers.permission_migration import OLD_TO_NEW_EVENT_COMPAT
global _ALL_EVENT_PERMISSIONS, _CACHE_TIME_APPS_READY
if _ALL_EVENT_PERMISSIONS and apps.ready == _CACHE_TIME_APPS_READY:
return _ALL_EVENT_PERMISSIONS
res = set(OLD_TO_NEW_EVENT_COMPAT.keys())
for pg in get_all_event_permission_groups().values():
for a in pg.actions:
res.add(f"{pg.name}:{a}")
_ALL_EVENT_PERMISSIONS = res
_CACHE_TIME_APPS_READY = apps.ready
return res
@cache_until_change(input_value=lambda: apps.ready)
def get_all_organizer_permissions() -> Set[str]:
from pretix.helpers.permission_migration import OLD_TO_NEW_ORGANIZER_COMPAT
global _ALL_ORGANIZER_PERMISSIONS, _CACHE_TIME_APPS_READY
if _ALL_ORGANIZER_PERMISSIONS and apps.ready == _CACHE_TIME_APPS_READY:
return _ALL_ORGANIZER_PERMISSIONS
res = set(OLD_TO_NEW_ORGANIZER_COMPAT.keys())
for pg in get_all_organizer_permission_groups().values():
for a in pg.actions:
res.add(f"{pg.name}:{a}")
_ALL_ORGANIZER_PERMISSIONS = res
_CACHE_TIME_APPS_READY = apps.ready
return res
src/pretix/base/services/export.py
+5 -5
View File
@@ -194,7 +194,7 @@ def init_event_exporter(identifier, **kwargs):
return None
def init_event_exporters(event, user=None, token=None, device=None, request=None, progress_callback=None, staff_session=False):
def init_event_exporters(event, user=None, token=None, device=None, request=None, staff_session=False, **kwargs):
if not user and not token and not device:
raise ValueError("No auth source given.")
perm_holder = device or token or user
@@ -211,7 +211,7 @@ def init_event_exporters(event, user=None, token=None, device=None, request=None
if not perm_holder.has_event_permission(event.organizer, event, permission_name, request) and not staff_session:
continue
exporter: BaseExporter = response(event=event, organizer=event.organizer, progress_callback=progress_callback)
exporter: BaseExporter = response(event=event, organizer=event.organizer, **kwargs)
if not exporter.available_for_user(user if user and user.is_authenticated else None):
continue
@@ -227,7 +227,7 @@ def init_organizer_exporter(identifier, **kwargs):
def init_organizer_exporters(
organizer, user=None, token=None, device=None, request=None, progress_callback=None, staff_session=False, event_qs=None
organizer, user=None, token=None, device=None, request=None, staff_session=False, event_qs=None, **kwargs
):
if not user and not token and not device:
raise ValueError("No auth source given.")
@@ -243,7 +243,7 @@ def init_organizer_exporters(
continue
if issubclass(response, OrganizerLevelExportMixin):
exporter: BaseExporter = response(event=Event.objects.none(), organizer=organizer, progress_callback=progress_callback)
exporter: BaseExporter = response(event=Event.objects.none(), organizer=organizer, **kwargs)
try:
if not perm_holder.has_organizer_permission(organizer, response.get_required_organizer_permission(), request) and not staff_session:
@@ -295,7 +295,7 @@ def init_organizer_exporters(
if not _has_permission_on_any_team_cache[permission_name]:
continue
exporter: BaseExporter = response(event=_event_list_cache[permission_name], organizer=organizer, progress_callback=progress_callback)
exporter: BaseExporter = response(event=_event_list_cache[permission_name], organizer=organizer, **kwargs)
if not exporter.available_for_user(user if user and user.is_authenticated else None):
continue