🐍 Python SDK

Официальный SDK для Python с поддержкой синхронных и асинхронных вызовов, интеграцией с популярными фреймворками

🎯 Что вы получите
  • Простой и Pythonic API для отправки событий
  • Поддержка синхронных и асинхронных вызовов
  • Интеграция с Django, Flask, FastAPI
  • Автоматическая обработка ошибок и повторные попытки
  • Пакетная отправка событий для оптимизации

Установка

# Установка через pip
pip install instantbase-python

# С поддержкой асинхронности (aiohttp)
pip install instantbase-python[async]

# С дополнительными зависимостями для фреймворков
pip install instantbase-python[django,flask,fastapi]
# Poetry
poetry add instantbase-python

# С асинхронной поддержкой
poetry add instantbase-python[async]

# Со всеми зависимостями
poetry add instantbase-python[all]
# Pipenv
pipenv install instantbase-python

# С асинхронной поддержкой
pipenv install instantbase-python[async]

Быстрый старт

from instantbase import InstantBase

# Инициализация клиента
ib = InstantBase(
    api_key='your_api_key',
    environment='production',  # или 'development'
    debug=False
)

# Отправка простого события
ib.track('user_signup', {
    'user_id': 'user_123',
    'method': 'email'
})

# Отправка события с пользователем
ib.track('purchase', {
    'user_id': 'user_123',
    'amount': 1499.99,
    'order_id': 'ORD-001'
}, metadata={
    'is_order': True,
    'funnel_sales_step': 4
})

# Идентификация пользователя
ib.identify('user_123', {
    'email': 'user@example.com',
    'name': 'Иван Петров',
    'plan': 'premium'
})

Асинхронный клиент

from instantbase import AsyncInstantBase
import asyncio

async def main():
    # Асинхронный клиент
    ib = AsyncInstantBase(
        api_key='your_api_key',
        environment='production'
    )
    
    # Асинхронная отправка событий
    await ib.track('user_signup', {
        'user_id': 'user_123',
        'method': 'email'
    })
    
    # Пакетная отправка
    events = [
        {'event': 'page_view', 'user_id': 'user_1'},
        {'event': 'page_view', 'user_id': 'user_2'},
        {'event': 'page_view', 'user_id': 'user_3'}
    ]
    await ib.track_batch(events)
    
    # Закрытие клиента
    await ib.close()

asyncio.run(main())

Конфигурация SDK

__init__ Параметры инициализации
Параметр Тип По умолчанию Описание
api_key str — Обязательный. Ваш API-ключ
environment str 'production' Режим работы. 'development' для отладки
debug bool False Включает логирование в stdout
timeout float 5.0 Таймаут запроса в секундах
max_retries int 3 Количество повторных попыток
batch_size int 10 Размер батча для пакетной отправки
flush_interval int 5 Интервал отправки батча (секунды)

Основные методы

track() Отправка события async

Основной метод для отправки событий.

from instantbase import InstantBase

ib = InstantBase(api_key='your_key')

# Простое событие
ib.track('button_click', {
    'user_id': 'user_123',
    'button': 'signup',
    'page': '/home'
})

# С метаданными
ib.track('purchase', {
    'user_id': 'user_123',
    'order_id': 'ORD-001',
    'amount': 1499.99
}, metadata={
    'is_order': True,
    'funnel_sales_step': 4
})

# Асинхронная версия
async def send_event():
    await async_ib.track('page_view', {
        'user_id': 'user_123',
        'page': '/products'
    })
Параметр Тип Описание
event str Название события
properties dict Свойства события (обязательно содержит user_id или anonymous_id)
metadata dict Метаданные события (опционально)
timestamp datetime Время события (опционально)
identify() Идентификация пользователя async

Устанавливает атрибуты пользователя.

# Синхронно
ib.identify('user_123', {
    'email': 'user@example.com',
    'name': 'Иван Петров',
    'plan': 'premium',
    'age': 30
})

# С метаданными атрибутов
ib.identify('user_123', {
    'email': 'user@example.com',
    'plan': 'premium'
}, metadata={
    'plan': {
        'attribute_display_name': 'Тариф',
        'attribute_category': 'Подписка',
        'attribute_color': '#2A6DF4'
    }
})

# Асинхронно
await async_ib.identify('user_123', {
    'email': 'user@example.com'
})
alias() Связывание идентификаторов async

Связывает анонимный ID с авторизованным пользователем.

# Синхронно
ib.alias('anonymous_123', 'user_456')

# Асинхронно
await async_ib.alias('anonymous_123', 'user_456')
track_batch() Пакетная отправка async

Отправка нескольких событий одним запросом.

# Синхронно
events = [
    {'event': 'page_view', 'user_id': 'user_1', 'properties': {'page': '/home'}},
    {'event': 'page_view', 'user_id': 'user_2', 'properties': {'page': '/products'}},
    {'event': 'purchase', 'user_id': 'user_1', 'properties': {'amount': 100}}
]
ib.track_batch(events)

# Асинхронно
await async_ib.track_batch(events)
flush() Принудительная отправка async

Принудительно отправляет накопленные события из буфера.

# Синхронно
ib.flush()

# Асинхронно
await async_ib.flush()

Интеграция с Django

Настройка в Django

# settings.py

INSTANTBASE = {
    'API_KEY': 'your_production_key',
    'DEBUG_KEY': 'your_test_key',  # для разработки
    'ENVIRONMENT': 'production',    # или 'development'
    'BATCH_SIZE': 50,
    'FLUSH_INTERVAL': 10,
    'ASYNC': True  # использовать асинхронный клиент
}
# middleware.py
from instantbase import InstantBase
from django.conf import settings

class InstantBaseMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
        self.ib = InstantBase(
            api_key=settings.INSTANTBASE['API_KEY'],
            environment=settings.INSTANTBASE['ENVIRONMENT'],
            batch_size=settings.INSTANTBASE['BATCH_SIZE'],
            flush_interval=settings.INSTANTBASE['FLUSH_INTERVAL']
        )
    
    def __call__(self, request):
        # Добавляем клиент в request
        request.ib = self.ib
        
        # Отслеживаем просмотр страницы
        if hasattr(request, 'user') and request.user.is_authenticated:
            self.ib.track('page_view', {
                'user_id': str(request.user.id),
                'page': request.path,
                'method': request.method
            })
        
        response = self.get_response(request)
        return response
    
    def process_exception(self, request, exception):
        # Отслеживаем ошибки
        self.ib.track('server_error', {
            'url': request.path,
            'error': str(exception),
            'user_id': str(request.user.id) if request.user.is_authenticated else None
        })

Использование в Django views

# views.py
from django.shortcuts import render, redirect
from django.contrib.auth.decorators import login_required

def product_view(request, product_id):
    # Отслеживаем просмотр товара
    request.ib.track('product_view', {
        'user_id': str(request.user.id) if request.user.is_authenticated else None,
        'anonymous_id': request.session.session_key,
        'product_id': product_id,
        'product_name': product.name,
        'price': float(product.price)
    }, metadata={
        'funnel_sales_step': 1
    })
    
    return render(request, 'product.html', {'product': product})

@login_required
def add_to_cart(request, product_id):
    # Отслеживаем добавление в корзину
    request.ib.track('add_to_cart', {
        'user_id': str(request.user.id),
        'product_id': product_id,
        'quantity': request.POST.get('quantity', 1)
    }, metadata={
        'funnel_sales_step': 2
    })
    
    return redirect('cart')

def checkout(request):
    # Начало оформления заказа
    request.ib.track('checkout_start', {
        'user_id': str(request.user.id) if request.user.is_authenticated else None,
        'cart_total': cart.total
    }, metadata={
        'funnel_sales_step': 3
    })
    
    if request.method == 'POST':
        # Успешная покупка
        request.ib.track('purchase', {
            'user_id': str(request.user.id),
            'order_id': order.id,
            'amount': float(order.total)
        }, metadata={
            'funnel_sales_step': 4,
            'is_order': True
        })
        
        return redirect('order_success')
    
    return render(request, 'checkout.html')

Интеграция с Flask

from flask import Flask, request, g
from instantbase import InstantBase
import functools

app = Flask(__name__)
app.config['INSTANTBASE_API_KEY'] = 'your_api_key'

def get_ib():
    """Получение клиента InstantBase для текущего запроса"""
    if 'ib' not in g:
        g.ib = InstantBase(
            api_key=app.config['INSTANTBASE_API_KEY'],
            environment='development' if app.debug else 'production'
        )
    return g.ib

@app.teardown_appcontext
def close_ib(error):
    """Закрытие клиента после запроса"""
    ib = g.pop('ib', None)
    if ib is not None:
        ib.flush()

@app.before_request
def before_request():
    """Отслеживание просмотров страниц"""
    ib = get_ib()
    ib.track('page_view', {
        'anonymous_id': request.remote_addr,
        'page': request.path,
        'method': request.method,
        'user_agent': request.user_agent.string
    })

@app.route('/product/<int:product_id>')
def product_view(product_id):
    ib = get_ib()
    
    # Получаем товар из БД
    product = get_product(product_id)
    
    # Отслеживаем просмотр товара
    ib.track('product_view', {
        'anonymous_id': request.remote_addr,
        'product_id': product_id,
        'product_name': product.name,
        'price': float(product.price)
    }, metadata={
        'funnel_sales_step': 1
    })
    
    return render_template('product.html', product=product)

@app.route('/api/checkout', methods=['POST'])
def checkout():
    ib = get_ib()
    data = request.json
    
    # Отслеживаем начало оформления
    ib.track('checkout_start', {
        'anonymous_id': request.remote_addr,
        'cart_total': data['total']
    }, metadata={
        'funnel_sales_step': 3
    })
    
    # Обработка заказа...
    order = process_order(data)
    
    # Отслеживаем покупку
    ib.track('purchase', {
        'user_id': order.user_id,
        'order_id': order.id,
        'amount': float(order.total)
    }, metadata={
        'funnel_sales_step': 4,
        'is_order': True
    })
    
    return jsonify({'order_id': order.id})

Интеграция с FastAPI

from fastapi import FastAPI, Request, Depends
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
from instantbase import AsyncInstantBase
import uuid

app = FastAPI()

# Хранилище клиента
ib_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Инициализация при старте
    global ib_client
    ib_client = AsyncInstantBase(
        api_key='your_api_key',
        environment='production',
        batch_size=20
    )
    yield
    # Закрытие при остановке
    if ib_client:
        await ib_client.close()

app = FastAPI(lifespan=lifespan)

async def get_ib() -> AsyncInstantBase:
    """Dependency для получения клиента"""
    return ib_client

@app.middleware("http")
async def track_requests(request: Request, call_next):
    """Middleware для отслеживания запросов"""
    response = await call_next(request)
    
    # Генерируем anonymous_id если нет user_id
    anonymous_id = request.cookies.get('anonymous_id')
    if not anonymous_id:
        anonymous_id = str(uuid.uuid4())
        response.set_cookie('anonymous_id', anonymous_id)
    
    # Получаем клиент
    ib = await get_ib()
    
    # Отслеживаем запрос
    await ib.track('api_request', {
        'anonymous_id': anonymous_id,
        'path': request.url.path,
        'method': request.method,
        'status_code': response.status_code
    })
    
    return response

@app.get("/api/products/{product_id}")
async def get_product(
    product_id: int,
    request: Request,
    ib: AsyncInstantBase = Depends(get_ib)
):
    # Получаем anonymous_id из cookies
    anonymous_id = request.cookies.get('anonymous_id')
    
    # Отслеживаем просмотр товара
    await ib.track('product_view', {
        'anonymous_id': anonymous_id,
        'product_id': product_id,
        'source': 'api'
    }, metadata={
        'funnel_sales_step': 1
    })
    
    product = await get_product_from_db(product_id)
    return product

@app.post("/api/purchase")
async def purchase(
    request: Request,
    purchase_data: PurchaseData,
    ib: AsyncInstantBase = Depends(get_ib)
):
    anonymous_id = request.cookies.get('anonymous_id')
    
    # Отслеживаем покупку
    await ib.track('purchase', {
        'anonymous_id': anonymous_id,
        'user_id': purchase_data.user_id if hasattr(purchase_data, 'user_id') else None,
        'order_id': purchase_data.order_id,
        'amount': purchase_data.amount,
        'items': purchase_data.items
    }, metadata={
        'funnel_sales_step': 4,
        'is_order': True
    })
    
    return JSONResponse({'status': 'success'})

Работа с Celery

# tasks.py
from celery import Celery
from instantbase import InstantBase
import os

app = Celery('tasks')
ib = InstantBase(api_key=os.environ['INSTANTBASE_KEY'])

@app.task
def track_event_async(event_name, properties, metadata=None):
    """Асинхронная отправка событий через Celery"""
    try:
        ib.track(event_name, properties, metadata)
    except Exception as e:
        # Логируем ошибку
        print(f"Failed to track event: {e}")
        # Можно повторить позже
        track_event_async.retry(countdown=60)

@app.task
def track_batch_async(events):
    """Пакетная отправка через Celery"""
    try:
        ib.track_batch(events)
    except Exception as e:
        print(f"Failed to track batch: {e}")
        track_batch_async.retry(countdown=60)

# Использование в views
def order_created(order):
    # Отправляем событие через Celery
    track_event_async.delay('purchase', {
        'user_id': order.user_id,
        'order_id': order.id,
        'amount': float(order.total)
    }, {
        'is_order': True
    })

Обработка ошибок

from instantbase import InstantBase, InstantBaseError, RateLimitError, AuthError

ib = InstantBase(api_key='your_key')

try:
    ib.track('purchase', {
        'user_id': 'user_123',
        'amount': 100
    })
except RateLimitError as e:
    print(f"Rate limit exceeded. Retry after: {e.retry_after}")
    # Сохраняем событие для повторной отправки
    save_for_retry(event)
except AuthError as e:
    print(f"Authentication error: {e}")
    # Проверяем API ключ
except InstantBaseError as e:
    print(f"General error: {e}")
    # Логируем и продолжаем

# Асинхронная версия
try:
    await async_ib.track('purchase', {
        'user_id': 'user_123',
        'amount': 100
    })
except RateLimitError as e:
    print(f"Rate limit: {e.retry_after}")

Контекстные менеджеры

from instantbase import InstantBase

# Автоматический flush при выходе из контекста
with InstantBase(api_key='your_key') as ib:
    ib.track('event1', {'user_id': 'user_1'})
    ib.track('event2', {'user_id': 'user_2'})
    # flush() вызовется автоматически

# Асинхронный контекстный менеджер
async with AsyncInstantBase(api_key='your_key') as ib:
    await ib.track('event1', {'user_id': 'user_1'})
    await ib.track('event2', {'user_id': 'user_2'})
    # Автоматическое закрытие и flush

Логирование и отладка

import logging
from instantbase import InstantBase

# Настройка логирования
logging.basicConfig(level=logging.DEBUG)

# SDK будет использовать этот логгер
ib = InstantBase(
    api_key='your_key',
    debug=True  # включить подробное логирование
)

# Или передать свой логгер
import logging
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

ib = InstantBase(
    api_key='your_key',
    logger=logger
)

# Пример вывода:
# DEBUG:instantbase:Initializing client with api_key=pro...
# DEBUG:instantbase:Tracking event: purchase
# DEBUG:instantbase:Sending request to https://api.instantbase.online/v1/track
# DEBUG:instantbase:Response: {"status":"ok","queued":true}

Сравнение синхронного и асинхронного клиентов

Характеристика SyncInstantBase AsyncInstantBase
Подходит для CLI скрипты, простые задачи Web-приложения, FastAPI, aiohttp
Производительность Блокирующий I/O Неблокирующий, высокая
Пакетная обработка ✅ ✅
Контекстный менеджер ✅ ✅
Зависимости requests aiohttp

Рекомендации

  • Для веб-приложений используйте асинхронный клиент — это не блокирует основной поток
  • Настраивайте batch_size в зависимости от нагрузки — 50-100 событий обычно оптимально
  • Всегда используйте flush() перед завершением работы — чтобы не потерять события
  • Оборачивайте вызовы в try-except — сетевые ошибки могут случаться
  • Для high-load проектов используйте Celery — для асинхронной отправки
  • Храните API ключи в переменных окружения — никогда не коммитьте их в репозиторий
⚠️ ВАЖНО: Production API-ключи должны храниться в безопасном месте. Используйте переменные окружения или секретные менеджеры.

Что дальше?

🐘 PHP SDK

Для проектов на PHP

Перейти →

📱 Мобильные SDK

Для iOS и Android приложений

Перейти →

Нужна помощь с Python SDK?

Напишите нам, и мы поможем настроить интеграцию

support@instantbase.online Telegram