Skip to content

[DEV-14781] Fail Downloads during Pipeline #4735

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
DavidMikolaKC wants to merge 2 commits into
base: qat
Choose a base branch
from
+32 −24
Open

[DEV-14781] Fail Downloads during Pipeline #4735

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3256626
[DEV-14781] Added initial check for pipeline error
DavidMikolaKC Aug 13, 2026
ab7e192
[DEV-14781] Ruff Format
DavidMikolaKC Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 32 additions & 24 deletions usaspending_api/download/filestreaming/download_generation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
EXCEL_ROW_LIMIT = 1_000_000
WAIT_FOR_PROCESS_SLEEP = 5
JOB_TYPE = "USAspendingDownloader"
PIPELINE_ERROR_STR = "Canceling statement due to conflict with recovery"

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -905,41 +906,43 @@ def execute_psql(temp_sql_file_path: str, source_path: str, download_job: Downlo
)

with subprocess_trace as span:
span.set_attributes({
"service": "bulk-download",
"resource": str(download_sql),
"span_type": "Internal",
"source_path": str(source_path),
"download_job_id": str(download_job.download_job_id),
"download_job_status": str(download_job.job_status.name),
"download_file_name": str(download_job.file_name),
"download_file_size": download_job.file_size if download_job.file_size is not None else 0,
"number_of_rows": download_job.number_of_rows if download_job.number_of_rows is not None else 0,
"number_of_columns": download_job.number_of_columns if download_job.number_of_columns is not None else 0,
"error_message": download_job.error_message if download_job.error_message else "",
"monthly_download": str(download_job.monthly_download),
"json_request": str(download_job.json_request) if download_job.json_request else "",
})
span.set_attributes(
{
"service": "bulk-download",
"resource": str(download_sql),
"span_type": "Internal",
"source_path": str(source_path),
"download_job_id": str(download_job.download_job_id),
"download_job_status": str(download_job.job_status.name),
"download_file_name": str(download_job.file_name),
"download_file_size": download_job.file_size if download_job.file_size is not None else 0,
"number_of_rows": download_job.number_of_rows if download_job.number_of_rows is not None else 0,
"number_of_columns": download_job.number_of_columns
if download_job.number_of_columns is not None
else 0,
"error_message": download_job.error_message if download_job.error_message else "",
"monthly_download": str(download_job.monthly_download),
"json_request": str(download_job.json_request) if download_job.json_request else "",
}
)

try:
log_time = time.perf_counter()

# Build PostgreSQL environment using helper
psql_env = build_psql_env(
dsn=retrieve_db_string(),
statement_timeout_hours=settings.DOWNLOAD_DB_TIMEOUT_IN_HOURS if (
download_job and not download_job.monthly_download) else None,
work_mem_mb=settings.DOWNLOAD_DB_WORK_MEM_IN_MB if (
download_job and not download_job.monthly_download) else None
statement_timeout_hours=settings.DOWNLOAD_DB_TIMEOUT_IN_HOURS
if (download_job and not download_job.monthly_download)
else None,
work_mem_mb=settings.DOWNLOAD_DB_WORK_MEM_IN_MB
if (download_job and not download_job.monthly_download)
else None,
)

# Execute psql using helper
run_psql_to_file(
sql_path=temp_sql_file_path,
output_path=source_path,
env=psql_env,
quiet=True,
on_error_stop=True
sql_path=temp_sql_file_path, output_path=source_path, env=psql_env, quiet=True, on_error_stop=True
)

duration = time.perf_counter() - log_time
Expand All @@ -948,6 +951,11 @@ def execute_psql(temp_sql_file_path: str, source_path: str, download_job: Downlo
download_job=download_job,
)
except subprocess.CalledProcessError as e:
message = e.output.decode()
if PIPELINE_ERROR_STR in message:
span.set_attribute("raised_exception", message)
if download_job:
fail_download(download_job, e, message)
write_to_log(message=f"PSQL Error: {e.output.decode()}", is_error=True, download_job=download_job)
raise e
except Exception as e:
Expand Down
Loading