qbitflow 1.2.0

Official Python SDK for QBitFlow - Next Generation Crypto Payment Processing

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
  • 1. Get Your API Key
  • 2. Initialize the Client
  • 3. Create a One-Time Payment
  • 4. Create a Recurring Subscription
  • 5. Check Transaction Status
• Configuration
• One-Time Payments
  • Create a Payment Session
  • With Redirect URLs
  • Get Payment Session
  • Get Completed Payment
  • List All Payments
  • List Combined Payments
  • Get Customer for Transaction
• Subscriptions
  • Create a Subscription
  • Frequency Units
  • Get Subscription
  • Get Payment History
  • Force Cancel
  • Execute Test Billing Cycle
• Refunds
  • List Active Refunds
  • List Inactive Refunds
  • Get Refund by Transaction
• Accounting Export
• Account Claims
  • Get a Claim Request
  • Create a Claim Request
  • Get Claim Funds
  • Trigger Test Claim Funds
• Transaction Status
  • Check Status
  • Transaction Types
  • Status Values
• Customer Management
• Product Management
• User Management
• API Key Management
• Webhook Handling
  • FastAPI Example
• 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",
    webhook_url="https://your-domain.com/webhook",
    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",
    webhook_url="https://your-domain.com/webhook"
)

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",
    webhook_url="https://your-domain.com/webhook",
)

# 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",
    webhook_url="https://your-domain.com/webhook",
)

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
    webhook_url="https://your-domain.com/webhook",
    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)

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="[e-mail]",
    phone_number="[phone]",
    reference="CRM-12345"
))

# Get
customer = client.customers.get("customer-uuid")
customer = client.customers.get_by_email("[e-mail]")

# List (paginated)
page = client.customers.get_all(limit=10)

# Update
updated = client.customers.update("customer-uuid", UpdateCustomerDto(
    name="John", last_name="Doe", email="[e-mail]"
))

# 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="[e-mail]",
    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

FastAPI Example

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

app = FastAPI()
client = QBitFlow(api_key="your_api_key")

@app.post("/webhook")
async def handle_webhook(
    request: Request,
    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")

    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}

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

• 📖 Documentation
• 📧 Email Support
• 🐛 Issue Tracker

Changelog

See CHANGELOG.md for a list of changes in each version.

Security

For security issues, please email [e-mail] instead of using the issue tracker.