Submitting RCRA Hazardous Waste Manifests in Python

This page solves one specific production failure: a hazardous-waste shipment leaves a generator site, the routing pipeline submits the electronic manifest to the EPA e-Manifest system, and either the submission is rejected for a bad waste code, a network retry silently files the manifest twice under two different tracking numbers, or the declared quantity does not reconcile with the weighbridge because the unit code was wrong.

Filing an electronic Uniform Hazardous Waste Manifest — EPA Form 8700-22 — is the regulated hand-off that ends a hazardous-waste collection route. The generator’s EPA identification number, the RCRA waste codes, and the declared quantity all have to be right on the first submission, and the returned Manifest Tracking Number (MTN) is the durable key every later audit references. Submitting RCRA hazardous-waste manifests in Python means building a typed manifest record, posting it with an idempotency key so a retry cannot create a duplicate MTN, and capturing the tracking number into a compliance record you can seal into an Audit Report Generation bundle later. It is the submission endpoint of the EPA e-Manifest Integration subsystem.

Environment & Data Prerequisites

The submission client needs an HTTP library and a validation library; everything else is the standard library. Pin both so a resubmission during an audit reproduces exactly:

python -m pip install \
  requests==2.32.3 \
  pydantic==2.9.2 \
  pytest==8.2.0

Each manifest is assembled from the routing pipeline’s shipment record. These are the fields the submission reasons over, with the rule each must satisfy before it leaves the process:

Field	Unit / type	Range / rule
`generator_epa_id`	string	12 chars, EPA-assigned generator ID
`waste_codes`	list of strings	each a valid RCRA code, e.g. `D001` (pattern `[A-Z]\d{3}`)
`quantity`	int	> 0, in the unit named by `unit_code`
`unit_code`	enum	`K` kilograms, `P` pounds, `T` tons, `M` metric tons
`container_count`	int	> 0
`container_code`	enum	e.g. `DM` (metal drum), `DF` (fiber drum)

Two rules matter before any code. The quantity is an integer in exactly the unit named by unit_code — never a bare float and never a value whose unit is implied — because a tonnage that is right in value but wrong in unit is the mismatch that fails reconciliation against the weighbridge. And the declared tonnage does not originate here: it flows from the Capacity & Weight Limits model that already carries payload in integer kilograms, so the manifest quantity should be derived from that same integer rather than re-measured.

Step-by-Step Implementation

Step 1 — Model and build the manifest record

The manifest is a typed record whose validators reject the three most common causes of a rejected submission — a malformed generator ID, an out-of-pattern waste code, and a non-positive quantity — before the network is ever touched.

from __future__ import annotations
import re
from enum import Enum
from pydantic import BaseModel, Field, field_validator

WASTE_CODE = re.compile(r"^[A-Z]\d{3}$")   # e.g. D001, F005, U134


class UnitCode(str, Enum):
    KILOGRAMS = "K"
    POUNDS = "P"
    TONS = "T"
    METRIC_TONS = "M"


class ManifestRecord(BaseModel):
    generator_epa_id: str = Field(min_length=12, max_length=12)
    waste_codes: list[str] = Field(min_length=1)
    quantity: int = Field(gt=0)                 # integer, in unit_code units
    unit_code: UnitCode
    container_count: int = Field(gt=0)
    container_code: str = Field(min_length=2, max_length=2)

    @field_validator("waste_codes")
    @classmethod
    def valid_waste_codes(cls, codes: list[str]) -> list[str]:
        bad = [c for c in codes if not WASTE_CODE.match(c)]
        if bad:
            raise ValueError(f"malformed RCRA waste codes: {bad}")
        return codes


shipment = ManifestRecord(
    generator_epa_id="OHD987654321",
    waste_codes=["D001"],                       # ignitable waste
    quantity=9300,                              # kilograms, from the payload model
    unit_code=UnitCode.KILOGRAMS,
    container_count=6,
    container_code="DM",                        # metal drums
)

Step 2 — Submit with an idempotency key

The submission must survive a network retry without creating a second manifest. The defense is a deterministic idempotency key: hash the canonical manifest, send that hash as an Idempotency-Key header, and the endpoint treats a repeat of the same key as a replay of the original request rather than a new filing.

import hashlib
import json
import requests


def idempotency_key(manifest: ManifestRecord) -> str:
    """A content-derived key: identical manifests produce an identical key."""
    canonical = json.dumps(
        manifest.model_dump(mode="json"), sort_keys=True, separators=(",", ":")
    ).encode("utf-8")
    return hashlib.sha256(canonical).hexdigest()


class ManifestRejected(Exception):
    def __init__(self, errors: list[dict]):
        self.errors = errors
        super().__init__(str(errors))


class ManifestSubmitter:
    def __init__(self, base_url: str, api_token: str, session: requests.Session | None = None):
        self.base_url = base_url.rstrip("/")
        self.session = session or requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_token}"})

    def submit(self, manifest: ManifestRecord) -> dict:
        resp = self.session.post(
            f"{self.base_url}/api/v1/emanifest/save",
            json=manifest.model_dump(mode="json"),
            headers={"Idempotency-Key": idempotency_key(manifest)},
            timeout=30,
        )
        if resp.status_code == 400:
            raise ManifestRejected(resp.json().get("errors", []))
        resp.raise_for_status()
        return resp.json()

Because the key is derived from the manifest content, a resend triggered by a timeout carries the same key and the e-Manifest system returns the tracking number it already assigned — one shipment, one MTN, no duplicates.

Step 3 — Capture the Manifest Tracking Number

The MTN is the durable identity of the shipment: a nine-digit number followed by three letters. Capture it into a typed receipt and validate its shape so a malformed or empty tracking number never propagates into the compliance record.

MTN_PATTERN = re.compile(r"^\d{9}[A-Z]{3}$")   # e.g. 012345678ELC


class ManifestReceipt(BaseModel):
    manifest_tracking_number: str
    status: str

    @field_validator("manifest_tracking_number")
    @classmethod
    def valid_mtn(cls, mtn: str) -> str:
        if not MTN_PATTERN.match(mtn):
            raise ValueError(f"malformed Manifest Tracking Number: {mtn!r}")
        return mtn


def capture_receipt(response: dict) -> ManifestReceipt:
    return ManifestReceipt(
        manifest_tracking_number=response["manifestTrackingNumber"],
        status=response["status"],
    )

Step 4 — Serialize the compliance record

The final step binds the MTN back to the manifest content and the submission time into one compliance record. This record is the leaf a tamper-evident report later seals, so it carries every field an auditor resolves to a regulation.

from datetime import datetime, timezone


def build_compliance_record(manifest: ManifestRecord, receipt: ManifestReceipt) -> dict:
    return {
        "manifest_tracking_number": receipt.manifest_tracking_number,
        "status": receipt.status,
        "generator_epa_id": manifest.generator_epa_id,
        "waste_codes": manifest.waste_codes,
        "quantity": manifest.quantity,
        "unit_code": manifest.unit_code.value,
        "container_count": manifest.container_count,
        "container_code": manifest.container_code,
        "submitted_at": datetime.now(timezone.utc).isoformat(),
    }

Compliance Output

A successful submission of the shipment above serializes to this record, written to the append-only compliance store:

{
  "manifest_tracking_number": "012345678ELC",
  "status": "Scheduled",
  "generator_epa_id": "OHD987654321",
  "waste_codes": ["D001"],
  "quantity": 9300,
  "unit_code": "K",
  "container_count": 6,
  "container_code": "DM",
  "submitted_at": "2026-07-16T14:22:05.481920+00:00"
}

Each field earns its place under the RCRA generator standards:

manifest_tracking_number — the EPA-assigned MTN, the primary key every subsequent record, correction, and audit references under 40 CFR 262 Subpart B (the e-Manifest recordkeeping requirement).
generator_epa_id — the site identification number that establishes who is legally the generator of record under 40 CFR 262.12; a manifest with the wrong or unregistered ID is invalid on its face.
waste_codes — the RCRA hazardous-waste codes that classify the shipment and drive its permitted handling under 40 CFR 262.11 (hazardous-waste determination).
quantity and unit_code — the declared amount and its unit, the value that must reconcile against the weighbridge ticket; the unit is carried explicitly so a tonnage is never ambiguous.
submitted_at — the tz-aware UTC filing time, so the record can be ordered against the route and the driver’s duty log during reconciliation.

Because the compliance record is content-addressable through the MTN and never mutated in place, it drops directly into the audit report generation Merkle set as one sealed leaf.

Verification

The tests inject a fake requests.Session so the submission path is exercised end to end without a network, proving a clean submit captures a valid MTN and that an idempotent retry returns the same tracking number.

import pytest


class _FakeResponse:
    def __init__(self, status_code: int, body: dict):
        self.status_code = status_code
        self._body = body

    def json(self) -> dict:
        return self._body

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise requests.HTTPError(f"status {self.status_code}")


class _FakeSession:
    def __init__(self, response: _FakeResponse):
        self.response = response
        self.headers: dict = {}
        self.seen_keys: list[str] = []

    def post(self, url, json, headers, timeout):
        self.seen_keys.append(headers["Idempotency-Key"])
        return self.response


def test_clean_submit_captures_mtn() -> None:
    session = _FakeSession(_FakeResponse(200, {"manifestTrackingNumber": "012345678ELC", "status": "Scheduled"}))
    sub = ManifestSubmitter("https://mock.emanifest", "tok", session=session)
    receipt = capture_receipt(sub.submit(shipment))
    assert receipt.manifest_tracking_number == "012345678ELC"


def test_retry_reuses_the_same_idempotency_key() -> None:
    session = _FakeSession(_FakeResponse(200, {"manifestTrackingNumber": "012345678ELC", "status": "Scheduled"}))
    sub = ManifestSubmitter("https://mock.emanifest", "tok", session=session)
    sub.submit(shipment)
    sub.submit(shipment)                        # a retry of the same manifest
    assert session.seen_keys[0] == session.seen_keys[1]


def test_invalid_waste_code_raises_before_network() -> None:
    with pytest.raises(ValueError, match="malformed RCRA waste codes"):
        ManifestRecord(
            generator_epa_id="OHD987654321", waste_codes=["D9"], quantity=100,
            unit_code=UnitCode.KILOGRAMS, container_count=1, container_code="DM",
        )

Common Errors

ManifestRejected: [{'code': 'E_WASTE_CODE', 'message': "waste code 'D0001' is not a recognized EPA hazardous waste code"}] — the endpoint returned HTTP 400 because a waste code passed the local pattern but is not a real RCRA code. Root cause: the [A-Z]\d{3} regex accepts syntactically valid but non-existent codes. Fix: validate waste_codes against a maintained set of recognized codes before submitting, not just the shape, and surface the endpoint’s E_WASTE_CODE message to the operator rather than retrying.

Two Manifest Tracking Numbers for one shipment. Root cause: the first POST timed out after the server had already filed the manifest, and a naive retry without an idempotency key filed it a second time under a new MTN — a duplicate the EPA system now holds. Fix: always send the content-derived Idempotency-Key header (Step 2) so the endpoint replays the original tracking number on any retry instead of creating a new filing.

Quantity reconciles wrong against the weighbridge. Root cause: the manifest declared quantity=9300 with unit_code="T" (tons) when the value was actually kilograms, overstating the shipment by three orders of magnitude. Fix: derive both the value and the unit from the same integer-kilogram payload figure, keep unit_code an explicit enum, and add a reconciliation check that the declared quantity converts to the weighbridge reading within tolerance — never let the unit be implied. A kilogram quantity converts to metric tons by $t = q / 1000$ , and that conversion should be asserted, not assumed.

FAQ

Why derive the idempotency key from the manifest content instead of a random UUID?

A random key makes each attempt look unique, which is the opposite of what you want on a retry: the whole point is that the second attempt at the same shipment must be recognized as the same request. Hashing the canonical manifest means an identical resubmission carries an identical key and replays to the original MTN, while a genuinely different shipment naturally gets a different key.

What if the endpoint is unreachable and I never learn the MTN?

Hold the built manifest and its idempotency key in a durable outbox and retry with that same key. Because the key is content-derived, a later successful retry either files the manifest for the first time or replays the tracking number the server already assigned, so you converge on exactly one MTN once connectivity returns without risking a duplicate.

Up: EPA e-Manifest Integration

EPA e-Manifest Integration — the subsystem this submission endpoint belongs to
Capacity & Weight Limits — where the declared tonnage originates as integer kilograms
Audit Report Generation — sealing the captured MTN record into a tamper-evident report