diff --git a/src/api/serializers.py b/src/api/serializers.py
index 2ac02e23e81d31d04572c295787c421cc36da605..087a3fc22b3d6710493193198cc3394d31874d75 100644
--- a/src/api/serializers.py
+++ b/src/api/serializers.py
@@ -157,7 +157,7 @@ class AssemblySerializer(HubModelSerializer):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
- cluster_qs = Assembly.objects.accessible_by_user(conference=self.conference, user=self.request_user).exclude(hierarchy=Assembly.Hierarchy.REGULAR)
+ cluster_qs = Assembly.objects.associated_with_user(conference=self.conference, user=self.request_user).exclude(hierarchy=Assembly.Hierarchy.REGULAR)
self.fields['parent'].queryset = cluster_qs
diff --git a/src/api/views/mixins.py b/src/api/views/mixins.py
index 9c36dd5e222b261983aa23cfd98d2ebc9e67a14f..8cc93edfcafef1aae7d8a3dc5578cfd40ae7f8a9 100644
--- a/src/api/views/mixins.py
+++ b/src/api/views/mixins.py
@@ -56,7 +56,7 @@ class ConferenceSlugAssemblyMixin(ConferenceSlugMixin):
if self._assembly is None:
assembly_slug = self.request.resolver_match.kwargs['assembly']
try:
- self._assembly = Assembly.objects.accessible_by_user(self.request.user, self.conference).get(slug=assembly_slug)
+ self._assembly = Assembly.objects.associated_with_user(self.conference, user=self.request.user).get(slug=assembly_slug)
except Assembly.DoesNotExist:
if issuing_token := self.kwargs.get('issuing_token', None):
try:
diff --git a/src/backoffice/views/assemblies.py b/src/backoffice/views/assemblies.py
index 5d0e94b96cd7318fdcef0ac29f327012228ae2cd..bae1782e3d2ec3300503a5cc088ed2bc034c9499 100644
--- a/src/backoffice/views/assemblies.py
+++ b/src/backoffice/views/assemblies.py
@@ -448,7 +448,7 @@ class AssemblyEditChildrenView(AssemblyMixin, View):
def get(self, *args, **kwargs):
candidates_qs = (
- Assembly.objects.accessible_by_user(conference=self.conference, user=self.request.user)
+ Assembly.objects.associated_with_user(conference=self.conference, user=self.request.user)
.filter(hierarchy=Assembly.Hierarchy.REGULAR, parent=None)
.exclude(state_assembly__in=[Assembly.State.NONE, Assembly.State.REJECTED, Assembly.State.HIDDEN, Assembly.State.PLANNED])
)
@@ -481,7 +481,7 @@ class AssemblyEditLinksView(AssemblyMixin, View):
if add_id is not None:
try:
- linkee = Assembly.objects.accessible_by_user(user=request.user, conference=self.conference).get(pk=add_id)
+ linkee = Assembly.objects.associated_with_user(user=request.user, conference=self.conference).get(pk=add_id)
except Assembly.DoesNotExist:
messages.warning('404 +Assembly %s', add_id)
linkee = None
@@ -502,7 +502,7 @@ class AssemblyEditLinksView(AssemblyMixin, View):
if remove_id is not None:
try:
- linkee = Assembly.objects.accessible_by_user(user=request.user, conference=self.conference).get(pk=remove_id)
+ linkee = Assembly.objects.associated_with_user(user=request.user, conference=self.conference).get(pk=remove_id)
except Assembly.DoesNotExist:
messages.warning('404 -Assembly %s', remove_id)
linkee = None
@@ -524,7 +524,7 @@ class AssemblyEditLinksView(AssemblyMixin, View):
def get(self, *args, **kwargs):
candidates_qs = (
- Assembly.objects.accessible_by_user(conference=self.conference, user=self.request.user)
+ Assembly.objects.associated_with_user(conference=self.conference, user=self.request.user)
.filter(hierarchy=Assembly.Hierarchy.REGULAR)
.exclude(state_assembly__in=[Assembly.State.NONE, Assembly.State.PLANNED, Assembly.State.HIDDEN, Assembly.State.REJECTED])
)
diff --git a/src/backoffice/views/assemblyteam.py b/src/backoffice/views/assemblyteam.py
index 4b29353b1cc69dd24ce697552438484638f2754f..2be313cf94632415f30a5adc0076864515b1af20 100644
--- a/src/backoffice/views/assemblyteam.py
+++ b/src/backoffice/views/assemblyteam.py
@@ -122,8 +122,7 @@ class AssembliesListMixin(AssemblyTeamMixin):
# get the mode
mode = (self.request.POST if self.request.method == 'POST' else self.request.GET).get('mode', self.default_assemblies_mode)
- # not using .accessible_by_user() here as we're the Assembly Team anyway
- qs = Assembly.objects.filter(conference=self.conference)
+ qs = Assembly.objects.associated_with_user(conference=self.conference, user=self.request.user, staff_can_see=True)
if mode == 'not_selected':
pass
elif self.active_page == 'assemblies':
diff --git a/src/core/models/assemblies.py b/src/core/models/assemblies.py
index 7e8139efa490eb36edf4637b6892b26b421b71d9..79921d45d3d1849922983b439ade04cb7b59f678 100644
--- a/src/core/models/assemblies.py
+++ b/src/core/models/assemblies.py
@@ -19,6 +19,7 @@ from django.utils.safestring import mark_safe
from django.utils.translation import gettext
from django.utils.translation import gettext_lazy as _
+from core.models.base_managers import ConferenceManagerMixin
from core.models.tags import TagItem
from core.validators import FileSizeValidator, ImageDimensionValidator
@@ -31,85 +32,28 @@ from .tags import TaggedItemMixin
from .users import PlatformUser
-class AssemblyManager(models.Manager):
- def accessible_by_user(self, user: PlatformUser, conference: Conference):
- assert user is not None
+class AssemblyManager(ConferenceManagerMixin['Assembly']):
+ staff_permissions = ['assembly_team']
+ assembly_filter = 'self'
- qs = self.get_queryset()
- if conference is not None:
- qs = qs.filter(conference=conference)
-
- # crude hack because Django makes it that hard to customize AnonymousUser
- if user is None or not user.is_authenticated:
- user = PlatformUser.get_anonymous_user()
-
- # see if we can return all assemblies (i.e. the user is marked as "staff")
- if user.is_authenticated:
- try:
- member = ConferenceMember.objects.get(conference=conference, user=user)
- if member.is_staff:
- return qs
- except ConferenceMember.DoesNotExist:
- pass
-
- # for everybody else, only show "public" assemblies
- return qs.filter(state_assembly__in=self.model.PUBLIC_STATES)
-
- def associated_to_user(self, user: PlatformUser, conference: Conference):
- assert user is not None
+ def apply_public_filter(self, queryset: 'QuerySet[Assembly]', member: ConferenceMember | None = None) -> 'QuerySet[Assembly]':
+ return queryset.filter(Q(state_assembly__in=self.model.PUBLIC_STATES) | Q(state_channel__in=self.model.PUBLIC_STATES))
+ def associated_to_user(self, user: PlatformUser, conference: Conference) -> 'QuerySet[Assembly]':
# guests cannot manage anything
if user is None or not user.is_authenticated:
return Assembly.objects.none()
-
- # prepare query
- qs = self.get_queryset()
- if conference is not None:
- qs = qs.filter(conference=conference)
-
- # lookup via ConferenceMember
- qs = qs.filter(members__member=user)
-
- # but don't show "hidden" ones to the user
- qs = qs.exclude(state_assembly__in=[Assembly.State.NONE], state_channel__in=[Assembly.State.NONE])
-
- # finally return the resulting QuerySet
- return qs
-
- def conference_accessible(self, conference: Conference):
return (
- self.get_queryset()
- .filter(conference=conference)
- .filter(Q(state_assembly__in=self.model.PUBLIC_STATES) | Q(state_channel__in=self.model.PUBLIC_STATES))
+ self.associated_with_user(conference, user=user)
+ .filter(members__member=user)
+ .exclude(
+ Q(
+ state_assembly__in=[Assembly.State.NONE, Assembly.State.HIDDEN],
+ state_channel__in=[Assembly.State.NONE, Assembly.State.HIDDEN],
+ )
+ )
)
- def manageable_by_user(self, user: PlatformUser, conference: Conference, staff_can_manage=True):
- assert user is not None
-
- # guests cannot manage anything
- if user is None or not user.is_authenticated:
- return Assembly.objects.none()
-
- # prepare query
- qs = self.get_queryset()
- if conference is not None:
- qs = qs.filter(conference=conference)
-
- # see if we can return all assemblies (i.e. the user is marked as "staff")
- if staff_can_manage:
- try:
- member = ConferenceMember.objects.get(conference=conference, user=user)
- if member.is_staff and member.has_perm('core.assembly_team'):
- return qs
- except ConferenceMember.DoesNotExist:
- pass
-
- # lookup via AssemblyMember
- qs = qs.filter(members__member=user, members__can_manage_assembly=True)
-
- # finally return the resulting QuerySet
- return qs
-
def get_banner_file_name(instance: 'Assembly', filename: str):
return str(Path(str(instance.id)).joinpath('banner', filename))
diff --git a/src/core/models/badges.py b/src/core/models/badges.py
index 345845c0f7211c73f32be24b07973ceede59e269..07e949411f6f6c18ecc71e7ee565237d638b058e 100644
--- a/src/core/models/badges.py
+++ b/src/core/models/badges.py
@@ -77,7 +77,7 @@ class BadgeManager(ConferenceManagerMixin['Badge']):
if not user.is_authenticated:
return qs.filter(state=Badge.State.PUBLIC)
- manageable = Assembly.objects.manageable_by_user(user, conference, staff_can_manage)
+ manageable = Assembly.objects.manageable_by_user(conference, user=user, staff_can_manage=staff_can_manage)
return qs.filter(Q(state=Badge.State.PUBLIC) | Q(users__user=user) | Q(issuing_assembly__in=manageable))
diff --git a/src/core/models/base_managers.py b/src/core/models/base_managers.py
index b53bcc16c7650d432a0208d1fe376f31e560fa7f..7afc9a73b718685c7a61aa195b78841ae0e838c6 100644
--- a/src/core/models/base_managers.py
+++ b/src/core/models/base_managers.py
@@ -5,7 +5,6 @@ from django.core.exceptions import ImproperlyConfigured
from django.db import models
from django.db.models import Model, QuerySet
-from core.models.assemblies import Assembly, AssemblyMember
from core.models.conference import ConferenceMember
if TYPE_CHECKING:
@@ -75,6 +74,9 @@ class ConferenceManagerMixin(models.Manager, Generic[_ModelType]):
else:
prefix = f'{self.assembly_filter}__'
queryset = queryset.select_related(self.assembly_filter)
+
+ from core.models.assemblies import Assembly, AssemblyMember # pylint: disable=C0415 # Circular import avoidance
+
allowed_members = AssemblyMember.objects.filter(member=member.user)
allowed_members = allowed_members.filter(can_manage_assembly=True) if only_manageable else allowed_members
queryset = queryset.filter(**{f'{prefix}members__in': allowed_members}).exclude(**{f'{prefix}state_assembly__in': [Assembly.State.HIDDEN]})