loyverse-sdk 0.1.0

Asynchronous Python SDK for the Loyverse API

Loyverse SDK

Asynchronous Python SDK for the Loyverse API, a point-of-sale (POS) system for managing business transactions, inventory, and customer data.

Overview

The SDK provides:

• Async/await interface using httpx for non-blocking API calls
• Type-safe request/response models using Pydantic
• Automatic pagination with cursor-based iteration via iter_all()
• Full CRUD operations for supported endpoints
• 16 endpoints: categories, customers, discounts, devices, employees, inventory, items, merchant, modifiers, receipts, shifts, stores, suppliers, taxes, webhooks, variants

Codebase Structure

src/loyverse_sdk/ contains:

• client.py - Main LoyverseClient class with endpoint access
• endpoints/ - Endpoint classes using mixin pattern for CRUD operations
• models/ - Pydantic models for request/response validation
• auth.py - Token-based authentication
• core/ - Configuration, logging, and utilities

Installation

uv pip install git+https://github.com/dagsdags212/loyverse_sdk.git

Setup

Set your API token as an environment variable:

export LOYVERSE_API_TOKEN=your_api_token

Or create a .env file in your project root:

LOYVERSE_API_TOKEN=your_api_token

Quick Start

import asyncio
from loyverse_sdk import LoyverseClient

async def main():
    # Create client (automatically loads token from environment)
    client = LoyverseClient()

    # List customers
    response = await client.customers.list(limit=10)
    print(f"Found {len(response.items)} customers")

    # Close connection
    await client.close()

asyncio.run(main())

Usage Examples

Customers Endpoint

The customers endpoint manages customer data from your POS system.

List customers with pagination:

# Get first page of customers
response = await client.customers.list(limit=50)

for customer in response.items:
    print(f"{customer.name} - {customer.email}")
    print(f"  Total visits: {customer.total_visits}")
    print(f"  Total spent: ${customer.total_spent}")

# Get next page using cursor
if response.cursor:
    next_page = await client.customers.list(limit=50, cursor=response.cursor)

Retrieve a single customer:

customer = await client.customers.retrieve(id="customer-uuid-here")
print(customer.name)
print(customer.phone_number)
print(customer.address)

Create a new customer:

new_customer = await client.customers.create({
    "name": "Jane Smith",
    "email": "[e-mail]",
    "phone_number": "[phone]",
    "address": "123 Main St",
    "city": "San Francisco",
    "postal_code": "94102",
    "customer_code": "CUST001"
})

print(f"Created customer: {new_customer.id}")

Update an existing customer:

updated = await client.customers.update(
    id=customer.id,
    payload={"email": "[e-mail]", "note": "VIP customer"}
)

print(f"Updated {updated.name}")

Delete a customer:

result = await client.customers.delete(id=customer.id)
print(result)  # {'deleted_object_ids': ['customer-uuid']}

Iterate through all customers:

# Automatically handles pagination across all pages
async for customer in client.customers.iter_all():
    print(f"{customer.name} - Last visit: {customer.last_visit}")

Filter customers by date:

from datetime import datetime

# Get customers created in the last 30 days
start_date = datetime.now() - timedelta(days=30)

async for customer in client.customers.iter_all(created_at_min=start_date):
    tenure = customer.tenure()  # timedelta between first and last visit
    print(f"{customer.name} - Customer for {tenure.days} days")

Other Endpoints

All endpoints follow the same pattern. Available endpoints:

client.categories   # Item categories
client.customers    # Customer records
client.discounts    # Discount rules
client.devices      # POS devices
client.employees    # Staff members
client.inventory    # Stock levels
client.items        # Inventory items
client.merchant     # Merchant info
client.modifiers    # Item modifiers
client.receipts     # Transaction receipts
client.shifts       # Employee shifts
client.stores       # Store locations
client.suppliers    # Supplier records
client.taxes        # Tax configurations
client.variants     # Item variants
client.webhooks     # Webhook subscriptions

Each endpoint supports operations based on the Loyverse API capabilities.

DuckDB Export

The SDK includes powerful export functionality to save all your Loyverse data to a local DuckDB database for analytics, reporting, and data warehousing.

Why DuckDB?

DuckDB is an analytics-focused database perfect for:

• Fast analytical queries on large datasets
• Local data warehousing without server infrastructure
• SQL analytics with familiar syntax
• Integration with Python, R, and BI tools
• Efficient storage with columnar compression

Features

• ✅ 15 main resource tables (categories, items, receipts, etc.)
• ✅ Relational schema with foreign keys and indexes
• ✅ Junction tables for many-to-many relationships
• ✅ Child tables for nested data (line items, modifier options)
• ✅ Full and incremental exports with date range filtering
• ✅ Streaming export for memory efficiency
• ✅ UPSERT support (INSERT OR REPLACE) to prevent duplicates
• ✅ Progress tracking with callback support

Quick Start

Full export:

import asyncio
from loyverse_sdk import LoyverseClient

async def main():
    client = LoyverseClient()

    # Export all data to DuckDB
    counts = await client.export_to_duckdb("loyverse.duckdb")

    print(f"Exported {sum(counts.values())} total records")
    # Output: {'categories': 15, 'customers': 1250, 'receipts': 45000, ...}

    await client.close()

asyncio.run(main())

Query exported data:

import duckdb

conn = duckdb.connect("loyverse.duckdb")

# Top 10 customers by total spent
result = conn.execute("""
    SELECT
        c.name,
        COUNT(DISTINCT r.id) as receipt_count,
        SUM(r.total_amount) as total_spent
    FROM customers c
    JOIN receipts r ON c.id = r.customer_id
    WHERE r.receipt_type = 'SALE'
    GROUP BY c.id, c.name
    ORDER BY total_spent DESC
    LIMIT 10
""").fetchall()

conn.close()

Export Methods

1. Full Export with Options

Export all or selected resources with comprehensive filtering:

from datetime import datetime, timedelta

client = LoyverseClient()

# Export with all options
counts = await client.export_to_duckdb(
    db_path="loyverse.duckdb",
    resources=["receipts", "customers", "items"],  # Optional: specific resources
    created_at_min=datetime(2024, 1, 1),           # Optional: start date
    created_at_max=datetime(2024, 12, 31),         # Optional: end date
    updated_at_min=datetime.now() - timedelta(days=7),  # Optional: updated after
    batch_size=1000,                                # Optional: records per batch
    progress_callback=lambda res, curr, total: print(f"{res}: {curr}"),  # Optional
    create_indexes=True                             # Optional: create indexes after
)

print(f"Exported: {counts}")
# Returns: {'receipts': 5000, 'customers': 1200, 'items': 350}

await client.close()

2. Single Resource Export

Export one resource with fine-grained control:

client = LoyverseClient()

# Export only receipts from last 30 days
count = await client.export_resource_to_duckdb(
    resource_name="receipts",
    db_path="loyverse.duckdb",
    created_at_min=datetime.now() - timedelta(days=30)
)

print(f"Exported {count} receipts")

await client.close()

3. Schema Initialization

Create database schema without exporting data:

client = LoyverseClient()

# Initialize empty database with schema
client.init_duckdb_schema("loyverse.duckdb")

# Or reset existing database
client.init_duckdb_schema("loyverse.duckdb", drop_existing=True)

Advanced Usage

Progress tracking:

def progress_callback(resource_name: str, current: int, total: int):
    """Called for each batch of records."""
    print(f"Exporting {resource_name}: {current:,} records processed...")

counts = await client.export_to_duckdb(
    "loyverse.duckdb",
    progress_callback=progress_callback
)

Incremental updates:

# Export only records updated in last 24 hours
yesterday = datetime.now() - timedelta(days=1)

counts = await client.export_to_duckdb(
    "loyverse.duckdb",
    updated_at_min=yesterday
)

# UPSERT semantics: existing records are updated, new ones inserted

Selective export:

# Export only what you need
counts = await client.export_to_duckdb(
    "loyverse.duckdb",
    resources=[
        "receipts",      # Transaction data
        "customers",     # Customer profiles
        "items",         # Product catalog
        "categories"     # Item categories
    ]
)

Database Schema

The exported database includes:

Main Tables (15):

• categories - Item categories
• stores - Store locations
• suppliers - Supplier records
• taxes - Tax configurations
• modifiers - Item modifiers
• discounts - Discount rules
• employees - Staff members
• customers - Customer records
• pos_devices - POS devices
• payment_types - Payment methods
• items - Inventory items
• variants - Item variants
• receipts - Transaction receipts
• inventory - Stock levels
• merchant - Merchant info

Junction Tables (8):

• employee_store - Employee-to-store assignments
• item_tax - Item-to-tax relationships
• item_modifier - Item-to-modifier relationships
• modifier_store - Modifier-to-store assignments
• tax_store - Tax-to-store assignments
• discount_store - Discount-to-store assignments
• payment_type_store - Payment type availability by store
• variant_store - Variant inventory by store

Child Tables (2):

• receipt_line_items - Individual line items per receipt
• modifier_options - Options within modifiers

Metadata:

• sync_metadata - Tracks export history and record counts

Example Queries

Daily revenue:

SELECT
    DATE(receipt_date) as date,
    COUNT(*) as receipt_count,
    SUM(total_amount) as revenue
FROM receipts
WHERE receipt_type = 'SALE'
  AND receipt_date >= CURRENT_DATE - INTERVAL '30 days'
GROUP BY DATE(receipt_date)
ORDER BY date DESC;

Best-selling items:

SELECT
    i.name,
    SUM(l.quantity) as units_sold,
    SUM(l.quantity * l.price) as revenue
FROM items i
JOIN receipt_line_items l ON i.id = l.item_id
JOIN receipts r ON l.receipt_id = r.id
WHERE r.receipt_type = 'SALE'
GROUP BY i.id, i.name
ORDER BY units_sold DESC
LIMIT 10;

Customer lifetime value:

SELECT
    c.name,
    c.total_visits,
    c.total_spent,
    c.total_spent / NULLIF(c.total_visits, 0) as avg_per_visit
FROM customers c
WHERE c.total_visits > 0
ORDER BY c.total_spent DESC
LIMIT 20;

Inventory by category:

SELECT
    cat.name as category,
    COUNT(DISTINCT i.id) as item_count,
    COUNT(DISTINCT v.id) as variant_count
FROM categories cat
LEFT JOIN items i ON cat.id = i.category_id
LEFT JOIN variants v ON i.id = v.item_id
GROUP BY cat.id, cat.name
ORDER BY item_count DESC;

Performance Tips

1. Batch size: Default is 1000 records per transaction. Increase for faster exports on powerful machines:

   counts = await client.export_to_duckdb("loyverse.duckdb", batch_size=5000)
   

2. Indexes: Created automatically after export. Disable for faster initial load:

   counts = await client.export_to_duckdb("loyverse.duckdb", create_indexes=False)
   

3. Memory: DuckDB is configured with 4GB memory limit by default. Efficient for datasets with millions of records.

4. Incremental updates: Export only changed records to minimize transfer time:

   # Daily sync: export only yesterday's data
   yesterday = datetime.now() - timedelta(days=1)
   counts = await client.export_to_duckdb("loyverse.duckdb", created_at_min=yesterday)
   

Use Cases

• Business Intelligence: Connect DuckDB to Metabase, Superset, or Tableau
• Custom Reports: Write SQL queries for specific business questions
• Data Science: Analyze sales patterns, customer behavior, inventory trends
• Backup: Maintain local copy of all POS data
• Data Warehouse: Centralize data for cross-system analytics
• Migration: Export data for migration to other systems

Complete Example

See examples/duckdb_export.py for comprehensive examples including:

• Full and selective exports
• Date range filtering
• Progress tracking
• Querying exported data
• Incremental updates

python examples/duckdb_export.py