[DEV-14824] Handle merge conflict

Author: sethstoudenmier

pyproject.toml

@@ -50,7 +50,7 @@ dependencies = [
     "openpyxl==3.1.*",
     "pandas==2.2.*",
     "psutil==5.9.*",
-    "psycopg2==2.9.9", # Pinning exact version because this package will drop support for Python versions in patches
+    "psycopg>=3.3.3",
     "py-gfm==2.0.0",
     "pydantic[dotenv]==1.9.*",
     "python-json-logger==2.0.7",

usaspending_api/awards/v2/views/accounts.py

@@ -1,15 +1,19 @@
 from collections import OrderedDict
 
-from psycopg2.sql import Identifier, Literal, SQL
+from psycopg.sql import SQL, Identifier, Literal
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from usaspending_api.common.cache_decorator import cache_response
 from usaspending_api.common.helpers.generic_helper import get_pagination
 from usaspending_api.common.helpers.sql_helpers import execute_sql_to_ordered_dictionary
-from usaspending_api.common.validator.award import get_internal_or_generated_award_id_model
-from usaspending_api.common.validator.pagination import customize_pagination_with_sort_columns
+from usaspending_api.common.validator.award import (
+    get_internal_or_generated_award_id_model,
+)
+from usaspending_api.common.validator.pagination import (
+    customize_pagination_with_sort_columns,
+)
 from usaspending_api.common.validator.tinyshield import validate_post_request
 from usaspending_api.references.helpers import generate_agency_slugs_for_agency_list
 

usaspending_api/awards/v2/views/funding.py

@@ -2,16 +2,23 @@
 from copy import deepcopy
 from itertools import chain
 
-from psycopg2.sql import Identifier, Literal, SQL
+from psycopg.sql import SQL, Identifier, Literal
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from usaspending_api.common.cache_decorator import cache_response
 from usaspending_api.common.helpers.generic_helper import get_simple_pagination_metadata
-from usaspending_api.common.helpers.sql_helpers import build_composable_order_by, execute_sql_to_ordered_dictionary
-from usaspending_api.common.validator.award import get_internal_or_generated_award_id_model
-from usaspending_api.common.validator.pagination import customize_pagination_with_sort_columns
+from usaspending_api.common.helpers.sql_helpers import (
+    build_composable_order_by,
+    execute_sql_to_ordered_dictionary,
+)
+from usaspending_api.common.validator.award import (
+    get_internal_or_generated_award_id_model,
+)
+from usaspending_api.common.validator.pagination import (
+    customize_pagination_with_sort_columns,
+)
 from usaspending_api.common.validator.tinyshield import validate_post_request
 from usaspending_api.references.helpers import generate_agency_slugs_for_agency_list
 

usaspending_api/awards/v2/views/funding_rollup.py

@@ -1,16 +1,17 @@
 from collections import OrderedDict
 
-from psycopg2.sql import Identifier, Literal, SQL
+from psycopg.sql import SQL, Identifier, Literal
 from rest_framework.request import Request
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from usaspending_api.common.cache_decorator import cache_response
 from usaspending_api.common.helpers.sql_helpers import execute_sql_to_ordered_dictionary
-from usaspending_api.common.validator.award import get_internal_or_generated_award_id_model
+from usaspending_api.common.validator.award import (
+    get_internal_or_generated_award_id_model,
+)
 from usaspending_api.common.validator.tinyshield import validate_post_request
 
-
 ROLLUP_SQL = SQL(
     """
     with gather_financial_accounts_by_awards as (
@@ -21,7 +22,8 @@
         from    vw_awards a
                 inner join financial_accounts_by_awards faba on faba.award_id = a.id
                 INNER JOIN submission_attributes sa ON faba.submission_id = sa.submission_id
-                INNER JOIN dabs_submission_window_schedule dabs ON sa.submission_window_id = dabs.id and dabs.submission_reveal_date <= now()
+                INNER JOIN dabs_submission_window_schedule dabs
+                ON sa.submission_window_id = dabs.id and dabs.submission_reveal_date <= now()
         where   {award_id_column} = {award_id}
     )
     select

usaspending_api/broker/helpers/delete_fabs_transactions.py

@@ -2,15 +2,15 @@
 
 from django.conf import settings
 from django.db import connections
+from psycopg.sql import SQL, Placeholder
 
 from usaspending_api.broker.helpers.delete_stale_fabs import delete_stale_fabs
 from usaspending_api.common.helpers.timing_helpers import timer
 
-
 logger = logging.getLogger("script")
 
 
-def delete_fabs_transactions(ids_to_delete):
+def delete_fabs_transactions(ids_to_delete: list) -> list:
     """ids_to_delete are published_fabs_ids"""
     if ids_to_delete:
         with timer(f"deleting {len(ids_to_delete)} stale FABS data", logger.info):
@@ -23,7 +23,7 @@ def delete_fabs_transactions(ids_to_delete):
     return update_and_delete_award_ids
 
 
-def get_delete_pks_for_afa_keys(afa_ids_to_delete):
+def get_delete_pks_for_afa_keys(afa_ids_to_delete: list) -> list:
     """
     When we read from FABS delete files, we are only reading in afa_generated_unique keys (AFA).  Unfortunately,
     AFAs on their own do not give us enough information to delete records since AFAs are reused in Broker.  This
@@ -35,16 +35,18 @@ def get_delete_pks_for_afa_keys(afa_ids_to_delete):
         return []
 
     uppercased = tuple(afa.upper() for afa in afa_ids_to_delete)
-
-    sql = """
+    placeholders = SQL(",").join(Placeholder() * len(uppercased))
+    sql = SQL(
+        """
     select  published_fabs_id
     from    published_fabs
-    where   upper(afa_generated_unique) in %s and
+    where   upper(afa_generated_unique) in ({}) and
             is_active is not true
     """
+    ).format(placeholders)
 
     with connections[settings.BROKER_DB_ALIAS].cursor() as cursor:
-        cursor.execute(sql, [uppercased])
+        cursor.execute(sql, uppercased)
         rows = cursor.fetchall()
 
     return [row[0] for row in rows]

usaspending_api/broker/management/commands/load_broker_table.py

@@ -1,10 +1,11 @@
+import argparse
 import csv
 import logging
+from io import StringIO
 
 from django.conf import settings
 from django.core.management import BaseCommand
 from django.db import connections
-from io import StringIO
 
 logger = logging.getLogger(__name__)
 
@@ -15,7 +16,7 @@ class Command(BaseCommand):
 
     CHUNK_SIZE = 50_000
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: argparse.ArgumentParser) -> None:
 
         broker_group = parser.add_argument_group(
             title="broker",
@@ -37,7 +38,8 @@ def add_arguments(self, parser):
 
         usaspending_group = parser.add_argument_group(
             title="usaspending",
-            description="Optional fields that reference components of the USAspending DB.  Defaults to Broker DB counterpart",
+            description="Optional fields that reference components of the USAspending DB. "
+            "Defaults to Broker DB counterpart",
         )
         usaspending_group.add_argument(
             "--usaspending-table-name",
@@ -52,7 +54,7 @@ def add_arguments(self, parser):
             help="Schema name in the USAspending DB to load data into",
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args, **options) -> None:
 
         broker_table_name = options["table_name"]
         broker_schema_name = options["schema_name"]
@@ -101,8 +103,10 @@ def handle(self, *args, **options):
                     writer = csv.writer(f)
                     writer.writerows(rows)
                     f.seek(0)
-                    usas_cursor.copy_expert(
-                        f"COPY {usas_schema_name}.{usas_table_name} ({usas_table_columns}) FROM STDIN WITH (DELIMITER ',', FORMAT CSV)",
-                        f,
-                    )
+                    with usas_cursor.copy(
+                        f"COPY {usas_schema_name}.{usas_table_name} ({usas_table_columns}) FROM STDIN "
+                        "WITH (DELIMITER ',', FORMAT CSV)",
+                    ) as copy:
+                        for row in f:
+                            copy.write(row)
                     usas_conn.commit()

usaspending_api/broker/management/commands/load_fpds_transactions.py

@@ -1,19 +1,35 @@
 import logging
-import psycopg2
 import re
-
 from datetime import datetime, timezone
+from typing import IO, AnyStr, List, Optional
+
+import psycopg
+from django.core.management import CommandParser
 from django.core.management.base import BaseCommand
-from typing import IO, List, AnyStr, Optional
 
-from usaspending_api.broker.helpers.last_load_date import get_last_load_date, update_last_load_date
-from usaspending_api.common.helpers.date_helper import datetime_command_line_argument_type
+from usaspending_api.broker.helpers.last_load_date import (
+    get_last_load_date,
+    update_last_load_date,
+)
+from usaspending_api.common.helpers.date_helper import (
+    datetime_command_line_argument_type,
+)
 from usaspending_api.common.helpers.etl_helpers import update_c_to_d_linkages
 from usaspending_api.common.helpers.sql_helpers import get_database_dsn_string
 from usaspending_api.common.retrieve_file_from_uri import RetrieveFileFromUri
-from usaspending_api.etl.award_helpers import update_awards, update_procurement_awards, prune_empty_awards
-from usaspending_api.etl.transaction_loaders.fpds_loader import load_fpds_transactions, failed_ids, delete_stale_fpds
-from usaspending_api.transactions.transaction_delete_journal_helpers import retrieve_deleted_fpds_transactions
+from usaspending_api.etl.award_helpers import (
+    prune_empty_awards,
+    update_awards,
+    update_procurement_awards,
+)
+from usaspending_api.etl.transaction_loaders.fpds_loader import (
+    delete_stale_fpds,
+    failed_ids,
+    load_fpds_transactions,
+)
+from usaspending_api.transactions.transaction_delete_journal_helpers import (
+    retrieve_deleted_fpds_transactions,
+)
 
 logger = logging.getLogger("script")
 
@@ -28,12 +44,13 @@ class Command(BaseCommand):
     modified_award_ids = []
 
     @staticmethod
-    def get_cursor_for_date_query(connection, date, count=False):
+    def get_cursor_for_date_query(
+        connection: psycopg.Connection, date: datetime, count: bool = False
+    ) -> psycopg.Cursor:
+        db_cursor = connection.cursor()
         if count:
-            db_cursor = connection.cursor()
             db_query = ALL_FPDS_QUERY.format("COUNT(*)")
         else:
-            db_cursor = connection.cursor("fpds_load", cursor_factory=psycopg2.extras.DictCursor)
             db_query = ALL_FPDS_QUERY.format("detached_award_procurement_id")
 
         if date:
@@ -54,7 +71,7 @@ def load_fpds_incrementally(self, date: Optional[datetime], chunk_size: int = CH
             stale_awards = delete_stale_fpds(detached_award_procurement_ids)
             self.modified_award_ids.extend(stale_awards)
 
-        with psycopg2.connect(dsn=get_database_dsn_string()) as connection:
+        with psycopg.connect(get_database_dsn_string()) as connection:
             logger.info("Fetching records to update")
             total_records = self.get_cursor_for_date_query(connection, date, True).fetchall()[0][0]
             records_processed = 0
@@ -93,7 +110,7 @@ def load_fpds_from_file(self, file_path: str) -> None:
         logger.info(f"Total transaction IDs in file: {total_count}")
 
     @staticmethod
-    def update_award_records(awards, skip_cd_linkage=True):
+    def update_award_records(awards: list, skip_cd_linkage: bool = True) -> None:
         if awards:
             unique_awards = set(awards)
             logger.info(f"{len(unique_awards)} award records impacted by transaction DML operations")
@@ -107,7 +124,7 @@ def update_award_records(awards, skip_cd_linkage=True):
         else:
             logger.info("No award records to update")
 
-    def add_arguments(self, parser):
+    def add_arguments(self, parser: CommandParser) -> None:
         mutually_exclusive_group = parser.add_mutually_exclusive_group(required=True)
 
         mutually_exclusive_group.add_argument(
@@ -131,16 +148,17 @@ def add_arguments(self, parser):
             "--file",
             metavar="FILEPATH",
             type=str,
-            help="Load/Reload transactions using the detached_award_procurement_id list stored at this file path (one ID per line)"
-            "to reload, one ID per line. Nonexistent IDs will be ignored.",
+            help="Load/Reload transactions using the detached_award_procurement_id list stored at this file path "
+            "(one ID per line) to reload, one ID per line. Nonexistent IDs will be ignored.",
         )
         mutually_exclusive_group.add_argument(
             "--reload-all",
             action="store_true",
-            help="Script will load or reload all FPDS records in source tables, from all time. This does NOT clear the USAspending database first",
+            help="Script will load or reload all FPDS records in source tables, from all time. "
+            "This does NOT clear the USAspending database first",
         )
 
-    def handle(self, *args, **options):
+    def handle(self, *args, **options) -> None:
 
         # Record script execution start time to update the FPDS last updated date in DB as appropriate
         update_time = datetime.now(timezone.utc)
@@ -173,7 +191,8 @@ def handle(self, *args, **options):
             raise SystemExit(1)
 
         if options["reload_all"] or options["since_last_load"]:
-            # we wait until after the load finishes to update the load date because if this crashes we'll need to load again
+            # we wait until after the load finishes to update the load date
+            # because if this crashes we'll need to load again
             update_last_load_date("fpds", update_time)
 
-        logger.info(f"Successfully Completed")
+        logger.info("Successfully Completed")