Scrape SAM.gov Contract Opportunities with Python (Search API + Dataset + Screenshots)

python -m venv .venv
source .venv/bin/activate
pip install requests tenacity python-dateutil pandas

import os
import time
import random
from dataclasses import dataclass
from typing import Any

import requests
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type


@dataclass
class SamGovConfig:
    api_key: str
    base_url: str = "https://api.sam.gov"
    # Common opportunities endpoint (update if SAM publishes a different path/version)
    opportunities_path: str = "/opportunities/v2/search"
    timeout: tuple[int, int] = (10, 45)  # connect, read
    min_delay_s: float = 0.4
    max_delay_s: float = 1.2


class SamGovClient:
    def __init__(self, cfg: SamGovConfig):
        self.cfg = cfg
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "proxiesapi-guides/1.0 (+https://proxiesapi.com)",
            "X-Api-Key": cfg.api_key,
        })

    def _polite_sleep(self):
        time.sleep(random.uniform(self.cfg.min_delay_s, self.cfg.max_delay_s))

    @retry(
        stop=stop_after_attempt(6),
        wait=wait_exponential(multiplier=1, min=2, max=30),
        retry=retry_if_exception_type((requests.RequestException,)),
        reraise=True,
    )
    def get(self, path: str, params: dict[str, Any]) -> dict:
        self._polite_sleep()
        url = self.cfg.base_url.rstrip("/") + path
        r = self.session.get(url, params=params, timeout=self.cfg.timeout)
        r.raise_for_status()
        return r.json()


def load_client() -> SamGovClient:
    api_key = os.getenv("SAMGOV_API_KEY")
    if not api_key:
        raise RuntimeError("Missing SAMGOV_API_KEY in environment")
    return SamGovClient(SamGovConfig(api_key=api_key))

from datetime import datetime, timedelta, timezone


def iso_date(d: datetime) -> str:
    # SAM endpoints often accept YYYY-MM-DD or ISO8601; keep it simple.
    return d.strftime("%Y-%m-%d")


def search_opportunities(
    client: SamGovClient,
    keyword: str | None = None,
    days_back: int = 7,
    page_size: int = 100,
    offset: int = 0,
) -> dict:
    end = datetime.now(timezone.utc)
    start = end - timedelta(days=days_back)

    params = {
        "limit": page_size,
        "offset": offset,
        # Parameter names vary by version; keep these configurable if needed.
        "postedFrom": iso_date(start),
        "postedTo": iso_date(end),
    }
    if keyword:
        params["q"] = keyword

    return client.get(client.cfg.opportunities_path, params=params)

if __name__ == "__main__":
    client = load_client()
    data = search_opportunities(client, keyword="software", days_back=3, page_size=25)
    print(data.keys())

from dateutil import parser as dtparse


def safe_dt(s: str | None):
    if not s:
        return None
    try:
        return dtparse.parse(s).isoformat()
    except Exception:
        return s


def normalize_notice(n: dict) -> dict:
    return {
        "notice_id": n.get("noticeId") or n.get("id") or n.get("notice_id"),
        "title": n.get("title") or n.get("noticeTitle"),
        "solicitation_number": n.get("solicitationNumber") or n.get("solicitation_number"),
        "agency": (n.get("department") or n.get("agency") or {}).get("name")
            if isinstance(n.get("department") or n.get("agency"), dict)
            else (n.get("department") or n.get("agency")),
        "office": (n.get("office") or {}).get("name") if isinstance(n.get("office"), dict) else n.get("office"),
        "posted_date": safe_dt(n.get("postedDate") or n.get("posted_date")),
        "response_deadline": safe_dt(n.get("responseDeadLine") or n.get("response_deadline") or n.get("dueDate")),
        "naics": n.get("naics") or n.get("naicsCode"),
        "place_of_performance": n.get("placeOfPerformance") or n.get("place_of_performance"),
        "type": n.get("type"),
        "set_aside": n.get("setAside") or n.get("set_aside"),
        "ui_url": n.get("uiLink") or n.get("url") or n.get("link"),
        "raw": n,
    }

import json
from typing import Iterable


def iter_notices(payload: dict) -> list[dict]:
    # Different versions use different keys.
    for key in ["opportunities", "data", "results", "notices"]:
        v = payload.get(key)
        if isinstance(v, list):
            return v
    return []


def crawl_all(
    client: SamGovClient,
    keyword: str | None = None,
    days_back: int = 7,
    page_size: int = 100,
    max_pages: int = 50,
) -> list[dict]:
    out = []

    for page in range(max_pages):
        offset = page * page_size
        payload = search_opportunities(
            client,
            keyword=keyword,
            days_back=days_back,
            page_size=page_size,
            offset=offset,
        )

        notices = iter_notices(payload)
        print(f"page {page+1}: {len(notices)}")

        for n in notices:
            out.append(normalize_notice(n))

        if len(notices) < page_size:
            break

    return out


if __name__ == "__main__":
    client = load_client()
    rows = crawl_all(client, keyword="cybersecurity", days_back=14, page_size=50)
    print("rows:", len(rows))
    print(json.dumps(rows[0], indent=2)[:800])

import pandas as pd


def export(rows: list[dict], stem: str = "samgov_opportunities"):
    jsonl_path = f"{stem}.jsonl"
    with open(jsonl_path, "w", encoding="utf-8") as f:
        for r in rows:
            # avoid duplicating raw huge objects in CSV workflows if you don’t need them
            f.write(json.dumps(r, ensure_ascii=False) + "\n")

    # flatten: drop raw for CSV by default
    flat = [{k: v for k, v in r.items() if k != "raw"} for r in rows]
    df = pd.DataFrame(flat)
    csv_path = f"{stem}.csv"
    df.to_csv(csv_path, index=False)

    print("wrote", jsonl_path, "and", csv_path, "rows:", len(rows))