QBitFlow Python SDK
Official Python SDK for QBitFlow - a comprehensive cryptocurrency payment processing platform that enables seamless integration of crypto payments, recurring subscriptions, and usage-based billing into your applications.
Features
- 🔐 Type-Safe: Full type hints for better IDE support
- 🚀 Easy to Use: Simple, intuitive API design
- 🔄 Automatic Retries: Built-in retry logic for failed requests
- ⚡ Real-time Updates: WebSocket support for transaction status monitoring
- 🧪 Well Tested: Comprehensive test coverage
- 📚 Great Documentation: Detailed docs with examples
- 🔌 Webhook Support: Handle payment notifications easily
- 💳 One-Time Payments: Accept cryptocurrency payments with ease
- 🔄 Recurring Subscriptions: Automated recurring billing in cryptocurrency
- 👥 Customer Management: Create and manage customer profiles
- 🛍️ Product Management: Organize your products and pricing
- 📈 Transaction Tracking: Real-time transaction status updates
- 💸 Refund Tracking: Monitor refund status
- 📊 Accounting Export: Export transaction data as JSON or CSV
- 🔑 Account Claims: Invite unclaimed users to set up their wallets
Table of Contents
- Features
- Installation
- Quick Start
- Configuration
- One-Time Payments
- Subscriptions
- Refunds
- Accounting Export
- Account Claims
- Transaction Status
- Customer Management
- Product Management
- User Management
- API Key Management
- Webhook Handling
- Error Handling
- API Reference
- License
Installation
Install the SDK using pip:
pip install qbitflow
Or install from source:
git clone https://github.com/qbitflow/qbitflow-python-sdk.git
cd qbitflow-python-sdk
pip install -e .
Quick Start
1. Get Your API Key
Sign up at QBitFlow and obtain your API key from the dashboard.
2. Initialize the Client
from qbitflow import QBitFlow
client = QBitFlow(api_key="your_api_key_here")
3. Create a One-Time Payment
response = client.one_time_payments.create_session(
product_id=1,
customer_uuid="customer-uuid",
success_url="https://your-domain.com/success",
cancel_url="https://your-domain.com/cancel"
)
print(f"Payment link: {response.link}")
# Send this link to your customer
4. Create a Recurring Subscription
from qbitflow import Duration
response = client.subscriptions.create_session(
product_id=1,
frequency=Duration(value=1, unit="months"),
trial_period=Duration(value=7, unit="days"), # Optional 7-day trial
customer_uuid="customer-uuid"
)
print(f"Subscription link: {response.link}")
5. Check Transaction Status
from qbitflow.dto.transaction.status import TransactionType, TransactionStatusValue
status = client.transaction_status.get(
transaction_uuid="transaction-uuid",
transaction_type=TransactionType.ONE_TIME_PAYMENT
)
if status.status == TransactionStatusValue.COMPLETED:
print(f"Payment completed! Transaction hash: {status.tx_hash}")
elif status.status == TransactionStatusValue.FAILED:
print(f"Payment failed: {status.message}")
Configuration
Configuration Options
| Option | Type | Default | Description |
|---|---|---|---|
api_key |
string | (required) | Your QBitFlow API key |
base_url |
string | https://api.qbitflow.app |
API base URL |
timeout |
int | 30 |
Request timeout in seconds |
max_retries |
int | 3 |
Number of retry attempts for failed requests |
One-Time Payments
Create a Payment Session
Provide either a product_id for an existing product, or product_name + description + price for an ad-hoc charge:
# From an existing product
response = client.one_time_payments.create_session(
product_id=1,
customer_uuid="customer-uuid"
)
# Ad-hoc payment
response = client.one_time_payments.create_session(
product_name="Custom Product",
description="Product description",
price=99.99, # USD
customer_uuid="customer-uuid"
)
print(response.uuid) # Session UUID
print(response.link) # Payment link for customer
Using your own references
Instead of storing QBitFlow's internal UUIDs, pass your own identifiers. Set reference to your
order/invoice ID, and use product_reference / customer_reference to select an existing product
or customer by your own reference:
response = client.one_time_payments.create_session(
reference="order-1234", # your own transaction reference
product_reference="PROD-PREMIUM", # use a product by your reference (instead of product_id)
customer_reference="user-42", # use a customer by your reference (instead of customer_uuid)
)
The reference is echoed back on the resulting Payment and in webhook payloads, and you can look
the payment up later with get_by_reference(). If no customer matches
customer_reference, one is created during checkout.
With Redirect URLs
response = client.one_time_payments.create_session(
product_id=1,
success_url="https://your-domain.com/success?uuid={{UUID}}&type={{TRANSACTION_TYPE}}",
cancel_url="https://your-domain.com/cancel",
customer_uuid="customer-uuid",
)
Available Placeholders:
{{UUID}}: The session UUID{{TRANSACTION_TYPE}}: The transaction type (e.g., "payment", "subscription")
Get Payment Session
Returns a OneTimePaymentSession with the base transaction fields.
session = client.one_time_payments.get_session("session-uuid")
print(session.product_name, session.price)
Get Completed Payment
payment = client.one_time_payments.get("payment-uuid")
print(payment.transaction_hash, payment.amount)
Get Payment by Reference
Look a payment up by the reference you assigned when creating the session — no need to store
QBitFlow's UUID:
payment = client.one_time_payments.get_by_reference("order-1234")
print(payment.uuid, payment.amount)
List All Payments
page = client.one_time_payments.get_all(limit=10)
print(page.items) # List of Payment objects
print(page.has_more()) # Whether there are more pages
print(page.next_cursor)
if page.has_more():
next_page = client.one_time_payments.get_all(limit=10, cursor=page.next_cursor)
List Combined Payments
Get all payments from both one-time and subscription sources in a single paginated list:
page = client.one_time_payments.get_all_combined(limit=20)
for item in page.items:
print(item.source) # "payment" or "subscription_history"
print(item.amount)
if item.subscription_uuid:
print(f"Subscription: {item.subscription_uuid}")
Get Customer for Transaction
customer = client.one_time_payments.get_customer_for_transaction("transaction-uuid")
print(f"{customer.name} {customer.last_name} — {customer.email}")
Subscriptions
Subscriptions require an existing product — provide either product_id or product_reference.
Create a Subscription
from qbitflow import Duration
response = client.subscriptions.create_session(
product_id=1,
frequency=Duration(value=1, unit="months"),
trial_period=Duration(value=7, unit="days"), # Optional
min_periods=3, # Optional: minimum billing periods
customer_uuid="customer-uuid",
)
print(response.link) # Send to customer
Like one-time payments, subscription sessions accept your own reference, product_reference,
and customer_reference instead of QBitFlow's internal IDs:
response = client.subscriptions.create_session(
reference="sub-1234", # your own subscription reference
product_reference="PLAN-PRO", # select a product by your reference
customer_reference="user-42", # select a customer by your reference
frequency=Duration(value=1, unit="months"),
)
Frequency Units
Available units for frequency and trial_period:
secondsminuteshoursdaysweeksmonths
Get Subscription
subscription = client.subscriptions.get("subscription-uuid")
print(subscription.subscription_status, subscription.next_billing_date)
Get Subscription by Reference
Look a subscription up by the reference you assigned when creating the session:
subscription = client.subscriptions.get_by_reference("sub-1234")
print(subscription.uuid, subscription.subscription_status)
Tracking status changes: You no longer need to poll
get()on a schedule (e.g. a cron job) to detect subscription lifecycle changes. Enable the Subscription status webhook in your QBitFlow dashboard settings and you will receive a notification on every status transition. See Subscription Status Webhook.
Get Payment History
history = client.subscriptions.get_payment_history("subscription-uuid")
for record in history:
print(record.uuid, record.amount, record.created_at)
Force Cancel
Force cancel a subscription immediately, bypassing the normal user-signed cancellation flow:
response = client.subscriptions.force_cancel("subscription-uuid")
print(response.message)
Execute Test Billing Cycle
Test Mode Only: Manually trigger a billing cycle to test webhook behaviour.
result = client.subscriptions.execute_test_billing_cycle("subscription-uuid")
print("Status link:", result.status_link)
Refunds
List Active Refunds
refunds = client.refunds.get_all()
for refund in refunds:
print(f"{refund.uuid}: {refund.status.value} — {refund.reason}")
List Inactive Refunds
Returns processed (approved/refused/failed) refunds with pagination:
page = client.refunds.get_all_inactive(limit=10)
for refund in page.items:
print(f"{refund.uuid}: {refund.status.value}")
if page.has_more():
next_page = client.refunds.get_all_inactive(limit=10, cursor=page.next_cursor)
Get Refund by Transaction
Public endpoint — no authentication required:
refund = client.refunds.get_by_transaction("transaction-uuid")
print(refund.status.value, refund.tx_hash)
Accounting Export
Export transaction data for a date range. Dates must be in YYYY-MM-DD format.
# JSON export — returns List[AccountingEvent]
events = client.accounting.export("2025-01-01", "2025-12-31", "json")
for event in events:
print(f"{event.payment_id} | {event.type} | ${event.gross_amount_usd}")
# CSV export — returns raw CSV string
csv_data = client.accounting.export("2025-01-01", "2025-12-31", "csv")
with open("accounting_2025.csv", "w") as f:
f.write(csv_data)
AccountingEvent fields include: payment_id, type, tx_time_utc, receipt_url, product_id, customer_uuid, chain, tx_hash, token_symbol, gross_amount_usd, platform_fee_usd, organization_fee_usd, net_amount_usd, and more.
Account Claims
QBitFlow lets organizations create users whose payments are held by the organization. When the organization is ready, they create a claim request — a one-time link that the user follows to set up their wallet and receive their accumulated funds.
Get a Claim Request
Retrieve the existing claim link for a user without creating a new one:
result = client.claim.get_request(user_id=42)
print(f"Claim link: {result.link}")
Create a Claim Request
Create a new claim request (or return the existing one) for a user:
result = client.claim.create_request(user_id=42)
print(f"Claim link: {result.link}")
# Send result.link to the user by email
Get Claim Funds
List pending fund transfers owed to users who have already claimed their accounts:
funds = client.claim.get_funds()
for fund in funds:
if not fund.funded:
print(f"Pending: ${fund.total_amount_owed} → user {fund.user_id}")
Trigger Test Claim Funds
Test Mode Only: Manually compute ledger totals for a user without waiting for the hourly job:
client.claim.trigger_test_claim_funds(user_id=42)
Transaction Status
Check Status
from qbitflow.dto.transaction.status import TransactionType
status = client.transaction_status.get(
"transaction-uuid",
TransactionType.ONE_TIME_PAYMENT
)
print(status.status) # TransactionStatusValue enum
print(status.tx_hash) # Blockchain transaction hash
Transaction Types
class TransactionType:
ONE_TIME_PAYMENT = 'payment'
CREATE_SUBSCRIPTION = 'createSubscription'
CANCEL_SUBSCRIPTION = 'cancelSubscription'
EXECUTE_SUBSCRIPTION_PAYMENT = 'executeSubscription'
INCREASE_ALLOWANCE = 'increaseAllowance'
Status Values
class TransactionStatusValue:
CREATED = 'created'
WAITING_CONFIRMATION = 'waitingConfirmation'
PENDING = 'pending'
COMPLETED = 'completed'
FAILED = 'failed'
CANCELLED = 'cancelled'
EXPIRED = 'expired'
Customer Management
from qbitflow.dto.customer import CreateCustomerDto, UpdateCustomerDto
# Create
customer = client.customers.create(CreateCustomerDto(
name="John", last_name="Doe",
email="john@example.com",
phone_number="+1234567890",
reference="CRM-12345"
))
# Get
customer = client.customers.get("customer-uuid")
customer = client.customers.get_by_email("john@example.com")
customer = client.customers.get_by_reference("CRM-12345")
# List (paginated)
page = client.customers.get_all(limit=10)
# Update
updated = client.customers.update("customer-uuid", UpdateCustomerDto(
name="John", last_name="Doe", email="john.doe@example.com"
))
# Delete
client.customers.delete("customer-uuid")
Product Management
from qbitflow.dto.product import CreateProductDto, UpdateProductDto
# Create
product = client.products.create(CreateProductDto(
name="Premium Subscription",
description="Access to all premium features",
price=29.99,
reference="PROD-PREMIUM"
))
# Get
product = client.products.get(1)
product = client.products.get_by_reference("PROD-PREMIUM")
# List all
products = client.products.get_all()
# Update
updated = client.products.update(1, UpdateProductDto(
name="Premium Plus",
description="Enhanced premium features",
price=39.99
))
# Delete
client.products.delete(1)
User Management
from qbitflow.dto.user import CreateUserDto, UpdateUserDto
# Create (admin only)
user = client.users.create(CreateUserDto(
name="Alice",
last_name="Smith",
email="alice@example.com",
role="user", # "user" or "admin"
organization_fee_bps=100 # optional, 1% fee
))
# Get current user (identified by API key)
me = client.users.get()
# Get by ID or list all (admin only)
user = client.users.get_by_id(42)
users = client.users.get_all()
# Update
updated = client.users.update(user.id, UpdateUserDto(
name="Alicia",
last_name="Smith",
email=user.email
))
# Delete (admin only)
client.users.delete(user.id)
API Key Management
from qbitflow.dto.api_key import CreateApiKeyDto
# Create
resp = client.api_keys.create(CreateApiKeyDto(
name="Production Key",
user_id=user_id,
test=False
))
print(f"Key (only shown once): {resp.key}")
# List API keys for the current user
keys = client.api_keys.get_all()
# List API keys for a specific user (admin only)
keys = client.api_keys.get_for_user(user_id)
# Delete
client.api_keys.delete(key_id)
Webhook Handling
Configuring Webhooks
Webhook URLs are no longer set per session. Instead, configure them once in your QBitFlow dashboard settings, and they apply consistently to every transaction:
- Transaction webhook — receives notifications when a payment or subscription
session changes status (e.g. completed, failed). Payload:
SessionWebhookResponse. - Subscription status webhook — receives notifications on every subscription
lifecycle transition (e.g.
trial → active,active → past_due,active → cancelled). Payload:SubscriptionStatusTransitionWebhook.
Migration note: Previous versions accepted a
webhook_urlargument onone_time_payments.create_session()andsubscriptions.create_session(). That parameter has been removed — set the Transaction webhook in the dashboard instead. Likewise, the Subscription status webhook replaces the old pattern of running a cron job that periodically callssubscriptions.get()to detect status changes.
A complete, runnable FastAPI example handling both webhook types lives in
test-internal/main.py.
Test Webhook Reachability
The dashboard's Test webhook action lets you confirm your endpoint is reachable
before going live. It sends a request with a fake payload that will not parse like a
real webhook — so your handler must short-circuit it. The request carries the webhook ID
in the X-Webhook-ID header; when that value equals TEST_WEBHOOK_ID, return HTTP 200
immediately and skip normal payload processing:
from qbitflow.requests.webhook import TEST_WEBHOOK_ID
# ...inside your handler, after verifying the signature:
if x_webhook_id == TEST_WEBHOOK_ID:
return {"status": "received", "message": "Test webhook acknowledged"}
Perform this check after signature verification but before parsing the payload — otherwise the fake payload will fail validation and the reachability check will report an error. Both examples below include this guard.
Transaction Webhook
Handles payment and subscription session status changes. Always verify the signature before trusting the payload:
from typing import Annotated
from fastapi import FastAPI, Request, Header, HTTPException
from qbitflow import QBitFlow
from qbitflow.dto.transaction.session import SessionWebhookResponse
from qbitflow.dto.transaction.status import TransactionStatusValue
from qbitflow.requests.webhook import TEST_WEBHOOK_ID
app = FastAPI()
client = QBitFlow(api_key="your_api_key")
@app.post("/webhook")
async def handle_webhook(
request: Request,
x_webhook_id: Annotated[str, Header()],
x_webhook_signature_256: Annotated[str, Header()],
x_webhook_timestamp: Annotated[str, Header()]
):
body = await request.body()
if not client.webhooks.verify(
payload=body,
signature=x_webhook_signature_256,
timestamp=x_webhook_timestamp
):
# Returning a >= 400 status causes QBitFlow to retry the webhook
raise HTTPException(status_code=401, detail="Invalid webhook signature")
# Reachability test from the dashboard — acknowledge and skip processing
if x_webhook_id == TEST_WEBHOOK_ID:
return {"status": "received", "message": "Test webhook acknowledged"}
event = SessionWebhookResponse.model_validate_json(body)
# event.session is automatically resolved to the correct session type:
# OneTimePaymentSession, SubscriptionSession, or PaygSubscriptionSession
if event.status.status == TransactionStatusValue.COMPLETED:
print(f"Payment completed: {event.session.product_name}")
print(f"Customer: {event.session.customer_uuid}")
print(f"Amount: ${event.session.price}")
# `reference` echoes back the value you set when creating the session, so you can
# match the transaction to your own order/invoice without storing our UUID.
print(f"Your reference: {event.session.reference}")
from qbitflow.dto.transaction.session import SubscriptionSession
if isinstance(event.session, SubscriptionSession):
print(f"Frequency: {event.session.frequency}s")
elif event.status.status == TransactionStatusValue.FAILED:
print(f"Payment failed: {event.status.message}")
return {"received": True}
Subscription Status Webhook
Once the Subscription status webhook is enabled in the dashboard, QBitFlow POSTs a
SubscriptionStatusTransitionWebhook payload whenever a subscription changes status —
no polling required. The payload carries subscription_uuid, subscription_reference
(your own reference, if set), previous_status, current_status, and updated_at:
from typing import Annotated
from fastapi import FastAPI, Request, Header, HTTPException
from qbitflow import QBitFlow
from qbitflow.dto.transaction.subscription import (
SubscriptionStatusTransitionWebhook,
SubscriptionStatus,
)
from qbitflow.requests.webhook import TEST_WEBHOOK_ID
app = FastAPI()
client = QBitFlow(api_key="your_api_key")
@app.post("/subscription-webhook")
async def handle_subscription_webhook(
request: Request,
x_webhook_id: Annotated[str, Header()],
x_webhook_signature_256: Annotated[str, Header()],
x_webhook_timestamp: Annotated[str, Header()]
):
body = await request.body()
if not client.webhooks.verify(
payload=body,
signature=x_webhook_signature_256,
timestamp=x_webhook_timestamp
):
raise HTTPException(status_code=401, detail="Invalid webhook signature")
# Reachability test from the dashboard — acknowledge and skip processing
if x_webhook_id == TEST_WEBHOOK_ID:
return {"status": "received", "message": "Test webhook acknowledged"}
event = SubscriptionStatusTransitionWebhook.model_validate_json(body)
print(f"Subscription {event.subscription_uuid} "
f"(ref: {event.subscription_reference}): "
f"{event.previous_status.value} -> {event.current_status.value}")
# React to the lifecycle transition — e.g. revoke access on cancellation
if event.current_status == SubscriptionStatus.CANCELLED:
print(f"Revoking access for {event.subscription_uuid}")
elif event.current_status == SubscriptionStatus.PAST_DUE:
print(f"Payment failed — notifying customer for {event.subscription_uuid}")
return {"received": True}
SubscriptionStatus values: active, cancelled, past_due, low_on_funds,
pending, trial, trial_expired.
Error Handling
from qbitflow.exceptions import (
QBitFlowError,
AuthenticationError,
NotFoundException,
ValidationError,
RateLimitError,
NetworkError,
APIError
)
try:
payment = client.one_time_payments.get("non-existent-uuid")
except AuthenticationError:
print("Invalid API key or authentication failed")
except NotFoundException as e:
print(f"Payment not found: {e.message}")
except ValidationError as e:
print(f"Validation error: {e.message}")
except RateLimitError as e:
print(f"Rate limit exceeded. Retry after: {e.response.get('retry_after')}")
except NetworkError as e:
print(f"Network error: {e.message}")
except APIError as e:
print(f"API error: {e.message} (status: {e.status_code})")
except QBitFlowError as e:
print(f"SDK error: {e.message}")
API Reference
QBitFlow
Constructor
QBitFlow(api_key: str, timeout: Optional[int] = None, max_retries: Optional[int] = None)
Properties
| Property | Type | Description |
|---|---|---|
customers |
CustomerRequests |
Customer CRUD operations |
products |
ProductRequests |
Product CRUD operations |
users |
UserRequests |
User management operations |
api_keys |
ApiKeyRequests |
API key management |
one_time_payments |
PaymentRequests |
One-time payment sessions and history |
subscriptions |
SubscriptionRequests |
Recurring subscription management |
refunds |
RefundRequests |
Refund retrieval |
accounting |
AccountingRequests |
Accounting data export (JSON/CSV) |
claim |
ClaimRequests |
Account claim and fund transfer |
transaction_status |
TransactionStatusRequests |
Transaction status polling |
webhooks |
WebhookRequests |
Webhook signature verification |
Testing
export QBITFLOW_API_KEY="your_test_api_key"
export QBITFLOW_BASE_URL="http://localhost:3001" # Optional local server
pytest tests/ -v
pytest tests/ --cov=qbitflow --cov-report=html
License
This project is licensed under the MPL-2.0 License - see the LICENSE file for details.
Support
Changelog
See CHANGELOG.md for a list of changes in each version.
Security
For security issues, please email security@qbitflow.app instead of using the issue tracker.
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file qbitflow-1.3.0.tar.gz.
File metadata
- Download URL: qbitflow-1.3.0.tar.gz
- Upload date:
- Size: 59.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.2.0 CPython/3.14.6
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
801b021461330a8108dccc62aafc9469bfa3703437e4096ed845a8b5fc109868
|
|
| MD5 |
acbdd0dd5968a8e3ab7bff68222f68c2
|
|
| BLAKE2b-256 |
7a455e8d9d5f0e0986f5e3cbad9b69eaac873b1c1a5c5049e99b9c0fb6f704b9
|
File details
Details for the file qbitflow-1.3.0-py3-none-any.whl.
File metadata
- Download URL: qbitflow-1.3.0-py3-none-any.whl
- Upload date:
- Size: 62.7 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.2.0 CPython/3.14.6
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
3846702ad9ea52e184d2a48f80a5f256468e51bf99ed27b5e3b85fee21216a36
|
|
| MD5 |
b121d611a894ae7192759fcc3cd5459b
|
|
| BLAKE2b-256 |
cce430313fe0b6614891eddddd9b52b977d7521e16a0f0c4a1243893461e60cc
|