Usage Examples

Basic examples of dataset upload via API.

Python

import requests

API_KEY = 'abn_xxxxxxxxxxxxxxxxxxxxx'
API_URL = 'https://workbench.ab-labz.com/api/v1'

with open('dataset.csv', 'rb') as f:
    response = requests.post(
        f'{API_URL}/datasets/upload/',
        params={'experiment_id': 'homepage_test'},
        headers={
            'Authorization': f'Bearer {API_KEY}',
            'Content-Type': 'text/csv'
        },
        data=f
    )
    response.raise_for_status()

Bash

curl -X POST "https://workbench.ab-labz.com/api/v1/datasets/upload/?experiment_id=homepage_test" \
  -H "Authorization: Bearer abn_xxxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: text/csv" \
  --data-binary @dataset.csv

GitHub Actions

name: Upload to AB-Labz

on:
  push:
    branches: [main]

jobs:
  upload:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Upload dataset
        run: |
          curl -X POST "https://workbench.ab-labz.com/api/v1/datasets/upload/?experiment_id=homepage_test" \
            -H "Authorization: Bearer ${{ secrets.ABLABZ_API_KEY }}" \
            -H "Content-Type: text/csv" \
            --data-binary @data/experiment.csv

ETL Integration

Example for systems like Airflow or cron. Using data mart (see Data Preparation):

import requests
import pandas as pd

API_KEY = 'abn_xxxxxxxxxxxxxxxxxxxxx'
API_URL = 'https://workbench.ab-labz.com/api/v1'

# Get list of active experiments
# (those that had data updated yesterday/today)
active_experiments = pd.read_sql("""
    SELECT DISTINCT experiment_id
    FROM experiments_datamart
    WHERE dt >= CURRENT_DATE - INTERVAL '1 day'
""", your_db_connection)

# For each experiment: export data and upload
for exp_id in active_experiments['experiment_id']:
    # Export data from data mart
    df = pd.read_sql(f"""
        SELECT 
            user_id,
            variant,
            dt as date,
            purchase_conversion,
            revenue,
            page_views
        FROM experiments_datamart
        WHERE experiment_id = '{exp_id}'
    """, your_db_connection)
    
    # Save to CSV
    csv_path = f'/tmp/{exp_id}.csv'
    df.to_csv(csv_path, index=False)
    
    # Upload to AB-Labz
    with open(csv_path, 'rb') as f:
        response = requests.post(
            f'{API_URL}/datasets/upload/',
            params={'experiment_id': exp_id},
            headers={
                'Authorization': f'Bearer {API_KEY}',
                'Content-Type': 'text/csv'
            },
            data=f
        )
        response.raise_for_status()
    
    print(f"Uploaded {exp_id}: {response.json()['size_mb']} MB")

Airflow DAG

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
import requests
import pandas as pd

def upload_active_experiments():
    """Upload all active experiments from data mart"""
    API_KEY = os.getenv('ABLABZ_API_KEY')
    API_URL = 'https://workbench.ab-labz.com/api/v1'
    
    # List of experiments with data for the last day
    active_exps = pd.read_sql("""
        SELECT DISTINCT experiment_id
        FROM experiments_datamart
        WHERE dt >= CURRENT_DATE - INTERVAL '1 day'
    """, connection)
    
    for exp_id in active_exps['experiment_id']:
        # Export all experiment data from data mart
        df = pd.read_sql(f"""
            SELECT user_id, variant, dt as date, purchase_conversion, revenue
            FROM experiments_datamart 
            WHERE experiment_id = '{exp_id}'
        """, connection)
        csv_path = f'/tmp/{exp_id}.csv'
        df.to_csv(csv_path, index=False)
        
        with open(csv_path, 'rb') as f:
            requests.post(
                f'{API_URL}/datasets/upload/',
                params={'experiment_id': exp_id},
                headers={'Authorization': f'Bearer {API_KEY}', 'Content-Type': 'text/csv'},
                data=f
            ).raise_for_status()

dag = DAG(
    'ablabz_upload',
    schedule_interval='0 3 * * *',
    start_date=datetime(2025, 1, 1),
    catchup=False
)

PythonOperator(
    task_id='upload_experiments',
    python_callable=upload_active_experiments,
    dag=dag
)