Function extenddb_request

tests/test_import_export.py:33–62

Send a raw DynamoDB-format request to extenddb with SigV4 authentication.

(operation: str, body: dict)

ENDPOINT = os.environ.get("EXTENDDB_TEST_ENDPOINT", "http://localhost:8000").strip()
def extenddb_request(operation: str, body: dict) -> dict:
    """Send a raw DynamoDB-format request to extenddb with SigV4 authentication."""
    body_bytes = json.dumps(body).encode("utf-8")
    headers = {
        "X-Amz-Target": f"DynamoDB_20120810.{operation}",
        "Content-Type": "application/x-amz-json-1.0",
    }

    # Sign the request with SigV4 using env var credentials.
    access_key = os.environ.get("AWS_ACCESS_KEY_ID", "")
    secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY", "")
    region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
    if access_key and secret_key:
        creds = Credentials(access_key, secret_key)
        aws_req = AWSRequest(method="POST", url=ENDPOINT, data=body_bytes, headers=headers)
        SigV4Auth(creds, "dynamodb", region).add_auth(aws_req)
        headers = dict(aws_req.headers)

    resp = requests.post(
        ENDPOINT,
        data=body_bytes,
        headers=headers,
        verify=not ENDPOINT.startswith("https://"),
    )
    result = resp.json()
    if resp.status_code >= 400:
        error_type = result.get("__type", "Unknown")
        error_msg = result.get("message", result.get("Message", ""))
        raise RuntimeError(f"{error_type}: {error_msg} (HTTP {resp.status_code})")
    return result
@pytest.fixture()
def unique_table_name():
    """Generate a unique table name."""