Checks OpenFDA Enforcement API for new Class I medical device recalls. Class I recalls are the most serious: reasonable probability that use of or exposure to a product will cause serious adverse health consequences or death.
| 1471 | # --------------------------------------------------------------------------- |
| 1472 | |
| 1473 | class FDARecallChecker: |
| 1474 | """Checks OpenFDA Enforcement API for new Class I medical device recalls. |
| 1475 | |
| 1476 | Class I recalls are the most serious: reasonable probability that use of or |
| 1477 | exposure to a product will cause serious adverse health consequences or death. |
| 1478 | """ |
| 1479 | |
| 1480 | ENFORCEMENT_URL = "https://api.fda.gov/device/enforcement.json" |
| 1481 | |
| 1482 | def __init__(self, api_key: str, state: dict): |
| 1483 | self.api_key = api_key |
| 1484 | self.state = state |
| 1485 | |
| 1486 | def check(self, source_id: str, source: dict) -> Optional[dict]: |
| 1487 | prev = self.state.get(source_id, {}) |
| 1488 | last_recall_nums = set(prev.get("seen_recall_numbers", [])) |
| 1489 | |
| 1490 | two_weeks_ago = (datetime.now() - timedelta(days=14)).strftime("%Y%m%d") |
| 1491 | today = datetime.now().strftime("%Y%m%d") |
| 1492 | |
| 1493 | search_q = ( |
| 1494 | f'classification:"Class I"' |
| 1495 | f' AND report_date:[{two_weeks_ago} TO {today}]' |
| 1496 | ) |
| 1497 | url = f"{self.ENFORCEMENT_URL}?search={requests.utils.quote(search_q)}&limit=20" |
| 1498 | if self.api_key: |
| 1499 | url += f"&api_key={self.api_key}" |
| 1500 | |
| 1501 | try: |
| 1502 | resp = requests.get(url, timeout=20) |
| 1503 | if resp.status_code == 404: |
| 1504 | self.state[source_id] = { |
| 1505 | "url": source["url"], |
| 1506 | "last_checked": datetime.now().isoformat(), |
| 1507 | "seen_recall_numbers": list(last_recall_nums), |
| 1508 | } |
| 1509 | return None |
| 1510 | resp.raise_for_status() |
| 1511 | results = resp.json().get("results", []) |
| 1512 | except Exception as e: |
| 1513 | print(f" ERROR OpenFDA Enforcement: {e}") |
| 1514 | return None |
| 1515 | |
| 1516 | new_items = [] |
| 1517 | all_recall_nums = list(last_recall_nums) |
| 1518 | |
| 1519 | for r in results: |
| 1520 | recall_num = r.get("recall_number", "") |
| 1521 | if not recall_num or recall_num in last_recall_nums: |
| 1522 | continue |
| 1523 | all_recall_nums.append(recall_num) |
| 1524 | |
| 1525 | product_desc = r.get("product_description", "")[:200] |
| 1526 | reason = r.get("reason_for_recall", "") |
| 1527 | firm = r.get("recalling_firm", "") |
| 1528 | report_date = r.get("report_date", "") |
| 1529 | quantity = r.get("product_quantity", "") |
| 1530 |