mirror of
https://github.com/Sea-Haven-Industries/proposal-system.git
synced 2026-09-30 22:53:13 +00:00
669e9c0e43
LAM-M2: Add sanitize_user_text() to suggestions Lambda that strips common prompt injection patterns (blocklist + delimiter neutralisation) before including user-supplied text in Bedrock prompts. LAM-M3: Add file size check in pdf-extract before downloading — rejects PDFs over 50 MB with a logged warning and ValueError. LAM-M6: Add validate_line_item_numerics() to suggestions Lambda that rejects Bedrock-generated line items with negative values, NaN/Inf, or amounts exceeding $10M ceiling. LAM-M9: Replace indefinite API key cache with 5-minute TTL in all four Lambdas (suggestions, pdf-extract, pdf-generate, library-ingest) so rotated Secrets Manager values take effect promptly.
628 lines
18 KiB
Python
628 lines
18 KiB
Python
"""Proposal System - PDF Generate Lambda.
|
|
|
|
Generates professional branded proposal PDFs using reportlab Platypus.
|
|
Triggered via SQS when an admin requests PDF generation.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import time
|
|
from datetime import datetime
|
|
from io import BytesIO
|
|
|
|
import boto3
|
|
import httpx
|
|
from reportlab.lib import colors
|
|
from reportlab.lib.enums import TA_CENTER, TA_RIGHT
|
|
from reportlab.lib.pagesizes import letter
|
|
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
|
|
from reportlab.lib.units import inch
|
|
from reportlab.platypus import (
|
|
Paragraph,
|
|
SimpleDocTemplate,
|
|
Spacer,
|
|
Table,
|
|
TableStyle,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))
|
|
|
|
GENERATED_BUCKET = os.environ.get("GENERATED_BUCKET", "")
|
|
API_BASE_URL = os.environ.get("API_BASE_URL", "")
|
|
INTERNAL_API_KEY_SECRET_ARN = os.environ.get("INTERNAL_API_KEY_SECRET_ARN", "")
|
|
|
|
s3 = boto3.client("s3")
|
|
secrets_client = boto3.client("secretsmanager")
|
|
|
|
_cached_api_key: str | None = None
|
|
_cached_api_key_ts: float = 0.0
|
|
_API_KEY_TTL_SECONDS = 300 # Fix: LAM-M9 — re-fetch every 5 minutes
|
|
|
|
# Fix: LAM-M8 — pattern for allowed S3 key characters
|
|
_SAFE_S3_KEY_RE = re.compile(r"^[a-zA-Z0-9\-_./\s]+$")
|
|
|
|
COMPANY_NAME = "Sea Haven Industries"
|
|
COMPANY_ADDRESS = "Sea Haven Industries LLC"
|
|
COMPANY_PHONE = ""
|
|
COMPANY_EMAIL = "info@seahavenind.com"
|
|
|
|
TERMS_AND_CONDITIONS = """
|
|
1. This proposal is valid for 30 days from the date of issue.
|
|
2. Payment terms: Net 30 days from invoice date.
|
|
3. Any changes to the scope of work may result in additional charges.
|
|
4. Work will be scheduled upon acceptance of this proposal.
|
|
5. All materials and workmanship are guaranteed for one (1) year from completion.
|
|
6. Client is responsible for providing access to the work area.
|
|
7. This proposal does not include permits unless specifically noted in the line items.
|
|
""".strip()
|
|
|
|
|
|
def _get_api_key() -> str:
|
|
"""Fetch internal API key from Secrets Manager with a 5-minute TTL cache.
|
|
|
|
Fix: LAM-M9 — the key was previously cached indefinitely. Now the cached
|
|
value expires after _API_KEY_TTL_SECONDS so rotated secrets take effect.
|
|
"""
|
|
global _cached_api_key, _cached_api_key_ts
|
|
now = time.monotonic()
|
|
if _cached_api_key is not None and (now - _cached_api_key_ts) < _API_KEY_TTL_SECONDS:
|
|
return _cached_api_key
|
|
if INTERNAL_API_KEY_SECRET_ARN:
|
|
resp = secrets_client.get_secret_value(SecretId=INTERNAL_API_KEY_SECRET_ARN)
|
|
_cached_api_key = resp["SecretString"]
|
|
else:
|
|
_cached_api_key = ""
|
|
_cached_api_key_ts = now
|
|
return _cached_api_key
|
|
|
|
|
|
def _validate_s3_key(key: str) -> str:
|
|
"""Fix: LAM-M8 — sanitize and validate S3 keys before use."""
|
|
sanitized = key.replace("../", "").replace("..\\", "")
|
|
while "//" in sanitized:
|
|
sanitized = sanitized.replace("//", "/")
|
|
sanitized = sanitized.strip("/").strip()
|
|
|
|
if not sanitized:
|
|
raise ValueError("S3 key is empty after sanitization")
|
|
if not _SAFE_S3_KEY_RE.match(sanitized):
|
|
raise ValueError(f"S3 key contains disallowed characters: {sanitized!r}")
|
|
return sanitized
|
|
|
|
|
|
def handler(event, context):
|
|
batch_item_failures = []
|
|
|
|
# Fix: LAM-M1 — validate event structure before processing
|
|
records = event.get("Records")
|
|
if not isinstance(records, list) or not records:
|
|
logger.warning("Event has no Records or Records is not a list, returning early")
|
|
return {"batchItemFailures": []}
|
|
|
|
for record in records:
|
|
if "body" not in record:
|
|
logger.warning(
|
|
"Record missing 'body' key, skipping: %s",
|
|
record.get("messageId", "unknown"),
|
|
)
|
|
continue
|
|
|
|
try:
|
|
body = json.loads(record["body"])
|
|
except (json.JSONDecodeError, TypeError) as e:
|
|
# Fix: LAM-M1 — malformed JSON cannot be retried, add to failures
|
|
logger.error("Malformed JSON in record %s: %s", record.get("messageId"), e)
|
|
batch_item_failures.append({"itemIdentifier": record["messageId"]})
|
|
continue
|
|
|
|
try:
|
|
payload = body.get("payload", body)
|
|
proposal_id = payload["proposalId"]
|
|
generate_pdf(proposal_id)
|
|
except Exception as e:
|
|
# Fix: LAM-M5 — include stack trace in error logging
|
|
logger.exception(
|
|
"Failed to process record %s: %s", record.get("messageId"), e
|
|
)
|
|
batch_item_failures.append({"itemIdentifier": record["messageId"]})
|
|
return {"batchItemFailures": batch_item_failures}
|
|
|
|
|
|
def generate_pdf(proposal_id: str):
|
|
proposal = fetch_proposal(proposal_id)
|
|
if not proposal:
|
|
logger.warning("Proposal %s not found", proposal_id)
|
|
return
|
|
|
|
line_items = fetch_line_items(proposal_id)
|
|
|
|
pdf_bytes = build_pdf(proposal, line_items)
|
|
|
|
proposal_number = proposal["proposalNumber"]
|
|
revision = proposal.get("currentRevision", 1)
|
|
s3_key = f"{proposal_number}/rev-{revision}.pdf"
|
|
|
|
# Fix: LAM-M8 — validate constructed S3 key
|
|
s3_key = _validate_s3_key(s3_key)
|
|
|
|
upload_pdf(s3_key, pdf_bytes)
|
|
|
|
register_pdf(proposal_id, s3_key)
|
|
|
|
logger.info("Generated PDF: %s (%d bytes)", s3_key, len(pdf_bytes))
|
|
|
|
|
|
def fetch_proposal(proposal_id: str) -> dict | None:
|
|
try:
|
|
resp = _retry_request(
|
|
"GET",
|
|
f"{API_BASE_URL}/api/proposals/{proposal_id}",
|
|
headers=_api_headers(),
|
|
)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception as e:
|
|
# Fix: LAM-M5 — include stack trace in error logging
|
|
logger.exception("Error fetching proposal: %s", e)
|
|
return None
|
|
|
|
|
|
def fetch_line_items(proposal_id: str) -> list[dict]:
|
|
try:
|
|
resp = _retry_request(
|
|
"GET",
|
|
f"{API_BASE_URL}/api/proposals/{proposal_id}/line-items",
|
|
headers=_api_headers(),
|
|
)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception as e:
|
|
# Fix: LAM-M5 — include stack trace in error logging
|
|
logger.exception("Error fetching line items: %s", e)
|
|
return []
|
|
|
|
|
|
def build_pdf(proposal: dict, line_items: list[dict]) -> bytes:
|
|
buffer = BytesIO()
|
|
|
|
doc = SimpleDocTemplate(
|
|
buffer,
|
|
pagesize=letter,
|
|
leftMargin=0.75 * inch,
|
|
rightMargin=0.75 * inch,
|
|
topMargin=0.75 * inch,
|
|
bottomMargin=0.75 * inch,
|
|
)
|
|
|
|
styles = _get_styles()
|
|
elements = []
|
|
|
|
# Header
|
|
elements.extend(_build_header(proposal, styles))
|
|
elements.append(Spacer(1, 0.3 * inch))
|
|
|
|
# Proposal metadata
|
|
elements.extend(_build_metadata(proposal, styles))
|
|
elements.append(Spacer(1, 0.3 * inch))
|
|
|
|
# Scope of work
|
|
elements.extend(_build_scope(proposal, styles))
|
|
elements.append(Spacer(1, 0.3 * inch))
|
|
|
|
# Line items table
|
|
elements.extend(_build_line_items_table(line_items, styles))
|
|
elements.append(Spacer(1, 0.4 * inch))
|
|
|
|
# Terms and conditions
|
|
elements.extend(_build_terms(styles))
|
|
|
|
doc.build(elements, onFirstPage=_page_footer, onLaterPages=_page_footer)
|
|
|
|
return buffer.getvalue()
|
|
|
|
|
|
def _get_styles():
|
|
styles = getSampleStyleSheet()
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"CompanyName",
|
|
parent=styles["Heading1"],
|
|
fontSize=18,
|
|
leading=22,
|
|
textColor=colors.HexColor("#1a237e"),
|
|
spaceAfter=2,
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"CompanyInfo",
|
|
parent=styles["Normal"],
|
|
fontSize=9,
|
|
leading=12,
|
|
textColor=colors.HexColor("#555555"),
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"ProposalTitle",
|
|
parent=styles["Heading2"],
|
|
fontSize=14,
|
|
leading=18,
|
|
textColor=colors.HexColor("#1a237e"),
|
|
spaceBefore=6,
|
|
spaceAfter=12,
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"SectionHeader",
|
|
parent=styles["Heading3"],
|
|
fontSize=11,
|
|
leading=14,
|
|
textColor=colors.HexColor("#1a237e"),
|
|
spaceBefore=8,
|
|
spaceAfter=6,
|
|
borderWidth=0,
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"MetaLabel",
|
|
parent=styles["Normal"],
|
|
fontSize=9,
|
|
leading=12,
|
|
textColor=colors.HexColor("#666666"),
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"MetaValue",
|
|
parent=styles["Normal"],
|
|
fontSize=10,
|
|
leading=13,
|
|
fontName="Helvetica-Bold",
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"ScopeText",
|
|
parent=styles["Normal"],
|
|
fontSize=10,
|
|
leading=14,
|
|
spaceBefore=4,
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"TermsText",
|
|
parent=styles["Normal"],
|
|
fontSize=8,
|
|
leading=11,
|
|
textColor=colors.HexColor("#555555"),
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"TotalLabel",
|
|
parent=styles["Normal"],
|
|
fontSize=11,
|
|
leading=14,
|
|
fontName="Helvetica-Bold",
|
|
alignment=TA_RIGHT,
|
|
)
|
|
)
|
|
|
|
styles.add(
|
|
ParagraphStyle(
|
|
"FooterText",
|
|
parent=styles["Normal"],
|
|
fontSize=8,
|
|
leading=10,
|
|
textColor=colors.HexColor("#888888"),
|
|
alignment=TA_CENTER,
|
|
)
|
|
)
|
|
|
|
return styles
|
|
|
|
|
|
def _build_header(proposal: dict, styles) -> list:
|
|
revision = proposal.get("currentRevision", 1)
|
|
revision_text = f" | Rev {revision}" if revision > 1 else ""
|
|
|
|
header_data = [
|
|
[
|
|
Paragraph(COMPANY_NAME, styles["CompanyName"]),
|
|
Paragraph(f"PROPOSAL{revision_text}", styles["ProposalTitle"]),
|
|
],
|
|
[
|
|
Paragraph(f"{COMPANY_EMAIL}", styles["CompanyInfo"]),
|
|
Paragraph(f"#{proposal['proposalNumber']}", styles["MetaValue"]),
|
|
],
|
|
]
|
|
|
|
header_table = Table(header_data, colWidths=[3.5 * inch, 3.5 * inch])
|
|
header_table.setStyle(
|
|
TableStyle(
|
|
[
|
|
("VALIGN", (0, 0), (-1, -1), "TOP"),
|
|
("ALIGN", (1, 0), (1, -1), "RIGHT"),
|
|
("LINEBELOW", (0, -1), (-1, -1), 1.5, colors.HexColor("#1a237e")),
|
|
("BOTTOMPADDING", (0, -1), (-1, -1), 8),
|
|
]
|
|
)
|
|
)
|
|
|
|
return [header_table]
|
|
|
|
|
|
def _build_metadata(proposal: dict, styles) -> list:
|
|
submitted_at = proposal.get("submittedAt", "")
|
|
if submitted_at:
|
|
try:
|
|
dt = datetime.fromisoformat(submitted_at.replace("Z", "+00:00"))
|
|
submitted_at = dt.strftime("%B %d, %Y")
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
approved_at = proposal.get("approvedAt", "")
|
|
if approved_at:
|
|
try:
|
|
dt = datetime.fromisoformat(approved_at.replace("Z", "+00:00"))
|
|
approved_at = dt.strftime("%B %d, %Y")
|
|
except (ValueError, TypeError):
|
|
pass
|
|
|
|
meta_data = [
|
|
[
|
|
Paragraph("Customer", styles["MetaLabel"]),
|
|
Paragraph("Site Address", styles["MetaLabel"]),
|
|
],
|
|
[
|
|
Paragraph(proposal.get("customerName", ""), styles["MetaValue"]),
|
|
Paragraph(proposal.get("customerAddress", ""), styles["MetaValue"]),
|
|
],
|
|
[
|
|
Paragraph("Work Order #", styles["MetaLabel"]),
|
|
Paragraph("Date", styles["MetaLabel"]),
|
|
],
|
|
[
|
|
Paragraph(proposal.get("workOrderNumber", ""), styles["MetaValue"]),
|
|
Paragraph(approved_at or submitted_at, styles["MetaValue"]),
|
|
],
|
|
[
|
|
Paragraph("Category", styles["MetaLabel"]),
|
|
Paragraph("Priority", styles["MetaLabel"]),
|
|
],
|
|
[
|
|
Paragraph(proposal.get("serviceCategory", ""), styles["MetaValue"]),
|
|
Paragraph(proposal.get("priority", ""), styles["MetaValue"]),
|
|
],
|
|
]
|
|
|
|
meta_table = Table(meta_data, colWidths=[3.5 * inch, 3.5 * inch])
|
|
meta_table.setStyle(
|
|
TableStyle(
|
|
[
|
|
("VALIGN", (0, 0), (-1, -1), "TOP"),
|
|
("TOPPADDING", (0, 0), (-1, -1), 2),
|
|
("BOTTOMPADDING", (0, 0), (-1, -1), 2),
|
|
]
|
|
)
|
|
)
|
|
|
|
return [meta_table]
|
|
|
|
|
|
def _build_scope(proposal: dict, styles) -> list:
|
|
scope = proposal.get("refinedScope") or proposal.get("scopeOfWork", "")
|
|
if not scope:
|
|
return []
|
|
|
|
return [
|
|
Paragraph("Scope of Work", styles["SectionHeader"]),
|
|
Paragraph(scope, styles["ScopeText"]),
|
|
]
|
|
|
|
|
|
def _build_line_items_table(line_items: list[dict], styles) -> list:
|
|
if not line_items:
|
|
return [Paragraph("No line items", styles["Normal"])]
|
|
|
|
elements = [Paragraph("Itemized Pricing", styles["SectionHeader"])]
|
|
|
|
header = ["#", "Description", "Qty", "Unit", "Unit Price", "Total"]
|
|
|
|
table_data = [header]
|
|
subtotal = 0.0
|
|
|
|
for i, li in enumerate(line_items, 1):
|
|
qty = li.get("quantity", "")
|
|
unit = li.get("unit", "")
|
|
unit_price = li.get("unitPrice")
|
|
total_price = li.get("totalPrice", 0)
|
|
pricing_mode = li.get("pricingMode", "TotalPrice")
|
|
|
|
subtotal += float(total_price or 0)
|
|
|
|
if pricing_mode == "TotalPrice":
|
|
up_str = "-"
|
|
elif unit_price is not None:
|
|
up_str = f"${float(unit_price):,.2f}"
|
|
else:
|
|
up_str = "-"
|
|
|
|
tp_str = f"${float(total_price):,.2f}" if total_price else "-"
|
|
|
|
row = [
|
|
str(i),
|
|
Paragraph(li.get("description", ""), styles["Normal"]),
|
|
str(qty) if qty else "",
|
|
unit,
|
|
up_str,
|
|
tp_str,
|
|
]
|
|
table_data.append(row)
|
|
|
|
col_widths = [
|
|
0.35 * inch,
|
|
3.15 * inch,
|
|
0.55 * inch,
|
|
0.7 * inch,
|
|
1.0 * inch,
|
|
1.0 * inch,
|
|
]
|
|
table = Table(table_data, colWidths=col_widths, repeatRows=1)
|
|
|
|
table.setStyle(
|
|
TableStyle(
|
|
[
|
|
# Header row
|
|
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a237e")),
|
|
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
|
|
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
|
|
("FONTSIZE", (0, 0), (-1, 0), 9),
|
|
("BOTTOMPADDING", (0, 0), (-1, 0), 6),
|
|
("TOPPADDING", (0, 0), (-1, 0), 6),
|
|
# Data rows
|
|
("FONTSIZE", (0, 1), (-1, -1), 9),
|
|
("TOPPADDING", (0, 1), (-1, -1), 4),
|
|
("BOTTOMPADDING", (0, 1), (-1, -1), 4),
|
|
("VALIGN", (0, 0), (-1, -1), "MIDDLE"),
|
|
# Alignment
|
|
("ALIGN", (0, 0), (0, -1), "CENTER"),
|
|
("ALIGN", (2, 0), (2, -1), "CENTER"),
|
|
("ALIGN", (3, 0), (3, -1), "CENTER"),
|
|
("ALIGN", (4, 0), (4, -1), "RIGHT"),
|
|
("ALIGN", (5, 0), (5, -1), "RIGHT"),
|
|
# Grid
|
|
("LINEBELOW", (0, 0), (-1, 0), 1, colors.HexColor("#1a237e")),
|
|
("LINEBELOW", (0, 1), (-1, -2), 0.5, colors.HexColor("#e0e0e0")),
|
|
("LINEBELOW", (0, -1), (-1, -1), 1, colors.HexColor("#1a237e")),
|
|
# Alternating row colors
|
|
*[
|
|
("BACKGROUND", (0, i), (-1, i), colors.HexColor("#f5f5f5"))
|
|
for i in range(2, len(table_data), 2)
|
|
],
|
|
]
|
|
)
|
|
)
|
|
|
|
elements.append(table)
|
|
elements.append(Spacer(1, 0.15 * inch))
|
|
|
|
# Total row
|
|
total_data = [
|
|
["", "", "", "", "TOTAL:", f"${subtotal:,.2f}"],
|
|
]
|
|
total_table = Table(total_data, colWidths=col_widths)
|
|
total_table.setStyle(
|
|
TableStyle(
|
|
[
|
|
("FONTNAME", (0, 0), (-1, -1), "Helvetica-Bold"),
|
|
("FONTSIZE", (0, 0), (-1, -1), 11),
|
|
("ALIGN", (4, 0), (4, 0), "RIGHT"),
|
|
("ALIGN", (5, 0), (5, 0), "RIGHT"),
|
|
("TOPPADDING", (0, 0), (-1, -1), 4),
|
|
("LINEABOVE", (4, 0), (5, 0), 1.5, colors.HexColor("#1a237e")),
|
|
]
|
|
)
|
|
)
|
|
elements.append(total_table)
|
|
|
|
return elements
|
|
|
|
|
|
def _build_terms(styles) -> list:
|
|
elements = [
|
|
Spacer(1, 0.2 * inch),
|
|
Paragraph("Terms & Conditions", styles["SectionHeader"]),
|
|
]
|
|
|
|
for line in TERMS_AND_CONDITIONS.split("\n"):
|
|
elements.append(Paragraph(line, styles["TermsText"]))
|
|
|
|
return elements
|
|
|
|
|
|
def _page_footer(canvas, doc):
|
|
canvas.saveState()
|
|
page_num = canvas.getPageNumber()
|
|
footer_text = f"Page {page_num}"
|
|
canvas.setFont("Helvetica", 8)
|
|
canvas.setFillColor(colors.HexColor("#888888"))
|
|
canvas.drawCentredString(letter[0] / 2, 0.4 * inch, footer_text)
|
|
canvas.drawString(
|
|
0.75 * inch,
|
|
0.4 * inch,
|
|
f"{COMPANY_NAME} — Confidential",
|
|
)
|
|
canvas.restoreState()
|
|
|
|
|
|
def upload_pdf(s3_key: str, pdf_bytes: bytes):
|
|
try:
|
|
s3.put_object(
|
|
Bucket=GENERATED_BUCKET,
|
|
Key=s3_key,
|
|
Body=pdf_bytes,
|
|
ContentType="application/pdf",
|
|
)
|
|
except Exception as e:
|
|
# Fix: LAM-M5 — include stack trace in error logging
|
|
logger.exception("Error uploading PDF: %s", e)
|
|
raise
|
|
|
|
|
|
def register_pdf(proposal_id: str, s3_key: str):
|
|
try:
|
|
resp = _retry_request(
|
|
"POST",
|
|
f"{API_BASE_URL}/api/generated-pdfs",
|
|
json={"proposalId": proposal_id, "s3Key": s3_key},
|
|
headers=_api_headers(),
|
|
)
|
|
if resp.status_code not in (200, 201):
|
|
logger.error("Failed to register PDF: %s %s", resp.status_code, resp.text)
|
|
except Exception as e:
|
|
# Fix: LAM-M5 — include stack trace in error logging
|
|
logger.exception("Error registering PDF: %s", e)
|
|
|
|
|
|
def _api_headers() -> dict:
|
|
headers = {"Content-Type": "application/json"}
|
|
api_key = _get_api_key()
|
|
if api_key:
|
|
headers["X-Internal-Api-Key"] = api_key
|
|
return headers
|
|
|
|
|
|
def _retry_request(
|
|
method: str, url: str, *, max_retries: int = 3, **kwargs
|
|
) -> httpx.Response:
|
|
kwargs.setdefault("timeout", 10)
|
|
for attempt in range(max_retries):
|
|
try:
|
|
resp = httpx.request(method, url, **kwargs)
|
|
if resp.status_code < 500:
|
|
return resp
|
|
except (httpx.ConnectError, httpx.ReadTimeout, httpx.WriteTimeout) as exc:
|
|
if attempt == max_retries - 1:
|
|
raise
|
|
logger.warning(
|
|
"Retryable error (attempt %d/%d): %s", attempt + 1, max_retries, exc
|
|
)
|
|
time.sleep(min(2**attempt, 4))
|
|
return resp # type: ignore[possibly-undefined]
|