We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
The meta data report is not able to download on stable
### Current Behavior

The meta data report is not able to download on stable and results in an error

### Expected Behavior

The meta data report should download

### Steps to Reproduce

1.go to manage.get.gov/admin/registrar/domain
2. click the meta data report
3. see it takes for ever to load and throws an error


### Environment

stable

works fine on others

### Additional Context

Adding additional columns on the meta data report has resulted in a lot more computation time when it comes to our full data set. Look at the logic around columns and you will see some duplication that could be reduced. 

### Issue Links

Relates to #1840 and #1556  

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/registrar/utility/csv_export.py`
Content:
```
1 import csv
2 import logging
3 from datetime import datetime
4 from registrar.models.domain import Domain
5 from registrar.models.domain_information import DomainInformation
6 from django.utils import timezone
7 from django.core.paginator import Paginator
8 from django.db.models import F, Value, CharField
9 from django.db.models.functions import Concat, Coalesce
10 
11 from registrar.models.public_contact import PublicContact
12 from registrar.utility.enums import DefaultEmail
13 
14 logger = logging.getLogger(__name__)
15 
16 
17 def write_header(writer, columns):
18     """
19     Receives params from the parent methods and outputs a CSV with a header row.
20     Works with write_header as long as the same writer object is passed.
21     """
22 
23     writer.writerow(columns)
24 
25 
26 def get_domain_infos(filter_condition, sort_fields):
27     domain_infos = (
28         DomainInformation.objects.select_related("domain", "authorizing_official")
29         .filter(**filter_condition)
30         .order_by(*sort_fields)
31     )
32 
33     # Do a mass concat of the first and last name fields for authorizing_official.
34     # The old operation was computationally heavy for some reason, so if we precompute
35     # this here, it is vastly more efficient.
36     domain_infos_cleaned = domain_infos.annotate(
37         ao=Concat(
38             Coalesce(F("authorizing_official__first_name"), Value("")),
39             Value(" "),
40             Coalesce(F("authorizing_official__last_name"), Value("")),
41             output_field=CharField(),
42         )
43     )
44     return domain_infos_cleaned
45 
46 
47 def parse_row(columns, domain_info: DomainInformation, security_emails_dict=None, get_domain_managers=False):
48     """Given a set of columns, generate a new row from cleaned column data"""
49 
50     # Domain should never be none when parsing this information
51     if domain_info.domain is None:
52         raise ValueError("Domain is none")
53 
54     domain = domain_info.domain  # type: ignore
55 
56     # Grab the security email from a preset dictionary.
57     # If nothing exists in the dictionary, grab from .contacts.
58     if security_emails_dict is not None and domain.name in security_emails_dict:
59         _email = security_emails_dict.get(domain.name)
60         security_email = _email if _email is not None else " "
61     else:
62         # If the dictionary doesn't contain that data, lets filter for it manually.
63         # This is a last resort as this is a more expensive operation.
64         security_contacts = domain.contacts.filter(contact_type=PublicContact.ContactTypeChoices.SECURITY)
65         _email = security_contacts[0].email if security_contacts else None
66         security_email = _email if _email is not None else " "
67 
68     # These are default emails that should not be displayed in the csv report
69     invalid_emails = {DefaultEmail.LEGACY_DEFAULT.value, DefaultEmail.PUBLIC_CONTACT_DEFAULT.value}
70     if security_email.lower() in invalid_emails:
71         security_email = "(blank)"
72 
73     if domain_info.federal_type:
74         domain_type = f"{domain_info.get_organization_type_display()} - {domain_info.get_federal_type_display()}"
75     else:
76         domain_type = domain_info.get_organization_type_display()
77 
78     # create a dictionary of fields which can be included in output
79     FIELDS = {
80         "Domain name": domain.name,
81         "Status": domain.get_state_display(),
82         "Expiration date": domain.expiration_date,
83         "Domain type": domain_type,
84         "Agency": domain_info.federal_agency,
85         "Organization name": domain_info.organization_name,
86         "City": domain_info.city,
87         "State": domain_info.state_territory,
88         "AO": domain_info.ao,  # type: ignore
89         "AO email": domain_info.authorizing_official.email if domain_info.authorizing_official else " ",
90         "Security contact email": security_email,
91         "Created at": domain.created_at,
92         "First ready": domain.first_ready,
93         "Deleted": domain.deleted,
94     }
95 
96     if get_domain_managers:
97         # Get each domain managers email and add to list
98         dm_emails = [dm.user.email for dm in domain.permissions.all()]
99 
100         # Set up the "matching header" + row field data
101         for i, dm_email in enumerate(dm_emails, start=1):
102             FIELDS[f"Domain manager email {i}"] = dm_email
103 
104     row = [FIELDS.get(column, "") for column in columns]
105     return row
106 
107 
108 def _get_security_emails(sec_contact_ids):
109     """
110     Retrieve security contact emails for the given security contact IDs.
111     """
112     security_emails_dict = {}
113     public_contacts = (
114         PublicContact.objects.only("email", "domain__name")
115         .select_related("domain")
116         .filter(registry_id__in=sec_contact_ids)
117     )
118 
119     # Populate a dictionary of domain names and their security contacts
120     for contact in public_contacts:
121         domain: Domain = contact.domain
122         if domain is not None and domain.name not in security_emails_dict:
123             security_emails_dict[domain.name] = contact.email
124         else:
125             logger.warning("csv_export -> Domain was none for PublicContact")
126 
127     return security_emails_dict
128 
129 
130 def update_columns_with_domain_managers(columns, max_dm_count):
131     """
132     Update the columns list to include "Domain manager email {#}" headers
133     based on the maximum domain manager count.
134     """
135     for i in range(1, max_dm_count + 1):
136         columns.append(f"Domain manager email {i}")
137 
138 
139 def write_csv(
140     writer,
141     columns,
142     sort_fields,
143     filter_condition,
144     get_domain_managers=False,
145     should_write_header=True,
146 ):
147     """
148     Receives params from the parent methods and outputs a CSV with fltered and sorted domains.
149     Works with write_header as longas the same writer object is passed.
150     get_domain_managers: Conditional bc we only use domain manager info for export_data_full_to_csv
151     should_write_header: Conditional bc export_data_growth_to_csv calls write_body twice
152     """
153 
154     all_domain_infos = get_domain_infos(filter_condition, sort_fields)
155 
156     # Store all security emails to avoid epp calls or excessive filters
157     sec_contact_ids = all_domain_infos.values_list("domain__security_contact_registry_id", flat=True)
158 
159     security_emails_dict = _get_security_emails(sec_contact_ids)
160 
161     # Reduce the memory overhead when performing the write operation
162     paginator = Paginator(all_domain_infos, 1000)
163 
164     if get_domain_managers and len(all_domain_infos) > 0:
165         # We want to get the max amont of domain managers an
166         # account has to set the column header dynamically
167         max_dm_count = max(len(domain_info.domain.permissions.all()) for domain_info in all_domain_infos)
168         update_columns_with_domain_managers(columns, max_dm_count)
169 
170     if should_write_header:
171         write_header(writer, columns)
172 
173     for page_num in paginator.page_range:
174         rows = []
175         page = paginator.page(page_num)
176         for domain_info in page.object_list:
177             try:
178                 row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)
179                 rows.append(row)
180             except ValueError:
181                 # This should not happen. If it does, just skip this row.
182                 # It indicates that DomainInformation.domain is None.
183                 logger.error("csv_export -> Error when parsing row, domain was None")
184                 continue
185 
186         writer.writerows(rows)
187 
188 
189 def export_data_type_to_csv(csv_file):
190     """All domains report with extra columns"""
191 
192     writer = csv.writer(csv_file)
193     # define columns to include in export
194     columns = [
195         "Domain name",
196         "Status",
197         "Expiration date",
198         "Domain type",
199         "Agency",
200         "Organization name",
201         "City",
202         "State",
203         "AO",
204         "AO email",
205         "Security contact email",
206         # For domain manager we are pass it in as a parameter below in write_body
207     ]
208 
209     # Coalesce is used to replace federal_type of None with ZZZZZ
210     sort_fields = [
211         "organization_type",
212         Coalesce("federal_type", Value("ZZZZZ")),
213         "federal_agency",
214         "domain__name",
215     ]
216     filter_condition = {
217         "domain__state__in": [
218             Domain.State.READY,
219             Domain.State.DNS_NEEDED,
220             Domain.State.ON_HOLD,
221         ],
222     }
223     write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=True, should_write_header=True)
224 
225 
226 def export_data_full_to_csv(csv_file):
227     """All domains report"""
228 
229     writer = csv.writer(csv_file)
230     # define columns to include in export
231     columns = [
232         "Domain name",
233         "Domain type",
234         "Agency",
235         "Organization name",
236         "City",
237         "State",
238         "Security contact email",
239     ]
240     # Coalesce is used to replace federal_type of None with ZZZZZ
241     sort_fields = [
242         "organization_type",
243         Coalesce("federal_type", Value("ZZZZZ")),
244         "federal_agency",
245         "domain__name",
246     ]
247     filter_condition = {
248         "domain__state__in": [
249             Domain.State.READY,
250             Domain.State.DNS_NEEDED,
251             Domain.State.ON_HOLD,
252         ],
253     }
254     write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
255 
256 
257 def export_data_federal_to_csv(csv_file):
258     """Federal domains report"""
259 
260     writer = csv.writer(csv_file)
261     # define columns to include in export
262     columns = [
263         "Domain name",
264         "Domain type",
265         "Agency",
266         "Organization name",
267         "City",
268         "State",
269         "Security contact email",
270     ]
271     # Coalesce is used to replace federal_type of None with ZZZZZ
272     sort_fields = [
273         "organization_type",
274         Coalesce("federal_type", Value("ZZZZZ")),
275         "federal_agency",
276         "domain__name",
277     ]
278     filter_condition = {
279         "organization_type__icontains": "federal",
280         "domain__state__in": [
281             Domain.State.READY,
282             Domain.State.DNS_NEEDED,
283             Domain.State.ON_HOLD,
284         ],
285     }
286     write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
287 
288 
289 def get_default_start_date():
290     # Default to a date that's prior to our first deployment
291     return timezone.make_aware(datetime(2023, 11, 1))
292 
293 
294 def get_default_end_date():
295     # Default to now()
296     return timezone.now()
297 
298 
299 def export_data_growth_to_csv(csv_file, start_date, end_date):
300     """
301     Growth report:
302     Receive start and end dates from the view, parse them.
303     Request from write_body READY domains that are created between
304     the start and end dates, as well as DELETED domains that are deleted between
305     the start and end dates. Specify sort params for both lists.
306     """
307 
308     start_date_formatted = (
309         timezone.make_aware(datetime.strptime(start_date, "%Y-%m-%d")) if start_date else get_default_start_date()
310     )
311 
312     end_date_formatted = (
313         timezone.make_aware(datetime.strptime(end_date, "%Y-%m-%d")) if end_date else get_default_end_date()
314     )
315 
316     writer = csv.writer(csv_file)
317 
318     # define columns to include in export
319     columns = [
320         "Domain name",
321         "Domain type",
322         "Agency",
323         "Organization name",
324         "City",
325         "State",
326         "Status",
327         "Expiration date",
328         "Created at",
329         "First ready",
330         "Deleted",
331     ]
332     sort_fields = [
333         "domain__first_ready",
334         "domain__name",
335     ]
336     filter_condition = {
337         "domain__state__in": [Domain.State.READY],
338         "domain__first_ready__lte": end_date_formatted,
339         "domain__first_ready__gte": start_date_formatted,
340     }
341 
342     # We also want domains deleted between sar and end dates, sorted
343     sort_fields_for_deleted_domains = [
344         "domain__deleted",
345         "domain__name",
346     ]
347     filter_condition_for_deleted_domains = {
348         "domain__state__in": [Domain.State.DELETED],
349         "domain__deleted__lte": end_date_formatted,
350         "domain__deleted__gte": start_date_formatted,
351     }
352 
353     write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)
354     write_csv(
355         writer,
356         columns,
357         sort_fields_for_deleted_domains,
358         filter_condition_for_deleted_domains,
359         get_domain_managers=False,
360         should_write_header=False,
361     )
362
```
--- END FILES ---

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

Here is an example:

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

```

diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py
--- a/src/registrar/utility/csv_export.py
+++ b/src/registrar/utility/csv_export.py
@@ -26,6 +26,7 @@
 def get_domain_infos(filter_condition, sort_fields):
     domain_infos = (
         DomainInformation.objects.select_related("domain", "authorizing_official")
+        .prefetch_related("domain__permissions")
         .filter(**filter_condition)
         .order_by(*sort_fields)
     )
@@ -49,6 +50,7 @@
 
     # Domain should never be none when parsing this information
     if domain_info.domain is None:
+        logger.error("Attemting to parse row for csv exports but Domain is none in a DomainInfo")
         raise ValueError("Domain is none")
 
     domain = domain_info.domain  # type: ignore
@@ -127,15 +129,6 @@
     return security_emails_dict
 
 
-def update_columns_with_domain_managers(columns, max_dm_count):
-    """
-    Update the columns list to include "Domain manager email {#}" headers
-    based on the maximum domain manager count.
-    """
-    for i in range(1, max_dm_count + 1):
-        columns.append(f"Domain manager email {i}")
-
-
 def write_csv(
     writer,
     columns,
@@ -161,19 +154,26 @@
     # Reduce the memory overhead when performing the write operation
     paginator = Paginator(all_domain_infos, 1000)
 
-    if get_domain_managers and len(all_domain_infos) > 0:
-        # We want to get the max amont of domain managers an
-        # account has to set the column header dynamically
-        max_dm_count = max(len(domain_info.domain.permissions.all()) for domain_info in all_domain_infos)
-        update_columns_with_domain_managers(columns, max_dm_count)
-
-    if should_write_header:
-        write_header(writer, columns)
+    # The maximum amount of domain managers an account has
+    # We get the max so we can set the column header accurately
+    max_dm_count = 0
+    total_body_rows = []
 
     for page_num in paginator.page_range:
         rows = []
         page = paginator.page(page_num)
         for domain_info in page.object_list:
+
+            # Get count of all the domain managers for an account
+            if get_domain_managers:
+                dm_count = domain_info.domain.permissions.count()
+                if dm_count > max_dm_count:
+                    max_dm_count = dm_count
+                    for i in range(1, max_dm_count + 1):
+                        column_name = f"Domain manager email {i}"
+                        if column_name not in columns:
+                            columns.append(column_name)
+
             try:
                 row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)
                 rows.append(row)
@@ -182,8 +182,11 @@
                 # It indicates that DomainInformation.domain is None.
                 logger.error("csv_export -> Error when parsing row, domain was None")
                 continue
+        total_body_rows.extend(rows)
 
-        writer.writerows(rows)
+    if should_write_header:
+        write_header(writer, columns)
+    writer.writerows(total_body_rows)
 
 
 def export_data_type_to_csv(csv_file):

{"golden_diff": "diff --git a/src/registrar/utility/csv_export.py b/src/registrar/utility/csv_export.py\n--- a/src/registrar/utility/csv_export.py\n+++ b/src/registrar/utility/csv_export.py\n@@ -26,6 +26,7 @@\n def get_domain_infos(filter_condition, sort_fields):\n     domain_infos = (\n         DomainInformation.objects.select_related(\"domain\", \"authorizing_official\")\n+        .prefetch_related(\"domain__permissions\")\n         .filter(**filter_condition)\n         .order_by(*sort_fields)\n     )\n@@ -49,6 +50,7 @@\n \n     # Domain should never be none when parsing this information\n     if domain_info.domain is None:\n+        logger.error(\"Attemting to parse row for csv exports but Domain is none in a DomainInfo\")\n         raise ValueError(\"Domain is none\")\n \n     domain = domain_info.domain  # type: ignore\n@@ -127,15 +129,6 @@\n     return security_emails_dict\n \n \n-def update_columns_with_domain_managers(columns, max_dm_count):\n-    \"\"\"\n-    Update the columns list to include \"Domain manager email {#}\" headers\n-    based on the maximum domain manager count.\n-    \"\"\"\n-    for i in range(1, max_dm_count + 1):\n-        columns.append(f\"Domain manager email {i}\")\n-\n-\n def write_csv(\n     writer,\n     columns,\n@@ -161,19 +154,26 @@\n     # Reduce the memory overhead when performing the write operation\n     paginator = Paginator(all_domain_infos, 1000)\n \n-    if get_domain_managers and len(all_domain_infos) > 0:\n-        # We want to get the max amont of domain managers an\n-        # account has to set the column header dynamically\n-        max_dm_count = max(len(domain_info.domain.permissions.all()) for domain_info in all_domain_infos)\n-        update_columns_with_domain_managers(columns, max_dm_count)\n-\n-    if should_write_header:\n-        write_header(writer, columns)\n+    # The maximum amount of domain managers an account has\n+    # We get the max so we can set the column header accurately\n+    max_dm_count = 0\n+    total_body_rows = []\n \n     for page_num in paginator.page_range:\n         rows = []\n         page = paginator.page(page_num)\n         for domain_info in page.object_list:\n+\n+            # Get count of all the domain managers for an account\n+            if get_domain_managers:\n+                dm_count = domain_info.domain.permissions.count()\n+                if dm_count > max_dm_count:\n+                    max_dm_count = dm_count\n+                    for i in range(1, max_dm_count + 1):\n+                        column_name = f\"Domain manager email {i}\"\n+                        if column_name not in columns:\n+                            columns.append(column_name)\n+\n             try:\n                 row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)\n                 rows.append(row)\n@@ -182,8 +182,11 @@\n                 # It indicates that DomainInformation.domain is None.\n                 logger.error(\"csv_export -> Error when parsing row, domain was None\")\n                 continue\n+        total_body_rows.extend(rows)\n \n-        writer.writerows(rows)\n+    if should_write_header:\n+        write_header(writer, columns)\n+    writer.writerows(total_body_rows)\n \n \n def export_data_type_to_csv(csv_file):\n", "issue": "The meta data report is not able to download on stable\n### Current Behavior\n\nThe meta data report is not able to download on stable and results in an error\n\n### Expected Behavior\n\nThe meta data report should download\n\n### Steps to Reproduce\n\n1.go to manage.get.gov/admin/registrar/domain\r\n2. click the meta data report\r\n3. see it takes for ever to load and throws an error\r\n\n\n### Environment\n\nstable\r\n\r\nworks fine on others\n\n### Additional Context\n\nAdding additional columns on the meta data report has resulted in a lot more computation time when it comes to our full data set. Look at the logic around columns and you will see some duplication that could be reduced. \n\n### Issue Links\n\nRelates to #1840 and #1556  \n", "before_files": [{"content": "import csv\nimport logging\nfrom datetime import datetime\nfrom registrar.models.domain import Domain\nfrom registrar.models.domain_information import DomainInformation\nfrom django.utils import timezone\nfrom django.core.paginator import Paginator\nfrom django.db.models import F, Value, CharField\nfrom django.db.models.functions import Concat, Coalesce\n\nfrom registrar.models.public_contact import PublicContact\nfrom registrar.utility.enums import DefaultEmail\n\nlogger = logging.getLogger(__name__)\n\n\ndef write_header(writer, columns):\n    \"\"\"\n    Receives params from the parent methods and outputs a CSV with a header row.\n    Works with write_header as long as the same writer object is passed.\n    \"\"\"\n\n    writer.writerow(columns)\n\n\ndef get_domain_infos(filter_condition, sort_fields):\n    domain_infos = (\n        DomainInformation.objects.select_related(\"domain\", \"authorizing_official\")\n        .filter(**filter_condition)\n        .order_by(*sort_fields)\n    )\n\n    # Do a mass concat of the first and last name fields for authorizing_official.\n    # The old operation was computationally heavy for some reason, so if we precompute\n    # this here, it is vastly more efficient.\n    domain_infos_cleaned = domain_infos.annotate(\n        ao=Concat(\n            Coalesce(F(\"authorizing_official__first_name\"), Value(\"\")),\n            Value(\" \"),\n            Coalesce(F(\"authorizing_official__last_name\"), Value(\"\")),\n            output_field=CharField(),\n        )\n    )\n    return domain_infos_cleaned\n\n\ndef parse_row(columns, domain_info: DomainInformation, security_emails_dict=None, get_domain_managers=False):\n    \"\"\"Given a set of columns, generate a new row from cleaned column data\"\"\"\n\n    # Domain should never be none when parsing this information\n    if domain_info.domain is None:\n        raise ValueError(\"Domain is none\")\n\n    domain = domain_info.domain  # type: ignore\n\n    # Grab the security email from a preset dictionary.\n    # If nothing exists in the dictionary, grab from .contacts.\n    if security_emails_dict is not None and domain.name in security_emails_dict:\n        _email = security_emails_dict.get(domain.name)\n        security_email = _email if _email is not None else \" \"\n    else:\n        # If the dictionary doesn't contain that data, lets filter for it manually.\n        # This is a last resort as this is a more expensive operation.\n        security_contacts = domain.contacts.filter(contact_type=PublicContact.ContactTypeChoices.SECURITY)\n        _email = security_contacts[0].email if security_contacts else None\n        security_email = _email if _email is not None else \" \"\n\n    # These are default emails that should not be displayed in the csv report\n    invalid_emails = {DefaultEmail.LEGACY_DEFAULT.value, DefaultEmail.PUBLIC_CONTACT_DEFAULT.value}\n    if security_email.lower() in invalid_emails:\n        security_email = \"(blank)\"\n\n    if domain_info.federal_type:\n        domain_type = f\"{domain_info.get_organization_type_display()} - {domain_info.get_federal_type_display()}\"\n    else:\n        domain_type = domain_info.get_organization_type_display()\n\n    # create a dictionary of fields which can be included in output\n    FIELDS = {\n        \"Domain name\": domain.name,\n        \"Status\": domain.get_state_display(),\n        \"Expiration date\": domain.expiration_date,\n        \"Domain type\": domain_type,\n        \"Agency\": domain_info.federal_agency,\n        \"Organization name\": domain_info.organization_name,\n        \"City\": domain_info.city,\n        \"State\": domain_info.state_territory,\n        \"AO\": domain_info.ao,  # type: ignore\n        \"AO email\": domain_info.authorizing_official.email if domain_info.authorizing_official else \" \",\n        \"Security contact email\": security_email,\n        \"Created at\": domain.created_at,\n        \"First ready\": domain.first_ready,\n        \"Deleted\": domain.deleted,\n    }\n\n    if get_domain_managers:\n        # Get each domain managers email and add to list\n        dm_emails = [dm.user.email for dm in domain.permissions.all()]\n\n        # Set up the \"matching header\" + row field data\n        for i, dm_email in enumerate(dm_emails, start=1):\n            FIELDS[f\"Domain manager email {i}\"] = dm_email\n\n    row = [FIELDS.get(column, \"\") for column in columns]\n    return row\n\n\ndef _get_security_emails(sec_contact_ids):\n    \"\"\"\n    Retrieve security contact emails for the given security contact IDs.\n    \"\"\"\n    security_emails_dict = {}\n    public_contacts = (\n        PublicContact.objects.only(\"email\", \"domain__name\")\n        .select_related(\"domain\")\n        .filter(registry_id__in=sec_contact_ids)\n    )\n\n    # Populate a dictionary of domain names and their security contacts\n    for contact in public_contacts:\n        domain: Domain = contact.domain\n        if domain is not None and domain.name not in security_emails_dict:\n            security_emails_dict[domain.name] = contact.email\n        else:\n            logger.warning(\"csv_export -> Domain was none for PublicContact\")\n\n    return security_emails_dict\n\n\ndef update_columns_with_domain_managers(columns, max_dm_count):\n    \"\"\"\n    Update the columns list to include \"Domain manager email {#}\" headers\n    based on the maximum domain manager count.\n    \"\"\"\n    for i in range(1, max_dm_count + 1):\n        columns.append(f\"Domain manager email {i}\")\n\n\ndef write_csv(\n    writer,\n    columns,\n    sort_fields,\n    filter_condition,\n    get_domain_managers=False,\n    should_write_header=True,\n):\n    \"\"\"\n    Receives params from the parent methods and outputs a CSV with fltered and sorted domains.\n    Works with write_header as longas the same writer object is passed.\n    get_domain_managers: Conditional bc we only use domain manager info for export_data_full_to_csv\n    should_write_header: Conditional bc export_data_growth_to_csv calls write_body twice\n    \"\"\"\n\n    all_domain_infos = get_domain_infos(filter_condition, sort_fields)\n\n    # Store all security emails to avoid epp calls or excessive filters\n    sec_contact_ids = all_domain_infos.values_list(\"domain__security_contact_registry_id\", flat=True)\n\n    security_emails_dict = _get_security_emails(sec_contact_ids)\n\n    # Reduce the memory overhead when performing the write operation\n    paginator = Paginator(all_domain_infos, 1000)\n\n    if get_domain_managers and len(all_domain_infos) > 0:\n        # We want to get the max amont of domain managers an\n        # account has to set the column header dynamically\n        max_dm_count = max(len(domain_info.domain.permissions.all()) for domain_info in all_domain_infos)\n        update_columns_with_domain_managers(columns, max_dm_count)\n\n    if should_write_header:\n        write_header(writer, columns)\n\n    for page_num in paginator.page_range:\n        rows = []\n        page = paginator.page(page_num)\n        for domain_info in page.object_list:\n            try:\n                row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)\n                rows.append(row)\n            except ValueError:\n                # This should not happen. If it does, just skip this row.\n                # It indicates that DomainInformation.domain is None.\n                logger.error(\"csv_export -> Error when parsing row, domain was None\")\n                continue\n\n        writer.writerows(rows)\n\n\ndef export_data_type_to_csv(csv_file):\n    \"\"\"All domains report with extra columns\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Status\",\n        \"Expiration date\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"AO\",\n        \"AO email\",\n        \"Security contact email\",\n        # For domain manager we are pass it in as a parameter below in write_body\n    ]\n\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=True, should_write_header=True)\n\n\ndef export_data_full_to_csv(csv_file):\n    \"\"\"All domains report\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Security contact email\",\n    ]\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n\n\ndef export_data_federal_to_csv(csv_file):\n    \"\"\"Federal domains report\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Security contact email\",\n    ]\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"organization_type__icontains\": \"federal\",\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n\n\ndef get_default_start_date():\n    # Default to a date that's prior to our first deployment\n    return timezone.make_aware(datetime(2023, 11, 1))\n\n\ndef get_default_end_date():\n    # Default to now()\n    return timezone.now()\n\n\ndef export_data_growth_to_csv(csv_file, start_date, end_date):\n    \"\"\"\n    Growth report:\n    Receive start and end dates from the view, parse them.\n    Request from write_body READY domains that are created between\n    the start and end dates, as well as DELETED domains that are deleted between\n    the start and end dates. Specify sort params for both lists.\n    \"\"\"\n\n    start_date_formatted = (\n        timezone.make_aware(datetime.strptime(start_date, \"%Y-%m-%d\")) if start_date else get_default_start_date()\n    )\n\n    end_date_formatted = (\n        timezone.make_aware(datetime.strptime(end_date, \"%Y-%m-%d\")) if end_date else get_default_end_date()\n    )\n\n    writer = csv.writer(csv_file)\n\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Status\",\n        \"Expiration date\",\n        \"Created at\",\n        \"First ready\",\n        \"Deleted\",\n    ]\n    sort_fields = [\n        \"domain__first_ready\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [Domain.State.READY],\n        \"domain__first_ready__lte\": end_date_formatted,\n        \"domain__first_ready__gte\": start_date_formatted,\n    }\n\n    # We also want domains deleted between sar and end dates, sorted\n    sort_fields_for_deleted_domains = [\n        \"domain__deleted\",\n        \"domain__name\",\n    ]\n    filter_condition_for_deleted_domains = {\n        \"domain__state__in\": [Domain.State.DELETED],\n        \"domain__deleted__lte\": end_date_formatted,\n        \"domain__deleted__gte\": start_date_formatted,\n    }\n\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n    write_csv(\n        writer,\n        columns,\n        sort_fields_for_deleted_domains,\n        filter_condition_for_deleted_domains,\n        get_domain_managers=False,\n        should_write_header=False,\n    )\n", "path": "src/registrar/utility/csv_export.py"}], "after_files": [{"content": "import csv\nimport logging\nfrom datetime import datetime\nfrom registrar.models.domain import Domain\nfrom registrar.models.domain_information import DomainInformation\nfrom django.utils import timezone\nfrom django.core.paginator import Paginator\nfrom django.db.models import F, Value, CharField\nfrom django.db.models.functions import Concat, Coalesce\n\nfrom registrar.models.public_contact import PublicContact\nfrom registrar.utility.enums import DefaultEmail\n\nlogger = logging.getLogger(__name__)\n\n\ndef write_header(writer, columns):\n    \"\"\"\n    Receives params from the parent methods and outputs a CSV with a header row.\n    Works with write_header as long as the same writer object is passed.\n    \"\"\"\n\n    writer.writerow(columns)\n\n\ndef get_domain_infos(filter_condition, sort_fields):\n    domain_infos = (\n        DomainInformation.objects.select_related(\"domain\", \"authorizing_official\")\n        .prefetch_related(\"domain__permissions\")\n        .filter(**filter_condition)\n        .order_by(*sort_fields)\n    )\n\n    # Do a mass concat of the first and last name fields for authorizing_official.\n    # The old operation was computationally heavy for some reason, so if we precompute\n    # this here, it is vastly more efficient.\n    domain_infos_cleaned = domain_infos.annotate(\n        ao=Concat(\n            Coalesce(F(\"authorizing_official__first_name\"), Value(\"\")),\n            Value(\" \"),\n            Coalesce(F(\"authorizing_official__last_name\"), Value(\"\")),\n            output_field=CharField(),\n        )\n    )\n    return domain_infos_cleaned\n\n\ndef parse_row(columns, domain_info: DomainInformation, security_emails_dict=None, get_domain_managers=False):\n    \"\"\"Given a set of columns, generate a new row from cleaned column data\"\"\"\n\n    # Domain should never be none when parsing this information\n    if domain_info.domain is None:\n        logger.error(\"Attemting to parse row for csv exports but Domain is none in a DomainInfo\")\n        raise ValueError(\"Domain is none\")\n\n    domain = domain_info.domain  # type: ignore\n\n    # Grab the security email from a preset dictionary.\n    # If nothing exists in the dictionary, grab from .contacts.\n    if security_emails_dict is not None and domain.name in security_emails_dict:\n        _email = security_emails_dict.get(domain.name)\n        security_email = _email if _email is not None else \" \"\n    else:\n        # If the dictionary doesn't contain that data, lets filter for it manually.\n        # This is a last resort as this is a more expensive operation.\n        security_contacts = domain.contacts.filter(contact_type=PublicContact.ContactTypeChoices.SECURITY)\n        _email = security_contacts[0].email if security_contacts else None\n        security_email = _email if _email is not None else \" \"\n\n    # These are default emails that should not be displayed in the csv report\n    invalid_emails = {DefaultEmail.LEGACY_DEFAULT.value, DefaultEmail.PUBLIC_CONTACT_DEFAULT.value}\n    if security_email.lower() in invalid_emails:\n        security_email = \"(blank)\"\n\n    if domain_info.federal_type:\n        domain_type = f\"{domain_info.get_organization_type_display()} - {domain_info.get_federal_type_display()}\"\n    else:\n        domain_type = domain_info.get_organization_type_display()\n\n    # create a dictionary of fields which can be included in output\n    FIELDS = {\n        \"Domain name\": domain.name,\n        \"Status\": domain.get_state_display(),\n        \"Expiration date\": domain.expiration_date,\n        \"Domain type\": domain_type,\n        \"Agency\": domain_info.federal_agency,\n        \"Organization name\": domain_info.organization_name,\n        \"City\": domain_info.city,\n        \"State\": domain_info.state_territory,\n        \"AO\": domain_info.ao,  # type: ignore\n        \"AO email\": domain_info.authorizing_official.email if domain_info.authorizing_official else \" \",\n        \"Security contact email\": security_email,\n        \"Created at\": domain.created_at,\n        \"First ready\": domain.first_ready,\n        \"Deleted\": domain.deleted,\n    }\n\n    if get_domain_managers:\n        # Get each domain managers email and add to list\n        dm_emails = [dm.user.email for dm in domain.permissions.all()]\n\n        # Set up the \"matching header\" + row field data\n        for i, dm_email in enumerate(dm_emails, start=1):\n            FIELDS[f\"Domain manager email {i}\"] = dm_email\n\n    row = [FIELDS.get(column, \"\") for column in columns]\n    return row\n\n\ndef _get_security_emails(sec_contact_ids):\n    \"\"\"\n    Retrieve security contact emails for the given security contact IDs.\n    \"\"\"\n    security_emails_dict = {}\n    public_contacts = (\n        PublicContact.objects.only(\"email\", \"domain__name\")\n        .select_related(\"domain\")\n        .filter(registry_id__in=sec_contact_ids)\n    )\n\n    # Populate a dictionary of domain names and their security contacts\n    for contact in public_contacts:\n        domain: Domain = contact.domain\n        if domain is not None and domain.name not in security_emails_dict:\n            security_emails_dict[domain.name] = contact.email\n        else:\n            logger.warning(\"csv_export -> Domain was none for PublicContact\")\n\n    return security_emails_dict\n\n\ndef write_csv(\n    writer,\n    columns,\n    sort_fields,\n    filter_condition,\n    get_domain_managers=False,\n    should_write_header=True,\n):\n    \"\"\"\n    Receives params from the parent methods and outputs a CSV with fltered and sorted domains.\n    Works with write_header as longas the same writer object is passed.\n    get_domain_managers: Conditional bc we only use domain manager info for export_data_full_to_csv\n    should_write_header: Conditional bc export_data_growth_to_csv calls write_body twice\n    \"\"\"\n\n    all_domain_infos = get_domain_infos(filter_condition, sort_fields)\n\n    # Store all security emails to avoid epp calls or excessive filters\n    sec_contact_ids = all_domain_infos.values_list(\"domain__security_contact_registry_id\", flat=True)\n\n    security_emails_dict = _get_security_emails(sec_contact_ids)\n\n    # Reduce the memory overhead when performing the write operation\n    paginator = Paginator(all_domain_infos, 1000)\n\n    # The maximum amount of domain managers an account has\n    # We get the max so we can set the column header accurately\n    max_dm_count = 0\n    total_body_rows = []\n\n    for page_num in paginator.page_range:\n        rows = []\n        page = paginator.page(page_num)\n        for domain_info in page.object_list:\n\n            # Get count of all the domain managers for an account\n            if get_domain_managers:\n                dm_count = domain_info.domain.permissions.count()\n                if dm_count > max_dm_count:\n                    max_dm_count = dm_count\n                    for i in range(1, max_dm_count + 1):\n                        column_name = f\"Domain manager email {i}\"\n                        if column_name not in columns:\n                            columns.append(column_name)\n\n            try:\n                row = parse_row(columns, domain_info, security_emails_dict, get_domain_managers)\n                rows.append(row)\n            except ValueError:\n                # This should not happen. If it does, just skip this row.\n                # It indicates that DomainInformation.domain is None.\n                logger.error(\"csv_export -> Error when parsing row, domain was None\")\n                continue\n        total_body_rows.extend(rows)\n\n    if should_write_header:\n        write_header(writer, columns)\n    writer.writerows(total_body_rows)\n\n\ndef export_data_type_to_csv(csv_file):\n    \"\"\"All domains report with extra columns\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Status\",\n        \"Expiration date\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"AO\",\n        \"AO email\",\n        \"Security contact email\",\n        # For domain manager we are pass it in as a parameter below in write_body\n    ]\n\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=True, should_write_header=True)\n\n\ndef export_data_full_to_csv(csv_file):\n    \"\"\"All domains report\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Security contact email\",\n    ]\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n\n\ndef export_data_federal_to_csv(csv_file):\n    \"\"\"Federal domains report\"\"\"\n\n    writer = csv.writer(csv_file)\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Security contact email\",\n    ]\n    # Coalesce is used to replace federal_type of None with ZZZZZ\n    sort_fields = [\n        \"organization_type\",\n        Coalesce(\"federal_type\", Value(\"ZZZZZ\")),\n        \"federal_agency\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"organization_type__icontains\": \"federal\",\n        \"domain__state__in\": [\n            Domain.State.READY,\n            Domain.State.DNS_NEEDED,\n            Domain.State.ON_HOLD,\n        ],\n    }\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n\n\ndef get_default_start_date():\n    # Default to a date that's prior to our first deployment\n    return timezone.make_aware(datetime(2023, 11, 1))\n\n\ndef get_default_end_date():\n    # Default to now()\n    return timezone.now()\n\n\ndef export_data_growth_to_csv(csv_file, start_date, end_date):\n    \"\"\"\n    Growth report:\n    Receive start and end dates from the view, parse them.\n    Request from write_body READY domains that are created between\n    the start and end dates, as well as DELETED domains that are deleted between\n    the start and end dates. Specify sort params for both lists.\n    \"\"\"\n\n    start_date_formatted = (\n        timezone.make_aware(datetime.strptime(start_date, \"%Y-%m-%d\")) if start_date else get_default_start_date()\n    )\n\n    end_date_formatted = (\n        timezone.make_aware(datetime.strptime(end_date, \"%Y-%m-%d\")) if end_date else get_default_end_date()\n    )\n\n    writer = csv.writer(csv_file)\n\n    # define columns to include in export\n    columns = [\n        \"Domain name\",\n        \"Domain type\",\n        \"Agency\",\n        \"Organization name\",\n        \"City\",\n        \"State\",\n        \"Status\",\n        \"Expiration date\",\n        \"Created at\",\n        \"First ready\",\n        \"Deleted\",\n    ]\n    sort_fields = [\n        \"domain__first_ready\",\n        \"domain__name\",\n    ]\n    filter_condition = {\n        \"domain__state__in\": [Domain.State.READY],\n        \"domain__first_ready__lte\": end_date_formatted,\n        \"domain__first_ready__gte\": start_date_formatted,\n    }\n\n    # We also want domains deleted between sar and end dates, sorted\n    sort_fields_for_deleted_domains = [\n        \"domain__deleted\",\n        \"domain__name\",\n    ]\n    filter_condition_for_deleted_domains = {\n        \"domain__state__in\": [Domain.State.DELETED],\n        \"domain__deleted__lte\": end_date_formatted,\n        \"domain__deleted__gte\": start_date_formatted,\n    }\n\n    write_csv(writer, columns, sort_fields, filter_condition, get_domain_managers=False, should_write_header=True)\n    write_csv(\n        writer,\n        columns,\n        sort_fields_for_deleted_domains,\n        filter_condition_for_deleted_domains,\n        get_domain_managers=False,\n        should_write_header=False,\n    )\n", "path": "src/registrar/utility/csv_export.py"}]}