diff --git a/payments_py/api/base_payments.py b/payments_py/api/base_payments.py index bb1fcc4..a8d8955 100644 --- a/payments_py/api/base_payments.py +++ b/payments_py/api/base_payments.py @@ -12,6 +12,9 @@ from payments_py.environments import get_environment from payments_py.common.helper import dict_keys_to_camel +# Default timeout for HTTP requests in seconds (connect, read) +DEFAULT_HTTP_TIMEOUT = (10, 30) + class BasePaymentsAPI: """ @@ -114,6 +117,7 @@ def get_backend_http_options( "Authorization": f"Bearer {self.nvm_api_key}", }, "verify": verify_ssl, + "timeout": DEFAULT_HTTP_TIMEOUT, } if body: # Convert to camelCase for consistency with TypeScript @@ -142,6 +146,7 @@ def get_public_http_options( "Content-Type": "application/json", }, "verify": verify_ssl, + "timeout": DEFAULT_HTTP_TIMEOUT, } if body: # Convert to camelCase for consistency with TypeScript diff --git a/payments_py/x402/__init__.py b/payments_py/x402/__init__.py index d64d821..c9e9201 100644 --- a/payments_py/x402/__init__.py +++ b/payments_py/x402/__init__.py @@ -72,7 +72,7 @@ from .resolve_scheme import resolve_scheme from .facilitator import NeverminedFacilitator from .facilitator_api import FacilitatorAPI -from .delegation_api import DelegationAPI, PaymentMethodSummary +from .delegation_api import DelegationAPI, DelegationSummary, PaymentMethodSummary from .a2a import X402A2AUtils, X402Metadata, PaymentStatus as X402PaymentStatus from .token import X402TokenAPI, decode_access_token @@ -127,6 +127,7 @@ "decode_access_token", # Delegation API "DelegationAPI", + "DelegationSummary", "PaymentMethodSummary", # High-level facilitator "NeverminedFacilitator", diff --git a/payments_py/x402/delegation_api.py b/payments_py/x402/delegation_api.py index 13a436b..68f6e44 100644 --- a/payments_py/x402/delegation_api.py +++ b/payments_py/x402/delegation_api.py @@ -2,11 +2,11 @@ Delegation API for managing card-delegation payment methods. Provides access to the user's enrolled Stripe payment methods -for use with the nvm:card-delegation x402 scheme. +and delegations for use with the nvm:card-delegation x402 scheme. """ import requests -from typing import List +from typing import Any, Dict, List, Optional from pydantic import BaseModel, ConfigDict, Field from payments_py.common.payments_error import PaymentsError from payments_py.common.types import PaymentOptions @@ -37,8 +37,40 @@ class PaymentMethodSummary(BaseModel): ) +class DelegationSummary(BaseModel): + """ + Summary of an existing card delegation. + + Attributes: + id: Delegation UUID + card_id: Associated PaymentMethod entity UUID + spending_limit_cents: Maximum spending limit in cents + spent_cents: Amount already spent in cents + duration_secs: Duration of the delegation in seconds + currency: Currency code (e.g., 'usd') + status: Delegation status (e.g., 'active', 'expired') + created_at: ISO 8601 creation timestamp + expires_at: ISO 8601 expiration timestamp + """ + + id: str + card_id: Optional[str] = Field(None, alias="cardId") + spending_limit_cents: Optional[int] = Field(None, alias="spendingLimitCents") + spent_cents: Optional[int] = Field(None, alias="spentCents") + duration_secs: Optional[int] = Field(None, alias="durationSecs") + currency: Optional[str] = None + status: Optional[str] = None + created_at: Optional[str] = Field(None, alias="createdAt") + expires_at: Optional[str] = Field(None, alias="expiresAt") + + model_config = ConfigDict( + populate_by_name=True, + from_attributes=True, + ) + + class DelegationAPI(BasePaymentsAPI): - """API for listing enrolled payment methods for card delegation.""" + """API for managing enrolled payment methods and delegations for card delegation.""" @classmethod def get_instance(cls, options: PaymentOptions) -> "DelegationAPI": @@ -77,3 +109,36 @@ def list_payment_methods(self) -> List[PaymentMethodSummary]: raise PaymentsError.internal( f"Network error while listing payment methods: {str(err)}" ) from err + + def list_delegations(self) -> List[DelegationSummary]: + """ + List the user's existing card delegations. + + Returns: + A list of delegation summaries + + Raises: + PaymentsError: If the request fails + """ + url = f"{self.environment.backend}/api/v1/delegation" + options = self.get_backend_http_options("GET") + + try: + response = requests.get(url, **options) + response.raise_for_status() + data = response.json() + return [DelegationSummary.model_validate(d) for d in data] + except requests.HTTPError as err: + try: + error_message = response.json().get( + "message", "Failed to list delegations" + ) + except Exception: + error_message = "Failed to list delegations" + raise PaymentsError.internal( + f"{error_message} (HTTP {response.status_code})" + ) from err + except Exception as err: + raise PaymentsError.internal( + f"Network error while listing delegations: {str(err)}" + ) from err diff --git a/payments_py/x402/types.py b/payments_py/x402/types.py index d2e13c8..f2b8d85 100644 --- a/payments_py/x402/types.py +++ b/payments_py/x402/types.py @@ -271,18 +271,29 @@ class CardDelegationConfig(BaseModel): """ Configuration for card delegation (fiat/Stripe) payments. + To reuse an existing delegation supply ``delegation_id``. + To reuse an existing card (PaymentMethod entity) supply ``card_id``. + When creating a brand-new delegation provide ``provider_payment_method_id``, + ``spending_limit_cents``, and ``duration_secs``. + Attributes: - provider_payment_method_id: Stripe payment method ID (e.g., 'pm_...') - spending_limit_cents: Maximum spending limit in cents - duration_secs: Duration of the delegation in seconds + card_id: PaymentMethod entity UUID -- preferred way to reference an enrolled card + delegation_id: Existing delegation UUID to reuse instead of creating a new one + provider_payment_method_id: Stripe payment method ID (e.g., 'pm_...'). Required only for new delegations. + spending_limit_cents: Maximum spending limit in cents. Required only for new delegations. + duration_secs: Duration of the delegation in seconds. Required only for new delegations. currency: Currency code (default: 'usd') merchant_account_id: Stripe Connect merchant account ID max_transactions: Maximum number of transactions allowed """ - provider_payment_method_id: str = Field(alias="providerPaymentMethodId") - spending_limit_cents: int = Field(alias="spendingLimitCents") - duration_secs: int = Field(alias="durationSecs") + card_id: Optional[str] = Field(None, alias="cardId") + delegation_id: Optional[str] = Field(None, alias="delegationId") + provider_payment_method_id: Optional[str] = Field( + None, alias="providerPaymentMethodId" + ) + spending_limit_cents: Optional[int] = Field(None, alias="spendingLimitCents") + duration_secs: Optional[int] = Field(None, alias="durationSecs") currency: Optional[str] = None merchant_account_id: Optional[str] = Field(None, alias="merchantAccountId") max_transactions: Optional[int] = Field(None, alias="maxTransactions") diff --git a/tests/integration/a2a/test_complete_message_send_flow.py b/tests/integration/a2a/test_complete_message_send_flow.py index 89a14d4..4f3cef1 100644 --- a/tests/integration/a2a/test_complete_message_send_flow.py +++ b/tests/integration/a2a/test_complete_message_send_flow.py @@ -557,11 +557,15 @@ async def test_non_blocking_execution_with_polling(): # The background task runs on the TestClient's ASGI event loop. That loop # only advances when the TestClient processes a request. After the last # poll that saw "completed", no further requests are in flight, so the - # background settlement may be starved of event-loop turns. We issue a - # few extra lightweight polls to give the loop enough ticks to schedule + # background settlement may be starved of event-loop turns. We keep + # issuing lightweight polls to give the loop enough ticks to schedule # the settlement (which itself runs settle_permissions via run_in_executor). - for _ in range(5): - if mock_payments.facilitator.settle_called.wait(timeout=0.1): + # + # IMPORTANT: We must keep making requests throughout the entire wait period. + # A blocking wait(timeout=N) without requests starves the event loop and + # the background task can never complete. + for _ in range(50): + if mock_payments.facilitator.settle_called.wait(timeout=0.2): break # Drive the ASGI event loop with a harmless poll client.post( @@ -575,6 +579,6 @@ async def test_non_blocking_execution_with_polling(): headers=headers, ) - settled = mock_payments.facilitator.settle_called.wait(timeout=10.0) + settled = mock_payments.facilitator.settle_called.is_set() assert settled, "Credits should be settled when task completes in non-blocking mode" assert mock_payments.facilitator.settle_call_count == 1