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

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

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)

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 (product_id is mandatory).

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

Frequency Units

Available units for frequency and trial_period:

seconds
minutes
hours
days
weeks
months

Get Subscription

subscription = client.subscriptions.get("subscription-uuid")
print(subscription.subscription_status, subscription.next_billing_date)

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")

# 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_url argument on one_time_payments.create_session() and subscriptions.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 calls subscriptions.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}")

        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, 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"{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.

Name		Name	Last commit message	Last commit date
Latest commit History 15 Commits
examples		examples
qbitflow		qbitflow
tests		tests
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
COMPLIANCE.md		COMPLIANCE.md
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
README.md		README.md
SECURITY.md		SECURITY.md
TRADEMARKS.md		TRADEMARKS.md
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt

Folders and files

Latest commit

History

Repository files navigation