Initial Python SDK release v1.0.0

This commit is contained in:
garfieldheron
2026-02-19 12:17:09 -05:00
commit 42443872c9
17 changed files with 1037 additions and 0 deletions

158
README.md Normal file
View File

@@ -0,0 +1,158 @@
# FetcherPay Python SDK
Official Python SDK for the FetcherPay API — One API. Every Rail.
## Installation
```bash
pip install fetcherpay
```
## Quick Start
```python
from fetcherpay import FetcherPay
client = FetcherPay(
api_key='fp_test_your_key',
environment='sandbox' # or 'production'
)
# Create a payment
payment = client.payments.create(
amount=10000, # $100.00 in cents
currency='USD',
source={'payment_method_id': 'pm_123'},
destination={'payment_method_id': 'pm_456'},
rail='auto' # Auto-select optimal rail
)
print(payment['id'], payment['status'])
```
## Configuration
| Option | Type | Default | Description |
|--------|------|---------|-------------|
| `api_key` | str | required | Your FetcherPay API key |
| `environment` | str | 'sandbox' | 'sandbox' or 'production' |
| `base_url` | str | auto | Override base URL |
| `timeout` | int | 30 | Request timeout (seconds) |
## API Reference
### Payments
```python
# Create payment
payment = client.payments.create(
amount=10000,
source={'payment_method_id': 'pm_123'},
destination={'payment_method_id': 'pm_456'},
idempotency_key='unique-key'
)
# Retrieve payment
payment = client.payments.retrieve('pay_xxx')
# List payments
payments = client.payments.list(limit=10)
# Cancel payment
client.payments.cancel('pay_xxx', reason='Customer request')
# Refund payment
client.payments.refund('pay_xxx', amount=5000, reason='Partial refund')
```
### Ledger
```python
# List accounts
accounts = client.ledger.list_accounts()
# Get account balance
account = client.ledger.retrieve_account('la_xxx')
print(account['balance']['available'])
# List entries
entries = client.ledger.list_entries(account_id='la_xxx')
```
### Payment Methods
```python
# Create bank account
pm = client.payment_methods.create(
type='bank_account',
bank_account={
'account_number': '000123456789',
'routing_number': '011000015',
'account_type': 'checking'
}
)
# List payment methods
methods = client.payment_methods.list()
```
### Webhooks
```python
# Create webhook endpoint
webhook = client.webhooks.create(
url='https://your-app.com/webhooks',
events=['payment.settled', 'payment.failed']
)
# Verify webhook signature
is_valid = client.verify_webhook_signature(
payload,
signature, # from X-FetcherPay-Signature header
webhook['secret']
)
```
## Error Handling
```python
from fetcherpay import (
FetcherPayError,
AuthenticationError,
ValidationError,
NotFoundError
)
try:
client.payments.create(...)
except AuthenticationError:
print('Invalid API key')
except ValidationError as e:
print(f'Validation failed: {e.param}')
except FetcherPayError as e:
print(f'API error: {e.type} ({e.status_code})')
```
## Webhook Verification
```python
import json
# In your webhook handler
def handle_webhook(request):
payload = request.body
signature = request.headers.get('X-FetcherPay-Signature')
secret = 'whsec_your_webhook_secret'
if client.verify_webhook_signature(payload, signature, secret):
event = json.loads(payload)
if event['type'] == 'payment.settled':
handle_payment_settled(event['data'])
return 'OK', 200
else:
return 'Invalid signature', 401
```
## License
MIT

51
examples/basic.py Normal file
View File

@@ -0,0 +1,51 @@
"""
FetcherPay Python SDK - Basic Example
"""
import os
from fetcherpay import FetcherPay
# Initialize client
client = FetcherPay(
api_key=os.environ.get('FETCHERPAY_API_KEY', 'sandbox'),
environment='sandbox'
)
def main():
try:
# Create a payment
print('Creating payment...')
payment = client.payments.create(
amount=10000, # $100.00 in cents
currency='USD',
source={'payment_method_id': 'pm_bank_123'},
destination={'payment_method_id': 'pm_merchant_456'},
rail='auto'
)
print(f'Payment created: {payment["id"]} Status: {payment["status"]}')
# Retrieve the payment
print('\nRetrieving payment...')
retrieved = client.payments.retrieve(payment['id'])
print(f'Retrieved: {retrieved["id"]} Timeline: {len(retrieved["timeline"])} events')
# List payments
print('\nListing payments...')
payments = client.payments.list(limit=5)
print(f'Found {len(payments["data"])} payments')
# List ledger accounts
print('\nListing ledger accounts...')
accounts = client.ledger.list_accounts()
print(f'Found {len(accounts["data"])} accounts')
for acc in accounts['data']:
print(f' - {acc["name"]}: ${acc["balance"]["available"] / 100} available')
except Exception as e:
print(f'Error: {e}')
raise
if __name__ == '__main__':
main()

View File

@@ -0,0 +1,41 @@
"""
FetcherPay Python SDK - Webhook Verification Example
"""
import json
from fetcherpay import FetcherPay
# Initialize client
client = FetcherPay(
api_key='sandbox',
environment='sandbox'
)
# Example webhook payload
payload = json.dumps({
'id': 'evt_123',
'type': 'payment.settled',
'created_at': '2026-02-18T20:00:00Z',
'data': {
'id': 'pay_456',
'status': 'settled',
'amount': 10000
}
})
# Your webhook secret from the dashboard
webhook_secret = 'whsec_your_secret_here'
# Simulate receiving a webhook
signature = 'sha256=...' # From X-FetcherPay-Signature header
# Verify the signature
is_valid = client.verify_webhook_signature(payload, signature, webhook_secret)
if is_valid:
print('✅ Webhook signature verified')
# Process the webhook event
event = json.loads(payload)
print(f'Event type: {event["type"]}')
else:
print('❌ Invalid webhook signature')

51
fetcherpay/__init__.py Normal file
View File

@@ -0,0 +1,51 @@
"""
FetcherPay Python SDK
One API. Every Rail.
Example:
>>> from fetcherpay import FetcherPay
>>>
>>> client = FetcherPay(
... api_key='fp_test_your_key',
... environment='sandbox'
... )
>>>
>>> payment = client.payments.create(
... amount=10000,
... currency='USD',
... source={'payment_method_id': 'pm_123'},
... destination={'payment_method_id': 'pm_456'}
... )
>>> print(payment.id)
"""
from .client import FetcherPay
from .exceptions import (
FetcherPayError,
AuthenticationError,
ValidationError,
NotFoundError,
)
from .types import (
Payment,
PaymentMethod,
LedgerAccount,
LedgerEntry,
WebhookEndpoint,
CreatePaymentRequest,
)
__version__ = '1.0.0'
__all__ = [
'FetcherPay',
'FetcherPayError',
'AuthenticationError',
'ValidationError',
'NotFoundError',
'Payment',
'PaymentMethod',
'LedgerAccount',
'LedgerEntry',
'WebhookEndpoint',
'CreatePaymentRequest',
]

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -0,0 +1 @@
# API modules

53
fetcherpay/api/ledger.py Normal file
View File

@@ -0,0 +1,53 @@
"""
Ledger API
"""
from typing import Optional
class LedgerAPI:
"""Ledger API client"""
def __init__(self, client):
self.client = client
def list_accounts(
self,
limit: int = 25,
cursor: Optional[str] = None,
type: Optional[str] = None
) -> dict:
"""List ledger accounts"""
params = {'limit': limit}
if cursor:
params['cursor'] = cursor
if type:
params['type'] = type
return self.client._request('GET', '/ledger/accounts', params=params)
def retrieve_account(self, account_id: str) -> dict:
"""Retrieve a ledger account"""
return self.client._request('GET', f'/ledger/accounts/{account_id}')
def list_entries(
self,
limit: int = 25,
cursor: Optional[str] = None,
account_id: Optional[str] = None,
payment_id: Optional[str] = None
) -> dict:
"""List ledger entries"""
params = {'limit': limit}
if cursor:
params['cursor'] = cursor
if account_id:
params['account_id'] = account_id
if payment_id:
params['payment_id'] = payment_id
return self.client._request('GET', '/ledger/entries', params=params)
def retrieve_entry(self, entry_id: str) -> dict:
"""Retrieve a ledger entry"""
return self.client._request('GET', f'/ledger/entries/{entry_id}')

View File

@@ -0,0 +1,73 @@
"""
Payment Methods API
"""
from typing import Optional, Dict, Any
class PaymentMethodsAPI:
"""Payment Methods API client"""
def __init__(self, client):
self.client = client
def create(
self,
type: str,
bank_account: Optional[Dict[str, Any]] = None,
card: Optional[Dict[str, Any]] = None,
usdc_wallet: Optional[Dict[str, Any]] = None,
metadata: Optional[Dict[str, Any]] = None,
idempotency_key: Optional[str] = None
) -> dict:
"""
Create a payment method
Args:
type: 'bank_account', 'card', or 'usdc_wallet'
bank_account: Bank account details
card: Card details
usdc_wallet: USDC wallet details
metadata: Optional metadata
idempotency_key: Idempotency key
"""
data = {'type': type}
if bank_account:
data['bank_account'] = bank_account
if card:
data['card'] = card
if usdc_wallet:
data['usdc_wallet'] = usdc_wallet
if metadata:
data['metadata'] = metadata
return self.client._request(
'POST',
'/payment-methods',
json=data,
idempotency_key=idempotency_key
)
def retrieve(self, payment_method_id: str) -> dict:
"""Retrieve a payment method"""
return self.client._request('GET', f'/payment-methods/{payment_method_id}')
def list(
self,
limit: int = 25,
cursor: Optional[str] = None,
type: Optional[str] = None
) -> dict:
"""List payment methods"""
params = {'limit': limit}
if cursor:
params['cursor'] = cursor
if type:
params['type'] = type
return self.client._request('GET', '/payment-methods', params=params)
def delete(self, payment_method_id: str) -> None:
"""Delete a payment method"""
self.client._request('DELETE', f'/payment-methods/{payment_method_id}')

133
fetcherpay/api/payments.py Normal file
View File

@@ -0,0 +1,133 @@
"""
Payments API
"""
from typing import Optional, Dict, Any, List
class PaymentsAPI:
"""Payments API client"""
def __init__(self, client):
self.client = client
def create(
self,
amount: int,
source: Dict[str, str],
destination: Dict[str, str],
currency: str = 'USD',
rail: str = 'auto',
description: Optional[str] = None,
metadata: Optional[Dict[str, Any]] = None,
idempotency_key: Optional[str] = None
) -> Dict[str, Any]:
"""
Create a new payment
Args:
amount: Amount in cents (e.g., 10000 for $100.00)
source: Payment source with payment_method_id
destination: Payment destination with payment_method_id
currency: Currency code (default: USD)
rail: Payment rail (auto, ach, rtp, card, crypto)
description: Optional payment description
metadata: Optional metadata dictionary
idempotency_key: Optional idempotency key
Returns:
Payment object
"""
data = {
'amount': amount,
'currency': currency,
'rail': rail,
'source': source,
'destination': destination,
}
if description:
data['description'] = description
if metadata:
data['metadata'] = metadata
return self.client._request(
'POST',
'/payments',
json=data,
idempotency_key=idempotency_key
)
def retrieve(self, payment_id: str) -> Dict[str, Any]:
"""Retrieve a payment by ID"""
return self.client._request('GET', f'/payments/{payment_id}')
def list(
self,
limit: int = 25,
cursor: Optional[str] = None,
status: Optional[str] = None,
rail: Optional[str] = None
) -> Dict[str, Any]:
"""
List payments
Args:
limit: Number of results (1-100)
cursor: Pagination cursor
status: Filter by status
rail: Filter by rail
"""
params = {'limit': limit}
if cursor:
params['cursor'] = cursor
if status:
params['status'] = status
if rail:
params['rail'] = rail
return self.client._request('GET', '/payments', params=params)
def cancel(
self,
payment_id: str,
reason: Optional[str] = None,
idempotency_key: Optional[str] = None
) -> Dict[str, Any]:
"""Cancel a pending payment"""
data = {'reason': reason} if reason else {}
return self.client._request(
'POST',
f'/payments/{payment_id}/cancel',
json=data,
idempotency_key=idempotency_key
)
def refund(
self,
payment_id: str,
amount: Optional[int] = None,
reason: Optional[str] = None,
idempotency_key: Optional[str] = None
) -> Dict[str, Any]:
"""
Refund a settled payment
Args:
payment_id: Payment ID to refund
amount: Amount to refund (omit for full refund)
reason: Refund reason
idempotency_key: Idempotency key
"""
data = {}
if amount:
data['amount'] = amount
if reason:
data['reason'] = reason
return self.client._request(
'POST',
f'/payments/{payment_id}/refund',
json=data,
idempotency_key=idempotency_key
)

View File

@@ -0,0 +1,80 @@
"""
Webhooks API
"""
from typing import Optional, List, Dict, Any
class WebhooksAPI:
"""Webhooks API client"""
def __init__(self, client):
self.client = client
def create(
self,
url: str,
events: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
idempotency_key: Optional[str] = None
) -> dict:
"""
Create a webhook endpoint
Args:
url: Endpoint URL
events: List of events to subscribe to
metadata: Optional metadata
idempotency_key: Idempotency key
"""
data = {'url': url}
if events:
data['events'] = events
if metadata:
data['metadata'] = metadata
return self.client._request(
'POST',
'/webhooks',
json=data,
idempotency_key=idempotency_key
)
def retrieve(self, webhook_id: str) -> dict:
"""Retrieve a webhook endpoint"""
return self.client._request('GET', f'/webhooks/{webhook_id}')
def list(
self,
limit: int = 25,
cursor: Optional[str] = None
) -> dict:
"""List webhook endpoints"""
params = {'limit': limit}
if cursor:
params['cursor'] = cursor
return self.client._request('GET', '/webhooks', params=params)
def update(
self,
webhook_id: str,
url: Optional[str] = None,
events: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None
) -> dict:
"""Update a webhook endpoint"""
data = {}
if url:
data['url'] = url
if events:
data['events'] = events
if metadata:
data['metadata'] = metadata
return self.client._request('PUT', f'/webhooks/{webhook_id}', json=data)
def delete(self, webhook_id: str) -> None:
"""Delete a webhook endpoint"""
self.client._request('DELETE', f'/webhooks/{webhook_id}')

153
fetcherpay/client.py Normal file
View File

@@ -0,0 +1,153 @@
"""
FetcherPay API Client
"""
import hmac
import hashlib
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib.parse import urljoin
from .exceptions import (
FetcherPayError,
AuthenticationError,
ValidationError,
NotFoundError,
IdempotencyError,
)
from .api.payments import PaymentsAPI
from .api.ledger import LedgerAPI
from .api.payment_methods import PaymentMethodsAPI
from .api.webhooks import WebhooksAPI
class FetcherPay:
"""
Main FetcherPay client
Args:
api_key: Your FetcherPay API key
environment: 'sandbox' or 'production'
base_url: Optional custom base URL
timeout: Request timeout in seconds (default: 30)
"""
def __init__(
self,
api_key: str,
environment: str = 'sandbox',
base_url: Optional[str] = None,
timeout: int = 30
):
self.api_key = api_key
self.environment = environment
self.timeout = timeout
if base_url:
self.base_url = base_url
elif environment == 'production':
self.base_url = 'https://api.fetcherpay.com/v1'
else:
self.base_url = 'https://sandbox.fetcherpay.com/v1'
# Create session with retries
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json',
})
# Initialize API modules
self.payments = PaymentsAPI(self)
self.ledger = LedgerAPI(self)
self.payment_methods = PaymentMethodsAPI(self)
self.webhooks = WebhooksAPI(self)
def _request(
self,
method: str,
path: str,
params: Optional[dict] = None,
json: Optional[dict] = None,
headers: Optional[dict] = None,
idempotency_key: Optional[str] = None
) -> dict:
"""Make an HTTP request"""
url = urljoin(self.base_url, path)
request_headers = {}
if headers:
request_headers.update(headers)
if idempotency_key:
request_headers['Idempotency-Key'] = idempotency_key
try:
response = self.session.request(
method=method,
url=url,
params=params,
json=json,
headers=request_headers,
timeout=self.timeout
)
# Handle errors
if response.status_code >= 400:
self._handle_error(response)
return response.json() if response.content else {}
except requests.exceptions.Timeout:
raise FetcherPayError('Request timeout', 'timeout_error')
except requests.exceptions.ConnectionError:
raise FetcherPayError('Connection error', 'connection_error')
def _handle_error(self, response):
"""Handle API errors"""
try:
data = response.json()
error = data.get('error', {})
except:
error = {}
message = error.get('message', 'An error occurred')
error_type = error.get('type', 'api_error')
param = error.get('param')
code = error.get('code')
if response.status_code == 401:
raise AuthenticationError(message)
elif response.status_code == 404:
raise NotFoundError(message)
elif response.status_code == 422:
raise ValidationError(message, param)
elif response.status_code == 409:
raise IdempotencyError(message)
else:
raise FetcherPayError(message, error_type, response.status_code, param, code)
def verify_webhook_signature(
self,
payload: str,
signature: str,
secret: str
) -> bool:
"""
Verify a webhook signature
Args:
payload: The raw webhook payload string
signature: The signature from X-FetcherPay-Signature header
secret: Your webhook secret
Returns:
True if signature is valid
"""
expected = hmac.new(
secret.encode('utf-8'),
payload.encode('utf-8'),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(signature, expected)

49
fetcherpay/exceptions.py Normal file
View File

@@ -0,0 +1,49 @@
"""
FetcherPay SDK Exceptions
"""
class FetcherPayError(Exception):
"""Base FetcherPay error"""
def __init__(
self,
message: str,
type: str = 'api_error',
status_code: int = None,
param: str = None,
code: str = None
):
super().__init__(message)
self.type = type
self.status_code = status_code
self.param = param
self.code = code
class AuthenticationError(FetcherPayError):
"""Authentication failed"""
def __init__(self, message: str):
super().__init__(message, 'authentication_error', 401)
class ValidationError(FetcherPayError):
"""Validation failed"""
def __init__(self, message: str, param: str = None):
super().__init__(message, 'validation_error', 422, param)
class NotFoundError(FetcherPayError):
"""Resource not found"""
def __init__(self, message: str):
super().__init__(message, 'not_found', 404)
class IdempotencyError(FetcherPayError):
"""Idempotency key conflict"""
def __init__(self, message: str):
super().__init__(message, 'idempotency_error', 409)

153
fetcherpay/types.py Normal file
View File

@@ -0,0 +1,153 @@
"""
FetcherPay SDK Types
"""
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from enum import Enum
class PaymentStatus(str, Enum):
PENDING = 'pending'
AUTHORIZED = 'authorized'
PROCESSING = 'processing'
SETTLED = 'settled'
FAILED = 'failed'
CANCELLED = 'cancelled'
REFUNDED = 'refunded'
PARTIALLY_REFUNDED = 'partially_refunded'
class Rail(str, Enum):
AUTO = 'auto'
ACH = 'ach'
RTP = 'rtp'
CARD = 'card'
CRYPTO = 'crypto'
class AccountType(str, Enum):
ASSET = 'asset'
LIABILITY = 'liability'
REVENUE = 'revenue'
EXPENSE = 'expense'
EQUITY = 'equity'
@dataclass
class Fee:
amount: int
rate: str
@dataclass
class TimelineEvent:
status: str
timestamp: str
detail: str
@dataclass
class Refund:
id: str
payment_id: str
amount: int
reason: Optional[str]
status: str
created_at: str
@dataclass
class Payment:
id: str
object: str
status: PaymentStatus
amount: int
currency: str
rail: str
rail_selected: str
description: Optional[str]
source: Dict[str, Any]
destination: Dict[str, Any]
fee: Optional[Fee]
timeline: List[TimelineEvent]
ledger_entry_ids: List[str]
refunds: List[Refund]
idempotency_key: Optional[str]
metadata: Dict[str, Any]
created_at: str
updated_at: str
@dataclass
class PaymentMethod:
id: str
object: str
type: str
status: str
bank_account: Optional[Dict[str, Any]]
card: Optional[Dict[str, Any]]
usdc_wallet: Optional[Dict[str, Any]]
metadata: Dict[str, Any]
created_at: str
@dataclass
class LedgerBalance:
pending: int
posted: int
available: int
@dataclass
class LedgerAccount:
id: str
object: str
name: str
type: AccountType
currency: str
balance: LedgerBalance
metadata: Dict[str, Any]
created_at: str
updated_at: str
@dataclass
class LedgerEntry:
id: str
object: str
journal_id: str
account_id: str
payment_id: Optional[str]
entry_type: str
amount: int
currency: str
status: str
description: Optional[str]
metadata: Dict[str, Any]
created_at: str
@dataclass
class WebhookEndpoint:
id: str
object: str
url: str
events: List[str]
status: str
secret: str
metadata: Dict[str, Any]
created_at: str
@dataclass
class ListResponse:
data: List[Any]
has_more: bool
next_cursor: Optional[str]
# Request types
CreatePaymentRequest = Dict[str, Any]
CreatePaymentMethodRequest = Dict[str, Any]
CreateWebhookRequest = Dict[str, Any]

41
setup.py Normal file
View File

@@ -0,0 +1,41 @@
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="fetcherpay",
version="1.0.0",
author="FetcherPay",
author_email="support@fetcherpay.com",
description="FetcherPay Python SDK",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/fetcherpay/fetcherpay-python",
packages=find_packages(),
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
],
python_requires=">=3.8",
install_requires=[
"requests>=2.25.0",
"typing-extensions>=4.0.0",
],
extras_require={
"dev": [
"pytest>=6.0",
"pytest-cov>=2.0",
"black>=21.0",
"isort>=5.0",
"mypy>=0.910",
]
}
)