From b0dc18cc3ff4670e13f5fa18f6fdd3a15b6e1db2 Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Tue, 12 Nov 2024 14:32:07 -0700
Subject: [PATCH 01/45] base
---
src/registrar/utility/csv_export.py | 147 ++++++++++++++++++----------
1 file changed, 98 insertions(+), 49 deletions(-)
diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py
index 64d960337..e90b27c29 100644
--- a/src/registrar/utility/csv_export.py
+++ b/src/registrar/utility/csv_export.py
@@ -15,6 +15,7 @@
from django.utils import timezone
from django.db.models.functions import Concat, Coalesce
from django.contrib.postgres.aggregates import StringAgg
+from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.generic_helper import convert_queryset_to_dict
from registrar.templatetags.custom_filters import get_region
from registrar.utility.constants import BranchChoices
@@ -50,11 +51,7 @@ def format_end_date(end_date):
return timezone.make_aware(datetime.strptime(end_date, "%Y-%m-%d")) if end_date else get_default_end_date()
-class BaseExport(ABC):
- """
- A generic class for exporting data which returns a csv file for the given model.
- Base class in an inheritance tree of 3.
- """
+class BaseModelDict(ABC):
@classmethod
@abstractmethod
@@ -65,13 +62,6 @@ def model(self):
"""
pass
- @classmethod
- def get_columns(cls):
- """
- Returns the columns for CSV export. Override in subclasses as needed.
- """
- return []
-
@classmethod
def get_sort_fields(cls):
"""
@@ -116,7 +106,7 @@ def get_filter_conditions(cls, **export_kwargs):
return Q()
@classmethod
- def get_computed_fields(cls):
+ def get_annotated_fields(cls):
"""
Get a dict of computed fields. These are fields that do not exist on the model normally
and will be passed to .annotate() when building a queryset.
@@ -136,25 +126,10 @@ def get_related_table_fields(cls):
Get a list of fields from related tables.
"""
return []
-
- @classmethod
- def update_queryset(cls, queryset, **kwargs):
- """
- Returns an updated queryset. Override in subclass to update queryset.
- """
- return queryset
-
- @classmethod
- def write_csv_before(cls, csv_writer, **export_kwargs):
- """
- Write to csv file before the write_csv method.
- Override in subclasses where needed.
- """
- pass
-
+
@classmethod
def annotate_and_retrieve_fields(
- cls, initial_queryset, computed_fields, related_table_fields=None, include_many_to_many=False, **kwargs
+ cls, initial_queryset, annotated_fields, related_table_fields=None, include_many_to_many=False, **kwargs
) -> QuerySet:
"""
Applies annotations to a queryset and retrieves specified fields,
@@ -162,7 +137,7 @@ def annotate_and_retrieve_fields(
Parameters:
initial_queryset (QuerySet): Initial queryset.
- computed_fields (dict, optional): Fields to compute {field_name: expression}.
+ annotated_fields (dict, optional): Fields to compute {field_name: expression}.
related_table_fields (list, optional): Extra fields to retrieve; defaults to annotation keys if None.
include_many_to_many (bool, optional): Determines if we should include many to many fields or not
**kwargs: Additional keyword arguments for specific parameters (e.g., public_contacts, domain_invitations,
@@ -176,8 +151,8 @@ def annotate_and_retrieve_fields(
# We can infer that if we're passing in annotations,
# we want to grab the result of said annotation.
- if computed_fields:
- related_table_fields.extend(computed_fields.keys())
+ if annotated_fields:
+ related_table_fields.extend(annotated_fields.keys())
# Get prexisting fields on the model
model_fields = set()
@@ -187,31 +162,33 @@ def annotate_and_retrieve_fields(
if many_to_many or not isinstance(field, ManyToManyField):
model_fields.add(field.name)
- queryset = initial_queryset.annotate(**computed_fields).values(*model_fields, *related_table_fields)
+ queryset = initial_queryset.annotate(**annotated_fields).values(*model_fields, *related_table_fields)
return cls.update_queryset(queryset, **kwargs)
@classmethod
- def export_data_to_csv(cls, csv_file, **export_kwargs):
+ def update_queryset(cls, queryset, **kwargs):
"""
- All domain metadata:
- Exports domains of all statuses plus domain managers.
+ Returns an updated queryset. Override in subclass to update queryset.
"""
- writer = csv.writer(csv_file)
- columns = cls.get_columns()
+ return queryset
+
+ @classmethod
+ def get_model_dict(cls):
sort_fields = cls.get_sort_fields()
kwargs = cls.get_additional_args()
select_related = cls.get_select_related()
prefetch_related = cls.get_prefetch_related()
exclusions = cls.get_exclusions()
annotations_for_sort = cls.get_annotations_for_sort()
- filter_conditions = cls.get_filter_conditions(**export_kwargs)
- computed_fields = cls.get_computed_fields()
+ filter_conditions = cls.get_filter_conditions(**kwargs)
+ annotated_fields = cls.get_annotated_fields()
related_table_fields = cls.get_related_table_fields()
model_queryset = (
cls.model()
- .objects.select_related(*select_related)
+ .objects
+ .select_related(*select_related)
.prefetch_related(*prefetch_related)
.filter(filter_conditions)
.exclude(exclusions)
@@ -219,13 +196,85 @@ def export_data_to_csv(cls, csv_file, **export_kwargs):
.order_by(*sort_fields)
.distinct()
)
-
- # Convert the queryset to a dictionary (including annotated fields)
annotated_queryset = cls.annotate_and_retrieve_fields(
- model_queryset, computed_fields, related_table_fields, **kwargs
+ model_queryset, annotated_fields, related_table_fields, **kwargs
)
models_dict = convert_queryset_to_dict(annotated_queryset, is_model=False)
+ return models_dict
+
+
+class UserPortfolioPermissionModelDict(BaseModelDict):
+
+ @classmethod
+ def model(cls):
+ # Return the model class that this export handles
+ return UserPortfolioPermission
+
+ @classmethod
+ def get_filter_conditions(cls, **export_kwargs):
+ """
+ Get a Q object of filter conditions to filter when building queryset.
+ """
+ return Q()
+
+ @classmethod
+ def get_exclusions(cls):
+ """
+ Get a Q object of exclusion conditions to pass to .exclude() when building queryset.
+ """
+ return Q()
+
+ @classmethod
+ def get_annotated_fields(cls):
+ """
+ Get a dict of computed fields. These are fields that do not exist on the model normally
+ and will be passed to .annotate() when building a queryset.
+ """
+ return {}
+
+ @classmethod
+ def parse_row(cls, columns, model):
+ """
+ Given a set of columns and a model dictionary, generate a new row from cleaned column data.
+ """
+ FIELDS = {"Not yet defined": "Not yet defined"}
+
+ row = [FIELDS.get(column, "") for column in columns]
+ return row
+
+
+class BaseExport(BaseModelDict):
+ """
+ A generic class for exporting data which returns a csv file for the given model.
+ Base class in an inheritance tree of 3.
+ """
+
+ @classmethod
+ def get_columns(cls):
+ """
+ Returns the columns for CSV export. Override in subclasses as needed.
+ """
+ return []
+
+ @classmethod
+ def write_csv_before(cls, csv_writer, **export_kwargs):
+ """
+ Write to csv file before the write_csv method.
+ Override in subclasses where needed.
+ """
+ pass
+
+ @classmethod
+ def export_data_to_csv(cls, csv_file, **export_kwargs):
+ """
+ All domain metadata:
+ Exports domains of all statuses plus domain managers.
+ """
+ writer = csv.writer(csv_file)
+ columns = cls.get_columns()
+ models_dict = cls.get_model_dict()
+
# Write to csv file before the write_csv
cls.write_csv_before(writer, **export_kwargs)
@@ -534,7 +583,7 @@ def get_prefetch_related(cls):
return ["permissions"]
@classmethod
- def get_computed_fields(cls, delimiter=", "):
+ def get_annotated_fields(cls, delimiter=", "):
"""
Get a dict of computed fields.
"""
@@ -751,7 +800,7 @@ def get_filter_conditions(cls):
)
@classmethod
- def get_computed_fields(cls, delimiter=", "):
+ def get_annotated_fields(cls, delimiter=", "):
"""
Get a dict of computed fields.
"""
@@ -846,7 +895,7 @@ def get_filter_conditions(cls):
)
@classmethod
- def get_computed_fields(cls, delimiter=", "):
+ def get_annotated_fields(cls, delimiter=", "):
"""
Get a dict of computed fields.
"""
@@ -1465,7 +1514,7 @@ def get_sort_fields(cls):
]
@classmethod
- def get_computed_fields(cls, delimiter=", "):
+ def get_annotated_fields(cls, delimiter=", "):
"""
Get a dict of computed fields.
"""
From 2265b70b5043c0efa6100f9ce4452f54427f9a30 Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Wed, 13 Nov 2024 13:10:11 -0700
Subject: [PATCH 02/45] Refactor part 1
---
src/registrar/models/utility/orm_helper.py | 6 +
src/registrar/utility/csv_export.py | 234 +++++++++++++++---
src/registrar/views/portfolio_members_json.py | 111 ++-------
3 files changed, 230 insertions(+), 121 deletions(-)
create mode 100644 src/registrar/models/utility/orm_helper.py
diff --git a/src/registrar/models/utility/orm_helper.py b/src/registrar/models/utility/orm_helper.py
new file mode 100644
index 000000000..24f7982e7
--- /dev/null
+++ b/src/registrar/models/utility/orm_helper.py
@@ -0,0 +1,6 @@
+from django.db.models.expressions import Func
+
+class ArrayRemove(Func):
+ """Custom Func to use array_remove to remove null values"""
+ function = "array_remove"
+ template = "%(function)s(%(expressions)s, NULL)"
\ No newline at end of file
diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py
index e90b27c29..d39ab996c 100644
--- a/src/registrar/utility/csv_export.py
+++ b/src/registrar/utility/csv_export.py
@@ -10,16 +10,20 @@
DomainInformation,
PublicContact,
UserDomainRole,
+ PortfolioInvitation,
)
-from django.db.models import Case, CharField, Count, DateField, F, ManyToManyField, Q, QuerySet, Value, When
+from django.db.models import Case, CharField, Count, DateField, F, ManyToManyField, Q, QuerySet, Value, When, TextField, OuterRef, Subquery
+from django.db.models.functions import Cast
from django.utils import timezone
from django.db.models.functions import Concat, Coalesce
from django.contrib.postgres.aggregates import StringAgg
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.generic_helper import convert_queryset_to_dict
+from registrar.models.utility.orm_helper import ArrayRemove
from registrar.templatetags.custom_filters import get_region
from registrar.utility.constants import BranchChoices
from registrar.utility.enums import DefaultEmail
+from django.contrib.postgres.aggregates import ArrayAgg
logger = logging.getLogger(__name__)
@@ -167,14 +171,7 @@ def annotate_and_retrieve_fields(
return cls.update_queryset(queryset, **kwargs)
@classmethod
- def update_queryset(cls, queryset, **kwargs):
- """
- Returns an updated queryset. Override in subclass to update queryset.
- """
- return queryset
-
- @classmethod
- def get_model_dict(cls):
+ def get_annotated_queryset(cls, request=None):
sort_fields = cls.get_sort_fields()
kwargs = cls.get_additional_args()
select_related = cls.get_select_related()
@@ -196,12 +193,21 @@ def get_model_dict(cls):
.order_by(*sort_fields)
.distinct()
)
- annotated_queryset = cls.annotate_and_retrieve_fields(
+
+ return cls.annotate_and_retrieve_fields(
model_queryset, annotated_fields, related_table_fields, **kwargs
)
- models_dict = convert_queryset_to_dict(annotated_queryset, is_model=False)
- return models_dict
+ @classmethod
+ def update_queryset(cls, queryset, **kwargs):
+ """
+ Returns an updated queryset. Override in subclass to update queryset.
+ """
+ return queryset
+
+ @classmethod
+ def get_models_dict(cls, request=None):
+ return convert_queryset_to_dict(cls.get_annotated_queryset(request), is_model=False)
class UserPortfolioPermissionModelDict(BaseModelDict):
@@ -212,36 +218,170 @@ def model(cls):
return UserPortfolioPermission
@classmethod
- def get_filter_conditions(cls, **export_kwargs):
+ def get_select_related(cls):
"""
- Get a Q object of filter conditions to filter when building queryset.
+ Get a list of tables to pass to select_related when building queryset.
"""
- return Q()
+ return ["user"]
@classmethod
- def get_exclusions(cls):
+ def get_filter_conditions(cls, portfolio):
"""
- Get a Q object of exclusion conditions to pass to .exclude() when building queryset.
+ Get a Q object of filter conditions to filter when building queryset.
"""
- return Q()
-
+ if not portfolio:
+ # Return nothing
+ return Q(id__in=[])
+
+ # Get all members on this portfolio
+ return Q(portfolio=portfolio)
+
@classmethod
- def get_annotated_fields(cls):
+ def get_annotated_fields(cls, portfolio):
"""
Get a dict of computed fields. These are fields that do not exist on the model normally
and will be passed to .annotate() when building a queryset.
"""
- return {}
+ if not portfolio:
+ # Return nothing
+ return {}
+ return {
+ "first_name": F("user__first_name"),
+ "last_name": F("user__last_name"),
+ "email_display": F("user__email"),
+ "last_active": Coalesce(
+ Cast(F("user__last_login"), output_field=TextField()),
+ Value("Invalid date"),
+ output_field=TextField(),
+ ),
+ "additional_permissions_display": F("additional_permissions"),
+ "member_display": Case(
+ When(
+ Q(user__email__isnull=False) & ~Q(user__email=""),
+ then=F("user__email")
+ ),
+ When(
+ Q(user__first_name__isnull=False) | Q(user__last_name__isnull=False),
+ then=Concat(
+ Coalesce(F("user__first_name"), Value("")),
+ Value(" "),
+ Coalesce(F("user__last_name"), Value("")),
+ ),
+ ),
+ default=Value(""),
+ output_field=CharField(),
+ ),
+ "domain_info": ArrayAgg(
+ Concat(
+ F("user__permissions__domain_id"),
+ Value(":"),
+ F("user__permissions__domain__name"),
+ output_field=CharField(),
+ ),
+ distinct=True,
+ filter=Q(user__permissions__domain__isnull=False)
+ & Q(user__permissions__domain__domain_info__portfolio=portfolio),
+ ),
+ "source": Value("permission", output_field=CharField()),
+ }
+
@classmethod
- def parse_row(cls, columns, model):
+ def get_annotated_queryset(cls, portfolio):
+ """Override of the base annotated queryset to pass in portfolio"""
+ model_queryset = (
+ cls.model()
+ .objects
+ .select_related(*cls.get_select_related())
+ .prefetch_related(*cls.get_prefetch_related())
+ .filter(cls.get_filter_conditions(portfolio))
+ .exclude(cls.get_exclusions())
+ .annotate(**cls.get_annotations_for_sort())
+ .order_by(*cls.get_sort_fields())
+ .distinct()
+ )
+
+ annotated_fields = cls.get_annotated_fields(portfolio)
+ related_table_fields = cls.get_related_table_fields()
+ return cls.annotate_and_retrieve_fields(
+ model_queryset, annotated_fields, related_table_fields
+ )
+
+
+class PortfolioInvitationModelDict(BaseModelDict):
+
+ @classmethod
+ def model(cls):
+ # Return the model class that this export handles
+ return PortfolioInvitation
+
+ @classmethod
+ def get_filter_conditions(cls, portfolio):
"""
- Given a set of columns and a model dictionary, generate a new row from cleaned column data.
+ Get a Q object of filter conditions to filter when building queryset.
"""
- FIELDS = {"Not yet defined": "Not yet defined"}
+ if not portfolio:
+ # Return nothing
+ return Q(id__in=[])
- row = [FIELDS.get(column, "") for column in columns]
- return row
+ # Get all members on this portfolio
+ return Q(
+ # Check if email matches the OuterRef("email")
+ email=OuterRef("email"),
+ # Check if the domain's portfolio matches the given portfolio)
+ domain__domain_info__portfolio=portfolio,
+ )
+
+ @classmethod
+ def get_annotated_fields(cls, portfolio):
+ """
+ Get a dict of computed fields. These are fields that do not exist on the model normally
+ and will be passed to .annotate() when building a queryset.
+ """
+ if not portfolio:
+ # Return nothing
+ return {}
+
+ domain_invitations = DomainInvitation.objects.filter(
+ email=OuterRef("email"), # Check if email matches the OuterRef("email")
+ domain__domain_info__portfolio=portfolio, # Check if the domain's portfolio matches the given portfolio
+ ).annotate(domain_info=Concat(F("domain__id"), Value(":"), F("domain__name"), output_field=CharField()))
+ return {
+ "first_name": Value(None, output_field=CharField()),
+ "last_name": Value(None, output_field=CharField()),
+ "email_display": F("email"),
+ "last_active": Value("Invited", output_field=TextField()),
+ "additional_permissions_display": F("additional_permissions"),
+ "member_display": F("email"),
+ "domain_info": ArrayRemove(
+ ArrayAgg(
+ Subquery(domain_invitations.values("domain_info")),
+ distinct=True,
+ )
+ ),
+ "source": Value("invitation", output_field=CharField()),
+ }
+
+ @classmethod
+ def get_annotated_queryset(cls, portfolio):
+ """Override of the base annotated queryset to pass in portfolio"""
+ model_queryset = (
+ cls.model()
+ .objects
+ .select_related(*cls.get_select_related())
+ .prefetch_related(*cls.get_prefetch_related())
+ .filter(cls.get_filter_conditions(portfolio))
+ .exclude(cls.get_exclusions())
+ .annotate(**cls.get_annotations_for_sort())
+ .order_by(*cls.get_sort_fields())
+ .distinct()
+ )
+
+ annotated_fields = cls.get_annotated_fields(portfolio)
+ related_table_fields = cls.get_related_table_fields()
+ return cls.annotate_and_retrieve_fields(
+ model_queryset, annotated_fields, related_table_fields
+ )
class BaseExport(BaseModelDict):
@@ -266,14 +406,14 @@ def write_csv_before(cls, csv_writer, **export_kwargs):
pass
@classmethod
- def export_data_to_csv(cls, csv_file, **export_kwargs):
+ def export_data_to_csv(cls, csv_file, request=None, **export_kwargs):
"""
All domain metadata:
Exports domains of all statuses plus domain managers.
"""
writer = csv.writer(csv_file)
columns = cls.get_columns()
- models_dict = cls.get_model_dict()
+ models_dict = cls.get_models_dict()
# Write to csv file before the write_csv
cls.write_csv_before(writer, **export_kwargs)
@@ -321,6 +461,43 @@ def parse_row(cls, columns, model):
"""
pass
+class MemberExport(BaseExport):
+
+ @classmethod
+ def model(self):
+ """
+ No model is defined for the member report as it is a combination of multiple fields.
+ This is a special edge case, but the base report requires this to be defined.
+ """
+ return None
+
+ @classmethod
+ def get_models_dict(cls, request=None):
+ portfolio = request.session.get("portfolio")
+ if not portfolio:
+ return {}
+
+ # Union the two querysets to combine UserPortfolioPermission + invites
+ permissions = UserPortfolioPermissionModelDict.get_annotated_queryset(portfolio)
+ invitations = PortfolioInvitationModelDict.get_annotated_queryset(portfolio)
+ objects = permissions.union(invitations)
+ return convert_queryset_to_dict(objects, is_model=False)
+
+ @classmethod
+ def get_columns(cls):
+ """
+ Returns the columns for CSV export. Override in subclasses as needed.
+ """
+ return []
+
+ @classmethod
+ @abstractmethod
+ def parse_row(cls, columns, model):
+ """
+ Given a set of columns and a model dictionary, generate a new row from cleaned column data.
+ Must be implemented by subclasses
+ """
+ pass
class DomainExport(BaseExport):
"""
@@ -1597,3 +1774,4 @@ def get_creator_active_requests_count_query(cls):
distinct=True,
)
return query
+
diff --git a/src/registrar/views/portfolio_members_json.py b/src/registrar/views/portfolio_members_json.py
index 17209f0d9..408d37ff5 100644
--- a/src/registrar/views/portfolio_members_json.py
+++ b/src/registrar/views/portfolio_members_json.py
@@ -7,10 +7,9 @@
from django.urls import reverse
from django.views import View
-from registrar.models.domain_invitation import DomainInvitation
-from registrar.models.portfolio_invitation import PortfolioInvitation
-from registrar.models.user_portfolio_permission import UserPortfolioPermission
+from registrar.models import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
+from registrar.utility.csv_export import UserPortfolioPermissionModelDict, PortfolioInvitationModelDict
from registrar.views.utility.mixins import PortfolioMembersPermission
@@ -39,7 +38,7 @@ def get(self, request):
page_number = request.GET.get("page", 1)
page_obj = paginator.get_page(page_number)
- members = [self.serialize_members(request, portfolio, item, request.user) for item in page_obj.object_list]
+ members = [self.serialize_members(portfolio, item, request.user) for item in page_obj.object_list]
return JsonResponse(
{
@@ -56,92 +55,25 @@ def get(self, request):
def initial_permissions_search(self, portfolio):
"""Perform initial search for permissions before applying any filters."""
- permissions = UserPortfolioPermission.objects.filter(portfolio=portfolio)
- permissions = (
- permissions.select_related("user")
- .annotate(
- first_name=F("user__first_name"),
- last_name=F("user__last_name"),
- email_display=F("user__email"),
- last_active=Coalesce(
- Cast(F("user__last_login"), output_field=TextField()), # Cast last_login to text
- Value("Invalid date"),
- output_field=TextField(),
- ),
- additional_permissions_display=F("additional_permissions"),
- member_display=Case(
- # If email is present and not blank, use email
- When(Q(user__email__isnull=False) & ~Q(user__email=""), then=F("user__email")),
- # If first name or last name is present, use concatenation of first_name + " " + last_name
- When(
- Q(user__first_name__isnull=False) | Q(user__last_name__isnull=False),
- then=Concat(
- Coalesce(F("user__first_name"), Value("")),
- Value(" "),
- Coalesce(F("user__last_name"), Value("")),
- ),
- ),
- # If neither, use an empty string
- default=Value(""),
- output_field=CharField(),
- ),
- domain_info=ArrayAgg(
- # an array of domains, with id and name, colon separated
- Concat(
- F("user__permissions__domain_id"),
- Value(":"),
- F("user__permissions__domain__name"),
- # specify the output_field to ensure union has same column types
- output_field=CharField(),
- ),
- distinct=True,
- filter=Q(user__permissions__domain__isnull=False) # filter out null values
- & Q(
- user__permissions__domain__domain_info__portfolio=portfolio
- ), # only include domains in portfolio
- ),
- source=Value("permission", output_field=CharField()),
- )
- .values(
- "id",
- "first_name",
- "last_name",
- "email_display",
- "last_active",
- "roles",
- "additional_permissions_display",
- "member_display",
- "domain_info",
- "source",
- )
+ queryset = UserPortfolioPermissionModelDict.get_annotated_queryset(portfolio)
+ return queryset.values(
+ "id",
+ "first_name",
+ "last_name",
+ "email_display",
+ "last_active",
+ "roles",
+ "additional_permissions_display",
+ "member_display",
+ "domain_info",
+ "source",
)
- return permissions
def initial_invitations_search(self, portfolio):
"""Perform initial invitations search and get related DomainInvitation data based on the email."""
# Get DomainInvitation query for matching email and for the portfolio
- domain_invitations = DomainInvitation.objects.filter(
- email=OuterRef("email"), # Check if email matches the OuterRef("email")
- domain__domain_info__portfolio=portfolio, # Check if the domain's portfolio matches the given portfolio
- ).annotate(domain_info=Concat(F("domain__id"), Value(":"), F("domain__name"), output_field=CharField()))
- # PortfolioInvitation query
- invitations = PortfolioInvitation.objects.filter(portfolio=portfolio)
- invitations = invitations.annotate(
- first_name=Value(None, output_field=CharField()),
- last_name=Value(None, output_field=CharField()),
- email_display=F("email"),
- last_active=Value("Invited", output_field=TextField()),
- additional_permissions_display=F("additional_permissions"),
- member_display=F("email"),
- # Use ArrayRemove to return an empty list when no domain invitations are found
- domain_info=ArrayRemove(
- ArrayAgg(
- Subquery(domain_invitations.values("domain_info")),
- distinct=True,
- )
- ),
- source=Value("invitation", output_field=CharField()),
- ).values(
+ queryset = PortfolioInvitationModelDict.get_annotated_queryset(portfolio)
+ return queryset.values(
"id",
"first_name",
"last_name",
@@ -153,7 +85,6 @@ def initial_invitations_search(self, portfolio):
"domain_info",
"source",
)
- return invitations
def apply_search_term(self, queryset, request):
"""Apply search term to the queryset."""
@@ -179,7 +110,7 @@ def apply_sorting(self, queryset, request):
queryset = queryset.order_by(sort_by)
return queryset
- def serialize_members(self, request, portfolio, item, user):
+ def serialize_members(self, portfolio, item, user):
# Check if the user can edit other users
user_can_edit_other_users = any(
user.has_perm(perm) for perm in ["registrar.full_access_permission", "registrar.change_user"]
@@ -213,9 +144,3 @@ def serialize_members(self, request, portfolio, item, user):
"svg_icon": ("visibility" if view_only else "settings"),
}
return member_json
-
-
-# Custom Func to use array_remove to remove null values
-class ArrayRemove(Func):
- function = "array_remove"
- template = "%(function)s(%(expressions)s, NULL)"
From 48ae7f4c117af8491870e22e6a0cc46ce102055e Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Wed, 13 Nov 2024 14:31:59 -0700
Subject: [PATCH 03/45] Refactor part 2
---
.../templates/includes/members_table.html | 11 +
src/registrar/utility/csv_export.py | 369 ++----------------
src/registrar/utility/model_dicts.py | 342 ++++++++++++++++
src/registrar/views/portfolio_members_json.py | 2 +-
4 files changed, 388 insertions(+), 336 deletions(-)
create mode 100644 src/registrar/utility/model_dicts.py
diff --git a/src/registrar/templates/includes/members_table.html b/src/registrar/templates/includes/members_table.html
index 5e0dc6116..b1118abb4 100644
--- a/src/registrar/templates/includes/members_table.html
+++ b/src/registrar/templates/includes/members_table.html
@@ -36,6 +36,17 @@
+ {% comment %} {% if user_domain_count and user_domain_count > 0 %} {% endcomment %}
+
+ {% comment %} {% endif %} {% endcomment %}
diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py
index d39ab996c..4b01c7e45 100644
--- a/src/registrar/utility/csv_export.py
+++ b/src/registrar/utility/csv_export.py
@@ -20,11 +20,14 @@
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.generic_helper import convert_queryset_to_dict
from registrar.models.utility.orm_helper import ArrayRemove
+from registrar.models.utility.portfolio_helper import UserPortfolioRoleChoices
from registrar.templatetags.custom_filters import get_region
from registrar.utility.constants import BranchChoices
from registrar.utility.enums import DefaultEmail
from django.contrib.postgres.aggregates import ArrayAgg
+from registrar.utility.model_dicts import BaseModelDict, PortfolioInvitationModelDict, UserPortfolioPermissionModelDict
+
logger = logging.getLogger(__name__)
@@ -55,335 +58,6 @@ def format_end_date(end_date):
return timezone.make_aware(datetime.strptime(end_date, "%Y-%m-%d")) if end_date else get_default_end_date()
-class BaseModelDict(ABC):
-
- @classmethod
- @abstractmethod
- def model(self):
- """
- Property to specify the model that the export class will handle.
- Must be implemented by subclasses.
- """
- pass
-
- @classmethod
- def get_sort_fields(cls):
- """
- Returns the sort fields for the CSV export. Override in subclasses as needed.
- """
- return []
-
- @classmethod
- def get_additional_args(cls):
- """
- Returns additional keyword arguments as an empty dictionary.
- Override in subclasses to provide specific arguments.
- """
- return {}
-
- @classmethod
- def get_select_related(cls):
- """
- Get a list of tables to pass to select_related when building queryset.
- """
- return []
-
- @classmethod
- def get_prefetch_related(cls):
- """
- Get a list of tables to pass to prefetch_related when building queryset.
- """
- return []
-
- @classmethod
- def get_exclusions(cls):
- """
- Get a Q object of exclusion conditions to pass to .exclude() when building queryset.
- """
- return Q()
-
- @classmethod
- def get_filter_conditions(cls, **export_kwargs):
- """
- Get a Q object of filter conditions to filter when building queryset.
- """
- return Q()
-
- @classmethod
- def get_annotated_fields(cls):
- """
- Get a dict of computed fields. These are fields that do not exist on the model normally
- and will be passed to .annotate() when building a queryset.
- """
- return {}
-
- @classmethod
- def get_annotations_for_sort(cls):
- """
- Get a dict of annotations to make available for order_by clause.
- """
- return {}
-
- @classmethod
- def get_related_table_fields(cls):
- """
- Get a list of fields from related tables.
- """
- return []
-
- @classmethod
- def annotate_and_retrieve_fields(
- cls, initial_queryset, annotated_fields, related_table_fields=None, include_many_to_many=False, **kwargs
- ) -> QuerySet:
- """
- Applies annotations to a queryset and retrieves specified fields,
- including class-defined and annotation-defined.
-
- Parameters:
- initial_queryset (QuerySet): Initial queryset.
- annotated_fields (dict, optional): Fields to compute {field_name: expression}.
- related_table_fields (list, optional): Extra fields to retrieve; defaults to annotation keys if None.
- include_many_to_many (bool, optional): Determines if we should include many to many fields or not
- **kwargs: Additional keyword arguments for specific parameters (e.g., public_contacts, domain_invitations,
- user_domain_roles).
-
- Returns:
- QuerySet: Contains dictionaries with the specified fields for each record.
- """
- if related_table_fields is None:
- related_table_fields = []
-
- # We can infer that if we're passing in annotations,
- # we want to grab the result of said annotation.
- if annotated_fields:
- related_table_fields.extend(annotated_fields.keys())
-
- # Get prexisting fields on the model
- model_fields = set()
- for field in cls.model()._meta.get_fields():
- # Exclude many to many fields unless we specify
- many_to_many = isinstance(field, ManyToManyField) and include_many_to_many
- if many_to_many or not isinstance(field, ManyToManyField):
- model_fields.add(field.name)
-
- queryset = initial_queryset.annotate(**annotated_fields).values(*model_fields, *related_table_fields)
-
- return cls.update_queryset(queryset, **kwargs)
-
- @classmethod
- def get_annotated_queryset(cls, request=None):
- sort_fields = cls.get_sort_fields()
- kwargs = cls.get_additional_args()
- select_related = cls.get_select_related()
- prefetch_related = cls.get_prefetch_related()
- exclusions = cls.get_exclusions()
- annotations_for_sort = cls.get_annotations_for_sort()
- filter_conditions = cls.get_filter_conditions(**kwargs)
- annotated_fields = cls.get_annotated_fields()
- related_table_fields = cls.get_related_table_fields()
-
- model_queryset = (
- cls.model()
- .objects
- .select_related(*select_related)
- .prefetch_related(*prefetch_related)
- .filter(filter_conditions)
- .exclude(exclusions)
- .annotate(**annotations_for_sort)
- .order_by(*sort_fields)
- .distinct()
- )
-
- return cls.annotate_and_retrieve_fields(
- model_queryset, annotated_fields, related_table_fields, **kwargs
- )
-
- @classmethod
- def update_queryset(cls, queryset, **kwargs):
- """
- Returns an updated queryset. Override in subclass to update queryset.
- """
- return queryset
-
- @classmethod
- def get_models_dict(cls, request=None):
- return convert_queryset_to_dict(cls.get_annotated_queryset(request), is_model=False)
-
-
-class UserPortfolioPermissionModelDict(BaseModelDict):
-
- @classmethod
- def model(cls):
- # Return the model class that this export handles
- return UserPortfolioPermission
-
- @classmethod
- def get_select_related(cls):
- """
- Get a list of tables to pass to select_related when building queryset.
- """
- return ["user"]
-
- @classmethod
- def get_filter_conditions(cls, portfolio):
- """
- Get a Q object of filter conditions to filter when building queryset.
- """
- if not portfolio:
- # Return nothing
- return Q(id__in=[])
-
- # Get all members on this portfolio
- return Q(portfolio=portfolio)
-
- @classmethod
- def get_annotated_fields(cls, portfolio):
- """
- Get a dict of computed fields. These are fields that do not exist on the model normally
- and will be passed to .annotate() when building a queryset.
- """
- if not portfolio:
- # Return nothing
- return {}
-
- return {
- "first_name": F("user__first_name"),
- "last_name": F("user__last_name"),
- "email_display": F("user__email"),
- "last_active": Coalesce(
- Cast(F("user__last_login"), output_field=TextField()),
- Value("Invalid date"),
- output_field=TextField(),
- ),
- "additional_permissions_display": F("additional_permissions"),
- "member_display": Case(
- When(
- Q(user__email__isnull=False) & ~Q(user__email=""),
- then=F("user__email")
- ),
- When(
- Q(user__first_name__isnull=False) | Q(user__last_name__isnull=False),
- then=Concat(
- Coalesce(F("user__first_name"), Value("")),
- Value(" "),
- Coalesce(F("user__last_name"), Value("")),
- ),
- ),
- default=Value(""),
- output_field=CharField(),
- ),
- "domain_info": ArrayAgg(
- Concat(
- F("user__permissions__domain_id"),
- Value(":"),
- F("user__permissions__domain__name"),
- output_field=CharField(),
- ),
- distinct=True,
- filter=Q(user__permissions__domain__isnull=False)
- & Q(user__permissions__domain__domain_info__portfolio=portfolio),
- ),
- "source": Value("permission", output_field=CharField()),
- }
-
- @classmethod
- def get_annotated_queryset(cls, portfolio):
- """Override of the base annotated queryset to pass in portfolio"""
- model_queryset = (
- cls.model()
- .objects
- .select_related(*cls.get_select_related())
- .prefetch_related(*cls.get_prefetch_related())
- .filter(cls.get_filter_conditions(portfolio))
- .exclude(cls.get_exclusions())
- .annotate(**cls.get_annotations_for_sort())
- .order_by(*cls.get_sort_fields())
- .distinct()
- )
-
- annotated_fields = cls.get_annotated_fields(portfolio)
- related_table_fields = cls.get_related_table_fields()
- return cls.annotate_and_retrieve_fields(
- model_queryset, annotated_fields, related_table_fields
- )
-
-
-class PortfolioInvitationModelDict(BaseModelDict):
-
- @classmethod
- def model(cls):
- # Return the model class that this export handles
- return PortfolioInvitation
-
- @classmethod
- def get_filter_conditions(cls, portfolio):
- """
- Get a Q object of filter conditions to filter when building queryset.
- """
- if not portfolio:
- # Return nothing
- return Q(id__in=[])
-
- # Get all members on this portfolio
- return Q(
- # Check if email matches the OuterRef("email")
- email=OuterRef("email"),
- # Check if the domain's portfolio matches the given portfolio)
- domain__domain_info__portfolio=portfolio,
- )
-
- @classmethod
- def get_annotated_fields(cls, portfolio):
- """
- Get a dict of computed fields. These are fields that do not exist on the model normally
- and will be passed to .annotate() when building a queryset.
- """
- if not portfolio:
- # Return nothing
- return {}
-
- domain_invitations = DomainInvitation.objects.filter(
- email=OuterRef("email"), # Check if email matches the OuterRef("email")
- domain__domain_info__portfolio=portfolio, # Check if the domain's portfolio matches the given portfolio
- ).annotate(domain_info=Concat(F("domain__id"), Value(":"), F("domain__name"), output_field=CharField()))
- return {
- "first_name": Value(None, output_field=CharField()),
- "last_name": Value(None, output_field=CharField()),
- "email_display": F("email"),
- "last_active": Value("Invited", output_field=TextField()),
- "additional_permissions_display": F("additional_permissions"),
- "member_display": F("email"),
- "domain_info": ArrayRemove(
- ArrayAgg(
- Subquery(domain_invitations.values("domain_info")),
- distinct=True,
- )
- ),
- "source": Value("invitation", output_field=CharField()),
- }
-
- @classmethod
- def get_annotated_queryset(cls, portfolio):
- """Override of the base annotated queryset to pass in portfolio"""
- model_queryset = (
- cls.model()
- .objects
- .select_related(*cls.get_select_related())
- .prefetch_related(*cls.get_prefetch_related())
- .filter(cls.get_filter_conditions(portfolio))
- .exclude(cls.get_exclusions())
- .annotate(**cls.get_annotations_for_sort())
- .order_by(*cls.get_sort_fields())
- .distinct()
- )
-
- annotated_fields = cls.get_annotated_fields(portfolio)
- related_table_fields = cls.get_related_table_fields()
- return cls.annotate_and_retrieve_fields(
- model_queryset, annotated_fields, related_table_fields
- )
-
-
class BaseExport(BaseModelDict):
"""
A generic class for exporting data which returns a csv file for the given model.
@@ -413,7 +87,7 @@ def export_data_to_csv(cls, csv_file, request=None, **export_kwargs):
"""
writer = csv.writer(csv_file)
columns = cls.get_columns()
- models_dict = cls.get_models_dict()
+ models_dict = cls.get_models_dict(request=request)
# Write to csv file before the write_csv
cls.write_csv_before(writer, **export_kwargs)
@@ -488,8 +162,18 @@ def get_columns(cls):
"""
Returns the columns for CSV export. Override in subclasses as needed.
"""
- return []
-
+ return [
+ "Email",
+ "Organization admin",
+ "Invited by",
+ "Last active",
+ "Domain requests",
+ "Member management",
+ "Domain management",
+ "Number of domains",
+ "Domains",
+ ]
+
@classmethod
@abstractmethod
def parse_row(cls, columns, model):
@@ -497,7 +181,22 @@ def parse_row(cls, columns, model):
Given a set of columns and a model dictionary, generate a new row from cleaned column data.
Must be implemented by subclasses
"""
- pass
+
+ is_admin = UserPortfolioRoleChoices.ORGANIZATION_ADMIN in (model.get("roles") or [])
+ FIELDS = {
+ "Email": model.get("email"),
+ "Organization admin": is_admin,
+ "Invited by": "TODO",
+ "Last active": "TODO",
+ "Domain requests": "TODO",
+ "Member management": "TODO",
+ "Domain management": "TODO",
+ "Number of domains": "TODO",
+ "Domains": "TODO",
+ }
+
+ row = [FIELDS.get(column, "") for column in columns]
+ return row
class DomainExport(BaseExport):
"""
@@ -757,7 +456,7 @@ def get_prefetch_related(cls):
"""
Get a list of tables to pass to prefetch_related when building queryset.
"""
- return ["permissions"]
+ return ["domain__permissions"]
@classmethod
def get_annotated_fields(cls, delimiter=", "):
@@ -797,7 +496,7 @@ class DomainDataTypeUser(DomainDataType):
"""
@classmethod
- def get_filter_conditions(cls, request=None):
+ def get_filter_conditions(cls, request=None, **kwargs):
"""
Get a Q object of filter conditions to filter when building queryset.
"""
diff --git a/src/registrar/utility/model_dicts.py b/src/registrar/utility/model_dicts.py
new file mode 100644
index 000000000..d3f71aa1e
--- /dev/null
+++ b/src/registrar/utility/model_dicts.py
@@ -0,0 +1,342 @@
+"""
+TODO: explanation here
+"""
+from abc import ABC, abstractmethod
+from registrar.models import (
+ DomainInvitation,
+ PortfolioInvitation,
+)
+from django.db.models import Case, CharField, F, ManyToManyField, Q, QuerySet, Value, When, TextField, OuterRef, Subquery
+from django.db.models.functions import Cast
+from django.db.models.functions import Concat, Coalesce
+from registrar.models.user_portfolio_permission import UserPortfolioPermission
+from registrar.models.utility.generic_helper import convert_queryset_to_dict
+from registrar.models.utility.orm_helper import ArrayRemove
+from django.contrib.postgres.aggregates import ArrayAgg
+
+
+class BaseModelDict(ABC):
+
+ @classmethod
+ @abstractmethod
+ def model(self):
+ """
+ Property to specify the model that the export class will handle.
+ Must be implemented by subclasses.
+ """
+ pass
+
+ @classmethod
+ def get_sort_fields(cls):
+ """
+ Returns the sort fields for the CSV export. Override in subclasses as needed.
+ """
+ return []
+
+ @classmethod
+ def get_additional_args(cls):
+ """
+ Returns additional keyword arguments as an empty dictionary.
+ Override in subclasses to provide specific arguments.
+ """
+ return {}
+
+ @classmethod
+ def get_select_related(cls):
+ """
+ Get a list of tables to pass to select_related when building queryset.
+ """
+ return []
+
+ @classmethod
+ def get_prefetch_related(cls):
+ """
+ Get a list of tables to pass to prefetch_related when building queryset.
+ """
+ return []
+
+ @classmethod
+ def get_exclusions(cls):
+ """
+ Get a Q object of exclusion conditions to pass to .exclude() when building queryset.
+ """
+ return Q()
+
+ @classmethod
+ def get_filter_conditions(cls, **export_kwargs):
+ """
+ Get a Q object of filter conditions to filter when building queryset.
+ """
+ return Q()
+
+ @classmethod
+ def get_annotated_fields(cls):
+ """
+ Get a dict of computed fields. These are fields that do not exist on the model normally
+ and will be passed to .annotate() when building a queryset.
+ """
+ return {}
+
+ @classmethod
+ def get_annotations_for_sort(cls):
+ """
+ Get a dict of annotations to make available for order_by clause.
+ """
+ return {}
+
+ @classmethod
+ def get_related_table_fields(cls):
+ """
+ Get a list of fields from related tables.
+ """
+ return []
+
+ @classmethod
+ def annotate_and_retrieve_fields(
+ cls, initial_queryset, annotated_fields, related_table_fields=None, include_many_to_many=False, **kwargs
+ ) -> QuerySet:
+ """
+ Applies annotations to a queryset and retrieves specified fields,
+ including class-defined and annotation-defined.
+
+ Parameters:
+ initial_queryset (QuerySet): Initial queryset.
+ annotated_fields (dict, optional): Fields to compute {field_name: expression}.
+ related_table_fields (list, optional): Extra fields to retrieve; defaults to annotation keys if None.
+ include_many_to_many (bool, optional): Determines if we should include many to many fields or not
+ **kwargs: Additional keyword arguments for specific parameters (e.g., public_contacts, domain_invitations,
+ user_domain_roles).
+
+ Returns:
+ QuerySet: Contains dictionaries with the specified fields for each record.
+ """
+ if related_table_fields is None:
+ related_table_fields = []
+
+ # We can infer that if we're passing in annotations,
+ # we want to grab the result of said annotation.
+ if annotated_fields:
+ related_table_fields.extend(annotated_fields.keys())
+
+ # Get prexisting fields on the model
+ model_fields = set()
+ for field in cls.model()._meta.get_fields():
+ # Exclude many to many fields unless we specify
+ many_to_many = isinstance(field, ManyToManyField) and include_many_to_many
+ if many_to_many or not isinstance(field, ManyToManyField):
+ model_fields.add(field.name)
+
+ queryset = initial_queryset.annotate(**annotated_fields).values(*model_fields, *related_table_fields)
+
+ return cls.update_queryset(queryset, **kwargs)
+
+ @classmethod
+ def get_annotated_queryset(cls, **kwargs):
+ sort_fields = cls.get_sort_fields()
+ # Get additional args and merge with incoming kwargs
+ additional_args = cls.get_additional_args()
+ kwargs.update(additional_args)
+ select_related = cls.get_select_related()
+ prefetch_related = cls.get_prefetch_related()
+ exclusions = cls.get_exclusions()
+ annotations_for_sort = cls.get_annotations_for_sort()
+ filter_conditions = cls.get_filter_conditions(**kwargs)
+ annotated_fields = cls.get_annotated_fields()
+ related_table_fields = cls.get_related_table_fields()
+
+ model_queryset = (
+ cls.model()
+ .objects
+ .select_related(*select_related)
+ .prefetch_related(*prefetch_related)
+ .filter(filter_conditions)
+ .exclude(exclusions)
+ .annotate(**annotations_for_sort)
+ .order_by(*sort_fields)
+ .distinct()
+ )
+ return cls.annotate_and_retrieve_fields(
+ model_queryset, annotated_fields, related_table_fields, **kwargs
+ )
+
+ @classmethod
+ def update_queryset(cls, queryset, **kwargs):
+ """
+ Returns an updated queryset. Override in subclass to update queryset.
+ """
+ return queryset
+
+ @classmethod
+ def get_models_dict(cls, **kwargs):
+ request = kwargs.get("request")
+ print(f"get_models_dict => request is: {request}")
+ return convert_queryset_to_dict(cls.get_annotated_queryset(**kwargs), is_model=False)
+
+
+class UserPortfolioPermissionModelDict(BaseModelDict):
+
+ @classmethod
+ def model(cls):
+ # Return the model class that this export handles
+ return UserPortfolioPermission
+
+ @classmethod
+ def get_select_related(cls):
+ """
+ Get a list of tables to pass to select_related when building queryset.
+ """
+ return ["user"]
+
+ @classmethod
+ def get_filter_conditions(cls, portfolio):
+ """
+ Get a Q object of filter conditions to filter when building queryset.
+ """
+ if not portfolio:
+ # Return nothing
+ return Q(id__in=[])
+
+ # Get all members on this portfolio
+ return Q(portfolio=portfolio)
+
+ @classmethod
+ def get_annotated_fields(cls, portfolio):
+ """
+ Get a dict of computed fields. These are fields that do not exist on the model normally
+ and will be passed to .annotate() when building a queryset.
+ """
+ if not portfolio:
+ # Return nothing
+ return {}
+
+ return {
+ "first_name": F("user__first_name"),
+ "last_name": F("user__last_name"),
+ "email_display": F("user__email"),
+ "last_active": Coalesce(
+ Cast(F("user__last_login"), output_field=TextField()),
+ Value("Invalid date"),
+ output_field=TextField(),
+ ),
+ "additional_permissions_display": F("additional_permissions"),
+ "member_display": Case(
+ When(
+ Q(user__email__isnull=False) & ~Q(user__email=""),
+ then=F("user__email")
+ ),
+ When(
+ Q(user__first_name__isnull=False) | Q(user__last_name__isnull=False),
+ then=Concat(
+ Coalesce(F("user__first_name"), Value("")),
+ Value(" "),
+ Coalesce(F("user__last_name"), Value("")),
+ ),
+ ),
+ default=Value(""),
+ output_field=CharField(),
+ ),
+ "domain_info": ArrayAgg(
+ Concat(
+ F("user__permissions__domain_id"),
+ Value(":"),
+ F("user__permissions__domain__name"),
+ output_field=CharField(),
+ ),
+ distinct=True,
+ filter=Q(user__permissions__domain__isnull=False)
+ & Q(user__permissions__domain__domain_info__portfolio=portfolio),
+ ),
+ "source": Value("permission", output_field=CharField()),
+ }
+
+ @classmethod
+ def get_annotated_queryset(cls, portfolio):
+ """Override of the base annotated queryset to pass in portfolio"""
+ model_queryset = (
+ cls.model()
+ .objects
+ .select_related(*cls.get_select_related())
+ .prefetch_related(*cls.get_prefetch_related())
+ .filter(cls.get_filter_conditions(portfolio))
+ .exclude(cls.get_exclusions())
+ .annotate(**cls.get_annotations_for_sort())
+ .order_by(*cls.get_sort_fields())
+ .distinct()
+ )
+
+ annotated_fields = cls.get_annotated_fields(portfolio)
+ related_table_fields = cls.get_related_table_fields()
+ return cls.annotate_and_retrieve_fields(
+ model_queryset, annotated_fields, related_table_fields
+ )
+
+
+class PortfolioInvitationModelDict(BaseModelDict):
+
+ @classmethod
+ def model(cls):
+ # Return the model class that this export handles
+ return PortfolioInvitation
+
+ @classmethod
+ def get_filter_conditions(cls, portfolio):
+ """
+ Get a Q object of filter conditions to filter when building queryset.
+ """
+ if not portfolio:
+ # Return nothing
+ return Q(id__in=[])
+
+ # Get all members on this portfolio
+ return Q(portfolio=portfolio)
+
+ @classmethod
+ def get_annotated_fields(cls, portfolio):
+ """
+ Get a dict of computed fields. These are fields that do not exist on the model normally
+ and will be passed to .annotate() when building a queryset.
+ """
+ if not portfolio:
+ # Return nothing
+ return {}
+
+ domain_invitations = DomainInvitation.objects.filter(
+ email=OuterRef("email"), # Check if email matches the OuterRef("email")
+ domain__domain_info__portfolio=portfolio, # Check if the domain's portfolio matches the given portfolio
+ ).annotate(domain_info=Concat(F("domain__id"), Value(":"), F("domain__name"), output_field=CharField()))
+ return {
+ "first_name": Value(None, output_field=CharField()),
+ "last_name": Value(None, output_field=CharField()),
+ "email_display": F("email"),
+ "last_active": Value("Invited", output_field=TextField()),
+ "additional_permissions_display": F("additional_permissions"),
+ "member_display": F("email"),
+ "domain_info": ArrayRemove(
+ ArrayAgg(
+ Subquery(domain_invitations.values("domain_info")),
+ distinct=True,
+ )
+ ),
+ "source": Value("invitation", output_field=CharField()),
+ }
+
+ @classmethod
+ def get_annotated_queryset(cls, portfolio):
+ """Override of the base annotated queryset to pass in portfolio"""
+ model_queryset = (
+ cls.model()
+ .objects
+ .select_related(*cls.get_select_related())
+ .prefetch_related(*cls.get_prefetch_related())
+ .filter(cls.get_filter_conditions(portfolio))
+ .exclude(cls.get_exclusions())
+ .annotate(**cls.get_annotations_for_sort())
+ .order_by(*cls.get_sort_fields())
+ .distinct()
+ )
+
+ annotated_fields = cls.get_annotated_fields(portfolio)
+ related_table_fields = cls.get_related_table_fields()
+ return cls.annotate_and_retrieve_fields(
+ model_queryset, annotated_fields, related_table_fields
+ )
diff --git a/src/registrar/views/portfolio_members_json.py b/src/registrar/views/portfolio_members_json.py
index 408d37ff5..3a909810c 100644
--- a/src/registrar/views/portfolio_members_json.py
+++ b/src/registrar/views/portfolio_members_json.py
@@ -9,7 +9,7 @@
from registrar.models import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
-from registrar.utility.csv_export import UserPortfolioPermissionModelDict, PortfolioInvitationModelDict
+from registrar.utility.model_dicts import PortfolioInvitationModelDict, UserPortfolioPermissionModelDict
from registrar.views.utility.mixins import PortfolioMembersPermission
From 1e43974a111beb419c595c2cd853d4b194f2f7fc Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Thu, 14 Nov 2024 08:11:38 -0700
Subject: [PATCH 04/45] view
---
src/registrar/config/urls.py | 6 ++++++
src/registrar/templates/includes/members_table.html | 2 +-
src/registrar/views/report_views.py | 12 ++++++++++++
3 files changed, 19 insertions(+), 1 deletion(-)
diff --git a/src/registrar/config/urls.py b/src/registrar/config/urls.py
index d289eaf90..e5606b986 100644
--- a/src/registrar/config/urls.py
+++ b/src/registrar/config/urls.py
@@ -21,6 +21,7 @@
ExportDomainRequestDataFull,
ExportDataTypeUser,
ExportDataTypeRequests,
+ ExportMembersPortfolio,
)
# --jsons
@@ -216,6 +217,11 @@
name="get-rejection-email-for-user-json",
),
path("admin/", admin.site.urls),
+ path(
+ "reports/export_members_portfolio/",
+ ExportMembersPortfolio.as_view(),
+ name="export_members_portfolio",
+ ),
path(
"reports/export_data_type_user/",
ExportDataTypeUser.as_view(),
diff --git a/src/registrar/templates/includes/members_table.html b/src/registrar/templates/includes/members_table.html
index b1118abb4..ae430501d 100644
--- a/src/registrar/templates/includes/members_table.html
+++ b/src/registrar/templates/includes/members_table.html
@@ -39,7 +39,7 @@
{% comment %} {% if user_domain_count and user_domain_count > 0 %} {% endcomment %}
- {% comment %} {% if user_domain_count and user_domain_count > 0 %} {% endcomment %}
+ {% if member_count and member_count > 0 %}
- {% comment %} {% endif %} {% endcomment %}
+ {% endif %}
diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py
index cf9cd5a2b..76a84094c 100644
--- a/src/registrar/utility/csv_export.py
+++ b/src/registrar/utility/csv_export.py
@@ -11,7 +11,21 @@
PublicContact,
UserDomainRole,
)
-from django.db.models import Case, CharField, Count, DateField, F, ManyToManyField, Q, QuerySet, Value, When, TextField, OuterRef, Subquery
+from django.db.models import (
+ Case,
+ CharField,
+ Count,
+ DateField,
+ F,
+ ManyToManyField,
+ Q,
+ QuerySet,
+ Value,
+ When,
+ TextField,
+ OuterRef,
+ Subquery,
+)
from django.db.models.functions import Cast
from django.utils import timezone
from django.db.models.functions import Concat, Coalesce
@@ -25,7 +39,11 @@
from registrar.utility.enums import DefaultEmail
from django.contrib.postgres.aggregates import ArrayAgg
-from registrar.utility.model_annotations import BaseModelAnnotation, PortfolioInvitationModelAnnotation, UserPortfolioPermissionModelAnnotation
+from registrar.utility.model_annotations import (
+ BaseModelAnnotation,
+ PortfolioInvitationModelAnnotation,
+ UserPortfolioPermissionModelAnnotation,
+)
logger = logging.getLogger(__name__)
@@ -134,6 +152,7 @@ def parse_row(cls, columns, model):
"""
pass
+
class MemberExport(BaseExport):
@classmethod
@@ -143,7 +162,7 @@ def model(self):
This is a special edge case, but the base report requires this to be defined.
"""
return None
-
+
@classmethod
def get_model_annotation_dict(cls, request=None, **kwargs):
portfolio = request.session.get("portfolio")
@@ -166,8 +185,12 @@ def get_model_annotation_dict(cls, request=None, **kwargs):
"invitation_date",
"invited_by",
]
- permissions = UserPortfolioPermissionModelAnnotation.get_annotated_queryset(portfolio, csv_report=True).values(*shared_columns)
- invitations = PortfolioInvitationModelAnnotation.get_annotated_queryset(portfolio, csv_report=True).values(*shared_columns)
+ permissions = UserPortfolioPermissionModelAnnotation.get_annotated_queryset(portfolio, csv_report=True).values(
+ *shared_columns
+ )
+ invitations = PortfolioInvitationModelAnnotation.get_annotated_queryset(portfolio, csv_report=True).values(
+ *shared_columns
+ )
queryset_dict = convert_queryset_to_dict(permissions.union(invitations), is_model=False)
return queryset_dict
@@ -199,7 +222,9 @@ def parse_row(cls, columns, model):
roles = model.get("roles")
additional_permissions = model.get("additional_permissions_display")
is_admin = UserPortfolioRoleChoices.ORGANIZATION_ADMIN in (roles or [])
- domain_request_display = UserPortfolioPermission.get_domain_request_permission_display(roles, additional_permissions)
+ domain_request_display = UserPortfolioPermission.get_domain_request_permission_display(
+ roles, additional_permissions
+ )
member_perm_display = UserPortfolioPermission.get_member_permission_display(roles, additional_permissions)
user_managed_domains = model.get("domain_info", [])
managed_domains_as_csv = ",".join(user_managed_domains)
@@ -231,6 +256,7 @@ def parse_row(cls, columns, model):
row = [FIELDS.get(column, "") for column in columns]
return row
+
class DomainExport(BaseExport):
"""
A collection of functions which return csv files regarding Domains. Although class is
@@ -1521,4 +1547,3 @@ def get_creator_active_requests_count_query(cls):
distinct=True,
)
return query
-
diff --git a/src/registrar/utility/model_annotations.py b/src/registrar/utility/model_annotations.py
index 38f00b072..dc6e6ea87 100644
--- a/src/registrar/utility/model_annotations.py
+++ b/src/registrar/utility/model_annotations.py
@@ -20,12 +20,26 @@
permissions = UserPortfolioPermissionAnnotation.get_annotated_queryset(portfolio, csv_report=True)
# Returns same fields but formatted for CSV export
"""
+
from abc import ABC, abstractmethod
from registrar.models import (
DomainInvitation,
PortfolioInvitation,
)
-from django.db.models import CharField, F, ManyToManyField, Q, QuerySet, Value, TextField, OuterRef, Subquery, Func, Case, When
+from django.db.models import (
+ CharField,
+ F,
+ ManyToManyField,
+ Q,
+ QuerySet,
+ Value,
+ TextField,
+ OuterRef,
+ Subquery,
+ Func,
+ Case,
+ When,
+)
from django.db.models.functions import Concat, Coalesce, Cast
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.generic_helper import convert_queryset_to_dict
@@ -39,9 +53,9 @@ class BaseModelAnnotation(ABC):
"""
Abstract base class that standardizes how models are annotated for csv exports or complex annotation queries.
For example, the Members table / csv export.
-
+
Subclasses define model-specific annotations, filters, and field formatting while inheriting
- common queryset building logic.
+ common queryset building logic.
Intended ensure consistent data presentation across both table UI components and CSV exports.
"""
@@ -118,7 +132,7 @@ def get_related_table_fields(cls):
Get a list of fields from related tables.
"""
return []
-
+
@classmethod
def annotate_and_retrieve_fields(
cls, initial_queryset, annotated_fields, related_table_fields=None, include_many_to_many=False, **kwargs
@@ -174,8 +188,7 @@ def get_annotated_queryset(cls, **kwargs):
model_queryset = (
cls.model()
- .objects
- .select_related(*select_related)
+ .objects.select_related(*select_related)
.prefetch_related(*prefetch_related)
.filter(filter_conditions)
.exclude(exclusions)
@@ -183,9 +196,7 @@ def get_annotated_queryset(cls, **kwargs):
.order_by(*sort_fields)
.distinct()
)
- return cls.annotate_and_retrieve_fields(
- model_queryset, annotated_fields, related_table_fields, **kwargs
- )
+ return cls.annotate_and_retrieve_fields(model_queryset, annotated_fields, related_table_fields, **kwargs)
@classmethod
def update_queryset(cls, queryset, **kwargs):
@@ -193,7 +204,7 @@ def update_queryset(cls, queryset, **kwargs):
Returns an updated queryset. Override in subclass to update queryset.
"""
return queryset
-
+
@classmethod
def get_model_annotation_dict(cls, **kwargs):
return convert_queryset_to_dict(cls.get_annotated_queryset(**kwargs), is_model=False)
@@ -205,6 +216,7 @@ class UserPortfolioPermissionModelAnnotation(BaseModelAnnotation):
Handles formatting of user details, permissions, and related domain information
for both UI display and CSV export.
"""
+
@classmethod
def model(cls):
# Return the model class that this export handles
@@ -247,10 +259,7 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
if csv_report:
domain_query = F("user__permissions__domain__name")
last_active_query = Func(
- F("user__last_login"),
- Value("YYYY-MM-DD"),
- function="to_char",
- output_field=TextField()
+ F("user__last_login"), Value("YYYY-MM-DD"), function="to_char", output_field=TextField()
)
else:
domain_query = Concat(
@@ -272,10 +281,7 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
),
"additional_permissions_display": F("additional_permissions"),
"member_display": Case(
- When(
- Q(user__email__isnull=False) & ~Q(user__email=""),
- then=F("user__email")
- ),
+ When(Q(user__email__isnull=False) & ~Q(user__email=""), then=F("user__email")),
When(
Q(user__first_name__isnull=False) | Q(user__last_name__isnull=False),
then=Concat(
@@ -290,17 +296,12 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
"domain_info": ArrayAgg(
domain_query,
distinct=True,
- filter=Q(user__permissions__domain__isnull=False)
+ filter=Q(user__permissions__domain__isnull=False)
& Q(user__permissions__domain__domain_info__portfolio=portfolio),
),
"source": Value("permission", output_field=CharField()),
"invitation_date": Coalesce(
- Func(
- F("invitation__created_at"),
- Value("YYYY-MM-DD"),
- function="to_char",
- output_field=TextField()
- ),
+ Func(F("invitation__created_at"), Value("YYYY-MM-DD"), function="to_char", output_field=TextField()),
Value("Invalid date"),
output_field=TextField(),
),
@@ -311,12 +312,16 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
Subquery(
LogEntry.objects.filter(
content_type=ContentType.objects.get_for_model(PortfolioInvitation),
- object_id=Cast(OuterRef("invitation__id"), output_field=TextField()), # Look up the invitation's ID
- action_flag=ADDITION
- ).order_by("action_time").values("user__email")[:1]
+ object_id=Cast(
+ OuterRef("invitation__id"), output_field=TextField()
+ ), # Look up the invitation's ID
+ action_flag=ADDITION,
+ )
+ .order_by("action_time")
+ .values("user__email")[:1]
),
Value("Unknown"),
- output_field=CharField()
+ output_field=CharField(),
),
}
@@ -394,12 +399,7 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
),
"source": Value("invitation", output_field=CharField()),
"invitation_date": Coalesce(
- Func(
- F("created_at"),
- Value("YYYY-MM-DD"),
- function="to_char",
- output_field=TextField()
- ),
+ Func(F("created_at"), Value("YYYY-MM-DD"), function="to_char", output_field=TextField()),
Value("Invalid date"),
output_field=TextField(),
),
@@ -412,11 +412,13 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
content_type=ContentType.objects.get_for_model(PortfolioInvitation),
# Look up the invitation's ID. LogEntry expects a string as this it is stored as json.
object_id=Cast(OuterRef("id"), output_field=TextField()),
- action_flag=ADDITION
- ).order_by("action_time").values("user__email")[:1]
+ action_flag=ADDITION,
+ )
+ .order_by("action_time")
+ .values("user__email")[:1]
),
Value("Unknown"),
- output_field=CharField()
+ output_field=CharField(),
),
}
diff --git a/src/registrar/views/portfolio_members_json.py b/src/registrar/views/portfolio_members_json.py
index 2fe0492d6..ebe537247 100644
--- a/src/registrar/views/portfolio_members_json.py
+++ b/src/registrar/views/portfolio_members_json.py
@@ -9,7 +9,10 @@
from registrar.models import UserPortfolioPermission
from registrar.models.utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices
-from registrar.utility.model_annotations import PortfolioInvitationModelAnnotation, UserPortfolioPermissionModelAnnotation
+from registrar.utility.model_annotations import (
+ PortfolioInvitationModelAnnotation,
+ UserPortfolioPermissionModelAnnotation,
+)
from registrar.views.utility.mixins import PortfolioMembersPermission
diff --git a/src/registrar/views/portfolios.py b/src/registrar/views/portfolios.py
index 7839d209e..2ad36b71e 100644
--- a/src/registrar/views/portfolios.py
+++ b/src/registrar/views/portfolios.py
@@ -386,6 +386,21 @@ class PortfolioMembersView(PortfolioMembersPermissionView, View):
def get(self, request):
"""Add additional context data to the template."""
return render(request, "portfolio_members.html")
+
+ def get_context_data(self, **kwargs):
+ """Add additional context data to the template."""
+
+ context = super().get_context_data(**kwargs)
+ portfolio = self.request.session.get("portfolio")
+ user_count = portfolio.portfolio_users.count()
+ invitation_count = PortfolioInvitation.objects.filter(
+ portfolio=portfolio
+ ).count()
+ context["member_count"] = user_count + invitation_count
+
+ # check if any portfolio invitations exist 4 portfolio
+ # check if any userportfolioroles exist 4 portfolio
+ return context
class NewMemberView(PortfolioMembersPermissionView, FormMixin):
From 3b378d3c81fbcf53753067260343ebecda8279d1 Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Mon, 18 Nov 2024 11:12:43 -0700
Subject: [PATCH 17/45] Cleanup
---
src/registrar/admin.py | 11 ++-
...te_user_portfolio_permission_invitation.py | 13 +++-
src/registrar/utility/model_annotations.py | 68 ++++++++++---------
src/registrar/views/portfolio_members_json.py | 5 +-
src/registrar/views/portfolios.py | 28 ++++----
5 files changed, 69 insertions(+), 56 deletions(-)
diff --git a/src/registrar/admin.py b/src/registrar/admin.py
index 2d05e760b..1679eeed2 100644
--- a/src/registrar/admin.py
+++ b/src/registrar/admin.py
@@ -1268,6 +1268,15 @@ class Meta:
_meta = Meta()
+ # Question for reviewers: should this include the invitation field?
+ # This is the same layout as before.
+ fieldsets = (
+ (
+ None,
+ {"fields": ("user", "portfolio", "invitation", "roles", "additional_permissions")},
+ ),
+ )
+
# Columns
list_display = [
"user",
@@ -1275,8 +1284,6 @@ class Meta:
"get_roles",
]
- readonly_fields = ["invitation"]
-
autocomplete_fields = ["user", "portfolio"]
search_fields = ["user__first_name", "user__last_name", "user__email", "portfolio__organization_name"]
search_help_text = "Search by first name, last name, email, or portfolio."
diff --git a/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py b/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
index 6e72792ac..7a73f7710 100644
--- a/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
+++ b/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
@@ -13,15 +13,22 @@ class Command(BaseCommand, PopulateScriptTemplate):
def handle(self, **kwargs):
"""Loops through each DomainRequest object and populates
its last_status_update and first_submitted_date values"""
- self.existing_invitations = PortfolioInvitation.objects.filter(portfolio__isnull=False, email__isnull=False).select_related('portfolio')
+ self.existing_invitations = PortfolioInvitation.objects.filter(
+ portfolio__isnull=False, email__isnull=False
+ ).select_related("portfolio")
filter_condition = {"invitation__isnull": True, "portfolio__isnull": False, "user__email__isnull": False}
self.mass_update_records(UserPortfolioPermission, filter_condition, fields_to_update=["invitation"])
def update_record(self, record: UserPortfolioPermission):
"""Associate the invitation to the right object"""
- record.invitation = self.existing_invitations.filter(email=record.user.email, portfolio=record.portfolio).first()
+ record.invitation = self.existing_invitations.filter(
+ email=record.user.email, portfolio=record.portfolio
+ ).first()
TerminalHelper.colorful_logger("INFO", "OKCYAN", f"{TerminalColors.OKCYAN}Adding invitation to {record}")
def should_skip_record(self, record) -> bool:
"""There is nothing to add if no invitation exists"""
- return not record or not self.existing_invitations.filter(email=record.user.email, portfolio=record.portfolio).exists()
+ return (
+ not record
+ or not self.existing_invitations.filter(email=record.user.email, portfolio=record.portfolio).exists()
+ )
diff --git a/src/registrar/utility/model_annotations.py b/src/registrar/utility/model_annotations.py
index dc6e6ea87..105296855 100644
--- a/src/registrar/utility/model_annotations.py
+++ b/src/registrar/utility/model_annotations.py
@@ -39,8 +39,10 @@
Func,
Case,
When,
+ Exists,
)
from django.db.models.functions import Concat, Coalesce, Cast
+from registrar.models.user_group import UserGroup
from registrar.models.user_portfolio_permission import UserPortfolioPermission
from registrar.models.utility.generic_helper import convert_queryset_to_dict
from registrar.models.utility.orm_helper import ArrayRemove
@@ -305,23 +307,8 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
Value("Invalid date"),
output_field=TextField(),
),
- # TODO - replace this with a "creator" field on portfolio invitation. This should be another ticket.
- # Grab the invitation creator from the audit log. This will need to be replaced with a creator field.
- # When that happens, just replace this with F("invitation__creator")
- "invited_by": Coalesce(
- Subquery(
- LogEntry.objects.filter(
- content_type=ContentType.objects.get_for_model(PortfolioInvitation),
- object_id=Cast(
- OuterRef("invitation__id"), output_field=TextField()
- ), # Look up the invitation's ID
- action_flag=ADDITION,
- )
- .order_by("action_time")
- .values("user__email")[:1]
- ),
- Value("Unknown"),
- output_field=CharField(),
+ "invited_by": PortfolioInvitationModelAnnotation.get_invited_by_from_audit_log_query(
+ object_id_query=Cast(OuterRef("invitation__id"), output_field=TextField())
),
}
@@ -362,6 +349,37 @@ def get_filter_conditions(cls, portfolio, **kwargs):
# Get all members on this portfolio
return Q(portfolio=portfolio)
+ @classmethod
+ def get_invited_by_from_audit_log_query(cls, object_id_query):
+ return Coalesce(
+ Subquery(
+ LogEntry.objects.filter(
+ content_type=ContentType.objects.get_for_model(PortfolioInvitation),
+ object_id=object_id_query,
+ action_flag=ADDITION,
+ )
+ .annotate(
+ display_email=Case(
+ When(
+ Exists(
+ UserGroup.objects.filter(
+ name__in=["cisa_analysts_group", "full_access_group"],
+ user=OuterRef("user"),
+ )
+ ),
+ then=Value("help@get.gov")
+ ),
+ default=F("user__email"),
+ output_field=CharField()
+ )
+ )
+ .order_by("action_time")
+ .values("display_email")
+ ),
+ Value("Unknown"),
+ output_field=CharField(),
+ )
+
@classmethod
def get_annotated_fields(cls, portfolio, csv_report=False):
"""
@@ -383,7 +401,6 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
email=OuterRef("email"), # Check if email matches the OuterRef("email")
domain__domain_info__portfolio=portfolio, # Check if the domain's portfolio matches the given portfolio
).annotate(domain_info=domain_query)
-
return {
"first_name": Value(None, output_field=CharField()),
"last_name": Value(None, output_field=CharField()),
@@ -406,19 +423,8 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
# TODO - replace this with a "creator" field on portfolio invitation. This should be another ticket.
# Grab the invitation creator from the audit log. This will need to be replaced with a creator field.
# When that happens, just replace this with F("invitation__creator")
- "invited_by": Coalesce(
- Subquery(
- LogEntry.objects.filter(
- content_type=ContentType.objects.get_for_model(PortfolioInvitation),
- # Look up the invitation's ID. LogEntry expects a string as this it is stored as json.
- object_id=Cast(OuterRef("id"), output_field=TextField()),
- action_flag=ADDITION,
- )
- .order_by("action_time")
- .values("user__email")[:1]
- ),
- Value("Unknown"),
- output_field=CharField(),
+ "invited_by": cls.get_invited_by_from_audit_log_query(
+ object_id_query=Cast(OuterRef("id"), output_field=TextField())
),
}
diff --git a/src/registrar/views/portfolio_members_json.py b/src/registrar/views/portfolio_members_json.py
index ebe537247..3bf761858 100644
--- a/src/registrar/views/portfolio_members_json.py
+++ b/src/registrar/views/portfolio_members_json.py
@@ -1,9 +1,6 @@
from django.http import JsonResponse
from django.core.paginator import Paginator
-from django.db.models import Value, F, CharField, TextField, Q, Case, When, OuterRef, Subquery
-from django.db.models.expressions import Func
-from django.db.models.functions import Cast, Coalesce, Concat
-from django.contrib.postgres.aggregates import ArrayAgg
+from django.db.models import F, Q
from django.urls import reverse
from django.views import View
diff --git a/src/registrar/views/portfolios.py b/src/registrar/views/portfolios.py
index 2ad36b71e..09ff159c5 100644
--- a/src/registrar/views/portfolios.py
+++ b/src/registrar/views/portfolios.py
@@ -385,22 +385,18 @@ class PortfolioMembersView(PortfolioMembersPermissionView, View):
def get(self, request):
"""Add additional context data to the template."""
- return render(request, "portfolio_members.html")
-
- def get_context_data(self, **kwargs):
- """Add additional context data to the template."""
-
- context = super().get_context_data(**kwargs)
- portfolio = self.request.session.get("portfolio")
- user_count = portfolio.portfolio_users.count()
- invitation_count = PortfolioInvitation.objects.filter(
- portfolio=portfolio
- ).count()
- context["member_count"] = user_count + invitation_count
-
- # check if any portfolio invitations exist 4 portfolio
- # check if any userportfolioroles exist 4 portfolio
- return context
+ # Get portfolio from session
+ portfolio = request.session.get("portfolio")
+ context = {}
+ if portfolio:
+ user_count = portfolio.portfolio_users.count()
+ invitation_count = PortfolioInvitation.objects.filter(
+ portfolio=portfolio
+ ).count()
+ context.update({
+ "member_count": user_count + invitation_count
+ })
+ return render(request, "portfolio_members.html", context=context)
class NewMemberView(PortfolioMembersPermissionView, FormMixin):
From 1c0f99eedc2f39502ecdde28df6dc011cc8af1c6 Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Mon, 18 Nov 2024 14:01:00 -0700
Subject: [PATCH 18/45] Use audit log instead
---
src/registrar/admin.py | 9 ---
...te_user_portfolio_permission_invitation.py | 34 ---------
...0138_userportfoliopermission_invitation.py | 25 -------
src/registrar/models/portfolio_invitation.py | 4 +-
.../models/user_portfolio_permission.py | 10 ---
src/registrar/utility/model_annotations.py | 70 +++++++++++++++----
6 files changed, 58 insertions(+), 94 deletions(-)
delete mode 100644 src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
delete mode 100644 src/registrar/migrations/0138_userportfoliopermission_invitation.py
diff --git a/src/registrar/admin.py b/src/registrar/admin.py
index 1679eeed2..0cab01d31 100644
--- a/src/registrar/admin.py
+++ b/src/registrar/admin.py
@@ -1268,15 +1268,6 @@ class Meta:
_meta = Meta()
- # Question for reviewers: should this include the invitation field?
- # This is the same layout as before.
- fieldsets = (
- (
- None,
- {"fields": ("user", "portfolio", "invitation", "roles", "additional_permissions")},
- ),
- )
-
# Columns
list_display = [
"user",
diff --git a/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py b/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
deleted file mode 100644
index 7a73f7710..000000000
--- a/src/registrar/management/commands/populate_user_portfolio_permission_invitation.py
+++ /dev/null
@@ -1,34 +0,0 @@
-import logging
-from django.core.management import BaseCommand
-from registrar.management.commands.utility.terminal_helper import PopulateScriptTemplate, TerminalColors, TerminalHelper
-from registrar.models import UserPortfolioPermission, PortfolioInvitation
-from auditlog.models import LogEntry
-
-logger = logging.getLogger(__name__)
-
-
-class Command(BaseCommand, PopulateScriptTemplate):
- help = "Loops through each UserPortfolioPermission object and populates the invitation field"
-
- def handle(self, **kwargs):
- """Loops through each DomainRequest object and populates
- its last_status_update and first_submitted_date values"""
- self.existing_invitations = PortfolioInvitation.objects.filter(
- portfolio__isnull=False, email__isnull=False
- ).select_related("portfolio")
- filter_condition = {"invitation__isnull": True, "portfolio__isnull": False, "user__email__isnull": False}
- self.mass_update_records(UserPortfolioPermission, filter_condition, fields_to_update=["invitation"])
-
- def update_record(self, record: UserPortfolioPermission):
- """Associate the invitation to the right object"""
- record.invitation = self.existing_invitations.filter(
- email=record.user.email, portfolio=record.portfolio
- ).first()
- TerminalHelper.colorful_logger("INFO", "OKCYAN", f"{TerminalColors.OKCYAN}Adding invitation to {record}")
-
- def should_skip_record(self, record) -> bool:
- """There is nothing to add if no invitation exists"""
- return (
- not record
- or not self.existing_invitations.filter(email=record.user.email, portfolio=record.portfolio).exists()
- )
diff --git a/src/registrar/migrations/0138_userportfoliopermission_invitation.py b/src/registrar/migrations/0138_userportfoliopermission_invitation.py
deleted file mode 100644
index abac42ae2..000000000
--- a/src/registrar/migrations/0138_userportfoliopermission_invitation.py
+++ /dev/null
@@ -1,25 +0,0 @@
-# Generated by Django 4.2.10 on 2024-11-14 21:05
-
-from django.db import migrations, models
-import django.db.models.deletion
-
-
-class Migration(migrations.Migration):
-
- dependencies = [
- ("registrar", "0137_suborganization_city_suborganization_state_territory"),
- ]
-
- operations = [
- migrations.AddField(
- model_name="userportfoliopermission",
- name="invitation",
- field=models.ForeignKey(
- blank=True,
- null=True,
- on_delete=django.db.models.deletion.PROTECT,
- related_name="created_user_portfolio_permission",
- to="registrar.portfolioinvitation",
- ),
- ),
- ]
diff --git a/src/registrar/models/portfolio_invitation.py b/src/registrar/models/portfolio_invitation.py
index 0ea410211..61a6b7397 100644
--- a/src/registrar/models/portfolio_invitation.py
+++ b/src/registrar/models/portfolio_invitation.py
@@ -9,8 +9,6 @@
from .utility.portfolio_helper import UserPortfolioPermissionChoices, UserPortfolioRoleChoices # type: ignore
from .utility.time_stamped_model import TimeStampedModel
from django.contrib.postgres.fields import ArrayField
-from django.contrib.admin.models import LogEntry, ADDITION
-from django.contrib.contenttypes.models import ContentType
logger = logging.getLogger(__name__)
@@ -103,7 +101,7 @@ def retrieve(self):
# and create a role for that user on this portfolio
user_portfolio_permission, _ = UserPortfolioPermission.objects.get_or_create(
- portfolio=self.portfolio, user=user, invitation=self
+ portfolio=self.portfolio, user=user
)
if self.roles and len(self.roles) > 0:
user_portfolio_permission.roles = self.roles
diff --git a/src/registrar/models/user_portfolio_permission.py b/src/registrar/models/user_portfolio_permission.py
index 424f09a17..cd48b1b71 100644
--- a/src/registrar/models/user_portfolio_permission.py
+++ b/src/registrar/models/user_portfolio_permission.py
@@ -65,16 +65,6 @@ class Meta:
help_text="Select one or more additional permissions.",
)
- # TODO - this needs a small script to update existing values
- invitation = models.ForeignKey(
- "registrar.PortfolioInvitation",
- null=True,
- blank=True,
- # We don't want to accidentally delete invitations
- on_delete=models.PROTECT,
- related_name="created_user_portfolio_permission",
- )
-
def __str__(self):
readable_roles = []
if self.roles:
diff --git a/src/registrar/utility/model_annotations.py b/src/registrar/utility/model_annotations.py
index 105296855..bd1ad2be0 100644
--- a/src/registrar/utility/model_annotations.py
+++ b/src/registrar/utility/model_annotations.py
@@ -243,6 +243,22 @@ def get_filter_conditions(cls, portfolio, **kwargs):
# Get all members on this portfolio
return Q(portfolio=portfolio)
+ @classmethod
+ def get_portfolio_invitation_id_query(cls):
+ """Gets the id of the portfolio invitation that created this UserPortfolioPermission.
+ This makes the assumption that if an invitation is retrieved, it must have created the given
+ UserPortfolioPermission object."""
+ return Cast(
+ Subquery(
+ PortfolioInvitation.objects.filter(
+ status=PortfolioInvitation.PortfolioInvitationStatus.RETRIEVED,
+ email=OuterRef(OuterRef("user__email")),
+ portfolio=OuterRef(OuterRef("portfolio"))
+ ).values("id")[:1]
+ ),
+ output_field=TextField()
+ )
+
@classmethod
def get_annotated_fields(cls, portfolio, csv_report=False):
"""
@@ -302,13 +318,11 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
& Q(user__permissions__domain__domain_info__portfolio=portfolio),
),
"source": Value("permission", output_field=CharField()),
- "invitation_date": Coalesce(
- Func(F("invitation__created_at"), Value("YYYY-MM-DD"), function="to_char", output_field=TextField()),
- Value("Invalid date"),
- output_field=TextField(),
+ "invitation_date": PortfolioInvitationModelAnnotation.get_invitation_date_query(
+ object_id_query=cls.get_portfolio_invitation_id_query()
),
- "invited_by": PortfolioInvitationModelAnnotation.get_invited_by_from_audit_log_query(
- object_id_query=Cast(OuterRef("invitation__id"), output_field=TextField())
+ "invited_by": PortfolioInvitationModelAnnotation.get_invited_by_query(
+ object_id_query=cls.get_portfolio_invitation_id_query()
),
}
@@ -350,7 +364,39 @@ def get_filter_conditions(cls, portfolio, **kwargs):
return Q(portfolio=portfolio)
@classmethod
- def get_invited_by_from_audit_log_query(cls, object_id_query):
+ def get_invitation_date_query(cls, object_id_query):
+ """Returns the date at which the given invitation was created.
+ Grabs this data from the audit log, given that a portfolio invitation object
+ is specified via object_id_query."""
+ return Coalesce(
+ Subquery(
+ LogEntry.objects.filter(
+ content_type=ContentType.objects.get_for_model(PortfolioInvitation),
+ object_id=object_id_query,
+ action_flag=ADDITION,
+ )
+ .annotate(
+ # Action time will always be equivalent to created_at in this context
+ display_date=Func(
+ F("action_time"),
+ Value("YYYY-MM-DD"),
+ function="to_char",
+ output_field=TextField()
+ )
+ )
+ .order_by("action_time")
+ .values("display_date")[:1]
+ ),
+ Value("Invalid date"),
+ output_field=TextField()
+ )
+
+
+ @classmethod
+ def get_invited_by_query(cls, object_id_query):
+ """Returns the user that created the given portfolio invitation.
+ Grabs this data from the audit log, given that a portfolio invitation object
+ is specified via object_id_query."""
return Coalesce(
Subquery(
LogEntry.objects.filter(
@@ -374,7 +420,7 @@ def get_invited_by_from_audit_log_query(cls, object_id_query):
)
)
.order_by("action_time")
- .values("display_email")
+ .values("display_email")[:1]
),
Value("Unknown"),
output_field=CharField(),
@@ -415,15 +461,13 @@ def get_annotated_fields(cls, portfolio, csv_report=False):
)
),
"source": Value("invitation", output_field=CharField()),
- "invitation_date": Coalesce(
- Func(F("created_at"), Value("YYYY-MM-DD"), function="to_char", output_field=TextField()),
- Value("Invalid date"),
- output_field=TextField(),
+ "invitation_date": cls.get_invitation_date_query(
+ object_id_query=Cast(OuterRef("id"), output_field=TextField())
),
# TODO - replace this with a "creator" field on portfolio invitation. This should be another ticket.
# Grab the invitation creator from the audit log. This will need to be replaced with a creator field.
# When that happens, just replace this with F("invitation__creator")
- "invited_by": cls.get_invited_by_from_audit_log_query(
+ "invited_by": cls.get_invited_by_query(
object_id_query=Cast(OuterRef("id"), output_field=TextField())
),
}
From 5f5bc4f616c95d2f739790614f53ed5de24961a9 Mon Sep 17 00:00:00 2001
From: zandercymatics <141044360+zandercymatics@users.noreply.github.com>
Date: Mon, 18 Nov 2024 14:17:19 -0700
Subject: [PATCH 19/45] lint
Just need unit tests!
---
src/registrar/models/utility/orm_helper.py | 2 +-
.../templates/includes/members_table.html | 2 +-
src/registrar/utility/csv_export.py | 12 ++------
src/registrar/utility/model_annotations.py | 30 +++++++++----------
src/registrar/views/portfolios.py | 8 ++---
5 files changed, 20 insertions(+), 34 deletions(-)
diff --git a/src/registrar/models/utility/orm_helper.py b/src/registrar/models/utility/orm_helper.py
index 4f4665216..63ff41d28 100644
--- a/src/registrar/models/utility/orm_helper.py
+++ b/src/registrar/models/utility/orm_helper.py
@@ -1,7 +1,7 @@
from django.db.models.expressions import Func
-class ArrayRemove(Func):
+class ArrayRemoveNull(Func):
"""Custom Func to use array_remove to remove null values"""
function = "array_remove"
diff --git a/src/registrar/templates/includes/members_table.html b/src/registrar/templates/includes/members_table.html
index 09cebda2e..c0c6d803c 100644
--- a/src/registrar/templates/includes/members_table.html
+++ b/src/registrar/templates/includes/members_table.html
@@ -8,7 +8,7 @@