RugMunchMedia/pryscraper
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(x402): extract constants and enums to pry_x402 sub-package
Some checks are pending
CI / lint (pull_request) Waiting to run
Details
CI / typecheck (pull_request) Waiting to run
Details
CI / test (pull_request) Waiting to run
Details
CI / Secret scan (gitleaks) (pull_request) Waiting to run
Details
CI / Security audit (bandit) (pull_request) Waiting to run
Details

Browse source
This commit is contained in:
Crypto Rug Munch 2026-07-03 16:04:51 +02:00
parent 090f284c37
commit 089a2c77f7
4 changed files with 298 additions and 108 deletions
Download patch file Download diff file Expand all files Collapse all files

44
pry_x402/__init__.py Normal file
View file

@ -0,0 +1,44 @@
"""Pry — x402 payment protocol sub-package.
Split from x402.py for maintainability.
"""
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Rug Munch Media LLC
#
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.
from __future__ import annotations
from pry_x402.constants import (
ASSET_DECIMALS,
EIP7702_RPC,
X402_DEFAULT_ASSET,
X402_DEFAULT_NETWORK,
X402_DIR,
X402_FACILITATOR_URL,
X402_OFFLINE_MODE,
X402_PRICING,
X402_TIMEOUT_SECONDS,
X402_WALLET,
X402Asset,
X402Network,
X402Scheme,
)
__all__ = [
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"X402Asset",
"X402Network",
"X402Scheme",
]

141
pry_x402/constants.py Normal file
View file

@ -0,0 +1,141 @@
"""Pry — x402 constants and enums.
Split from x402.py for maintainability and to avoid import cycles.
"""
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Rug Munch Media LLC
#
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.
from __future__ import annotations
import os
from enum import StrEnum
from typing import Any
from paths import PRY_DATA_DIR
X402_DIR = PRY_DATA_DIR / "x402"
X402_DIR.mkdir(parents=True, exist_ok=True)
# x402 pricing per operation (USDC atomic units, 6 decimals)
X402_PRICING: dict[str, dict[str, Any]] = {
"scrape": {"price_usd": 0.001, "description": "Single URL scrape"},
"crawl": {"price_usd": 0.01, "description": "Crawl up to 10 pages"},
"extract": {"price_usd": 0.005, "description": "Structured extraction"},
"monitor": {"price_usd": 0.02, "description": "Create scheduled monitor"},
"llm_call": {"price_usd": 0.01, "description": "LLM extraction call"},
"template_execute": {"price_usd": 0.002, "description": "Execute scraper template"},
"template_batch_execute": {"price_usd": 0.01, "description": "Execute up to 20 template items"},
"browser_automation": {"price_usd": 0.05, "description": "Browser automation"},
"bulk_crawl": {"price_usd": 0.10, "description": "Crawl up to 1000 pages"},
"pdf_extract": {"price_usd": 0.01, "description": "PDF table extraction"},
"ocr_extract": {"price_usd": 0.005, "description": "Image OCR"},
"graphql_query": {"price_usd": 0.003, "description": "GraphQL query execution"},
"schema_extract": {"price_usd": 0.002, "description": "Schema.org/JSON-LD extraction"},
}
class X402Scheme(StrEnum):
"""Payment schemes supported by x402."""
EXACT = "exact"
UPTO = "upto"
class X402Network(StrEnum):
"""Blockchain networks supported by x402."""
BASE = "base"
SOLANA = "solana"
ETHEREUM = "ethereum"
POLYGON = "polygon"
ARBITRUM = "arbitrum"
OPTIMISM = "optimism"
BNB = "bnb"
AVALANCHE = "avalanche"
BASE_SEPOLIA = "base-sepolia"
TRON = "tron"
BITCOIN = "bitcoin"
LITECOIN = "litecoin"
DOGECOIN = "dogecoin"
class X402Asset(StrEnum):
"""Token assets supported by x402."""
USDC = "USDC"
USDT = "USDT"
DAI = "DAI"
ETH = "ETH"
SOL = "SOL"
MATIC = "MATIC"
TRX = "TRX"
BTC = "BTC"
LTC = "LTC"
DOGE = "DOGE"
# Prefer gopass for the x402 wallet and facilitator URL. Env vars override only
# if the backend returns an empty string. This lets you:
# gopass insert -m pry/x402_wallet # persistent across restarts
# PRY_X402_WALLET=0x... # one-off override (CI, dev)
try:
from secrets_backend import get_secret as _gs
X402_WALLET = _gs("x402_wallet") or os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = _gs("x402_facilitator") or os.getenv(
"PRY_X402_FACILITATOR", "https://x402.org/facilitator"
)
except ImportError:
X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator")
X402_DEFAULT_NETWORK = os.getenv("PRY_X402_NETWORK", X402Network.BASE.value)
X402_DEFAULT_ASSET = os.getenv("PRY_X402_ASSET", X402Asset.USDC.value)
X402_OFFLINE_MODE = os.getenv("PRY_X402_OFFLINE", "false").lower() in ("1", "true", "yes")
X402_TIMEOUT_SECONDS = float(os.getenv("PRY_X402_TIMEOUT", "15"))
# EIP-7702 self-verify RPC endpoints (public defaults, override via env)
EIP7702_RPC: dict[str, str] = {
"base": os.getenv("PRY_X402_BASE_RPC", "https://mainnet.base.org"),
"ethereum": os.getenv("PRY_X402_ETH_RPC", "https://eth.llamarpc.com"),
"polygon": os.getenv("PRY_X402_POLYGON_RPC", "https://polygon.drpc.org"),
"arbitrum": os.getenv("PRY_X402_ARBITRUM_RPC", "https://arb1.arbitrum.io/rpc"),
"optimism": os.getenv("PRY_X402_OPTIMISM_RPC", "https://mainnet.optimism.io"),
"bnb": os.getenv("PRY_X402_BNB_RPC", "https://bsc-dataseed.binance.org"),
"avalanche": os.getenv("PRY_X402_AVALANCHE_RPC", "https://api.avax.network/ext/bc/C/rpc"),
"base-sepolia": os.getenv("PRY_X402_BASE_SEPOLIA_RPC", "https://sepolia.base.org"),
}
# Asset decimals for atomic conversion
ASSET_DECIMALS: dict[str, int] = {
X402Asset.USDC.value: 6,
X402Asset.USDT.value: 6,
X402Asset.DAI.value: 18,
X402Asset.ETH.value: 18,
X402Asset.SOL.value: 9,
X402Asset.MATIC.value: 18,
X402Asset.TRX.value: 6,
X402Asset.BTC.value: 8,
X402Asset.LTC.value: 8,
X402Asset.DOGE.value: 8,
}
__all__ = [
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"X402Asset",
"X402Network",
"X402Scheme",
]

98
tests/test_x402_surface_snapshot.py Normal file
View file

@ -0,0 +1,98 @@
"""Snapshot tests for the x402 public API surface.
These tests guard against accidental breakage when refactoring
x402.py into the pry_x402/ sub-package.
"""
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Rug Munch Media LLC
#
# Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper
# Licensed under MIT. See LICENSE.
from __future__ import annotations
class TestX402PublicSurface:
"""The x402 module must expose the same public API as before."""
def test_required_symbols_exist(self) -> None:
import x402
required = {
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402Asset",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402Network",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402Scheme",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
"FacilitatorConfig",
"FacilitatorRouter",
"X402Handler",
"build_payment_required",
"create_batch_payment",
"create_payment_request",
"get_facilitator_router",
"is_batch_paid",
"require_payment_legacy",
"set_facilitator_router",
"verify_batch_payment",
}
missing = required - set(dir(x402))
assert not missing, f"Missing x402 public symbols: {missing}"
def test_pry_x402_re_exports_constants(self) -> None:
import pry_x402
required = {
"ASSET_DECIMALS",
"EIP7702_RPC",
"X402Asset",
"X402_DEFAULT_ASSET",
"X402_DEFAULT_NETWORK",
"X402_DIR",
"X402_FACILITATOR_URL",
"X402Network",
"X402_OFFLINE_MODE",
"X402_PRICING",
"X402Scheme",
"X402_TIMEOUT_SECONDS",
"X402_WALLET",
}
missing = required - set(dir(pry_x402))
assert not missing, f"Missing pry_x402 public symbols: {missing}"
def test_enum_values_unchanged(self) -> None:
from x402 import X402Asset, X402Network, X402Scheme
assert X402Scheme.EXACT == "exact"
assert X402Scheme.UPTO == "upto"
assert X402Network.BASE == "base"
assert X402Asset.USDC == "USDC"
def test_pricing_keys_unchanged(self) -> None:
import x402
expected = {
"scrape",
"crawl",
"extract",
"monitor",
"llm_call",
"template_execute",
"template_batch_execute",
"browser_automation",
"bulk_crawl",
"pdf_extract",
"ocr_extract",
"graphql_query",
"schema_extract",
}
assert set(x402.X402_PRICING.keys()) == expected

123
x402.py
View file

@ -28,121 +28,28 @@ import threading
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import StrEnum
from typing import Any, cast
import httpx
from paths import PRY_DATA_DIR
from pry_x402.constants import (
ASSET_DECIMALS,
EIP7702_RPC,
X402_DEFAULT_ASSET,
X402_DEFAULT_NETWORK,
X402_DIR,
X402_FACILITATOR_URL,
X402_OFFLINE_MODE,
X402_PRICING,
X402_TIMEOUT_SECONDS,
X402_WALLET,
X402Asset,
X402Network,
X402Scheme,
)
logger = logging.getLogger(__name__)
X402_DIR = PRY_DATA_DIR / "x402"
X402_DIR.mkdir(parents=True, exist_ok=True)
# x402 pricing per operation (USDC atomic units, 6 decimals)
X402_PRICING: dict[str, dict[str, Any]] = {
"scrape": {"price_usd": 0.001, "description": "Single URL scrape"},
"crawl": {"price_usd": 0.01, "description": "Crawl up to 10 pages"},
"extract": {"price_usd": 0.005, "description": "Structured extraction"},
"monitor": {"price_usd": 0.02, "description": "Create scheduled monitor"},
"llm_call": {"price_usd": 0.01, "description": "LLM extraction call"},
"template_execute": {"price_usd": 0.002, "description": "Execute scraper template"},
"template_batch_execute": {"price_usd": 0.01, "description": "Execute up to 20 template items"},
"browser_automation": {"price_usd": 0.05, "description": "Browser automation"},
"bulk_crawl": {"price_usd": 0.10, "description": "Crawl up to 1000 pages"},
"pdf_extract": {"price_usd": 0.01, "description": "PDF table extraction"},
"ocr_extract": {"price_usd": 0.005, "description": "Image OCR"},
"graphql_query": {"price_usd": 0.003, "description": "GraphQL query execution"},
"schema_extract": {"price_usd": 0.002, "description": "Schema.org/JSON-LD extraction"},
}
class X402Scheme(StrEnum):
"""Payment schemes supported by x402."""
EXACT = "exact"
UPTO = "upto"
class X402Network(StrEnum):
"""Blockchain networks supported by x402."""
BASE = "base"
SOLANA = "solana"
ETHEREUM = "ethereum"
POLYGON = "polygon"
ARBITRUM = "arbitrum"
OPTIMISM = "optimism"
BNB = "bnb"
AVALANCHE = "avalanche"
BASE_SEPOLIA = "base-sepolia"
TRON = "tron"
BITCOIN = "bitcoin"
LITECOIN = "litecoin"
DOGECOIN = "dogecoin"
class X402Asset(StrEnum):
"""Token assets supported by x402."""
USDC = "USDC"
USDT = "USDT"
DAI = "DAI"
ETH = "ETH"
SOL = "SOL"
MATIC = "MATIC"
TRX = "TRX"
BTC = "BTC"
LTC = "LTC"
DOGE = "DOGE"
# Prefer gopass for the x402 wallet and facilitator URL. Env vars override only
# if the backend returns an empty string. This lets you:
# gopass insert -m pry/x402_wallet # persistent across restarts
# PRY_X402_WALLET=0x... # one-off override (CI, dev)
try:
from secrets_backend import get_secret as _gs
X402_WALLET = _gs("x402_wallet") or os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = _gs("x402_facilitator") or os.getenv(
"PRY_X402_FACILITATOR", "https://x402.org/facilitator"
)
except ImportError:
X402_WALLET = os.getenv("PRY_X402_WALLET", "pry-default-wallet")
X402_FACILITATOR_URL = os.getenv("PRY_X402_FACILITATOR", "https://x402.org/facilitator")
X402_DEFAULT_NETWORK = os.getenv("PRY_X402_NETWORK", X402Network.BASE.value)
X402_DEFAULT_ASSET = os.getenv("PRY_X402_ASSET", X402Asset.USDC.value)
X402_OFFLINE_MODE = os.getenv("PRY_X402_OFFLINE", "false").lower() in ("1", "true", "yes")
X402_TIMEOUT_SECONDS = float(os.getenv("PRY_X402_TIMEOUT", "15"))
# EIP-7702 self-verify RPC endpoints (public defaults, override via env)
EIP7702_RPC: dict[str, str] = {
"base": os.getenv("PRY_X402_BASE_RPC", "https://mainnet.base.org"),
"ethereum": os.getenv("PRY_X402_ETH_RPC", "https://eth.llamarpc.com"),
"polygon": os.getenv("PRY_X402_POLYGON_RPC", "https://polygon.drpc.org"),
"arbitrum": os.getenv("PRY_X402_ARBITRUM_RPC", "https://arb1.arbitrum.io/rpc"),
"optimism": os.getenv("PRY_X402_OPTIMISM_RPC", "https://mainnet.optimism.io"),
"bnb": os.getenv("PRY_X402_BNB_RPC", "https://bsc-dataseed.binance.org"),
"avalanche": os.getenv("PRY_X402_AVALANCHE_RPC", "https://api.avax.network/ext/bc/C/rpc"),
"base-sepolia": os.getenv("PRY_X402_BASE_SEPOLIA_RPC", "https://sepolia.base.org"),
}
# Asset decimals for atomic conversion
ASSET_DECIMALS: dict[str, int] = {
X402Asset.USDC.value: 6,
X402Asset.USDT.value: 6,
X402Asset.DAI.value: 18,
X402Asset.ETH.value: 18,
X402Asset.SOL.value: 9,
X402Asset.MATIC.value: 18,
X402Asset.TRX.value: 6,
X402Asset.BTC.value: 8,
X402Asset.LTC.value: 8,
X402Asset.DOGE.value: 8,
}
def _to_atomic(amount_usd: float, asset: str = X402_DEFAULT_ASSET) -> int:
"""Convert USD amount to atomic units for the given asset."""