Data Export - Qwairy Documentation

Export your Qwairy data for analysis in spreadsheets, databases, or BI platforms.

This guide uses the API client from the Guides index. Copy it to your project first.

What You’ll Build

Export pipelines for:

CSV files: For Excel, Google Sheets
JSON files: For data pipelines
BigQuery: For Google Cloud analytics
Snowflake: For enterprise data warehousing

Fetch All Data

First, gather data from all endpoints.

async function fetchAllData(client, brandId, period = 30) {
  const [performance, competitors, sources] = await Promise.all([
    client.getPerformance(brandId, { period }),
    client.getCompetitors(brandId, { period, limit: 100 }),
    client.getSourceDomains(brandId, { period, limit: 100 }),
  ]);

  return {
    exportedAt: new Date().toISOString(),
    brandId,
    period,
    performance: {
      scores: performance.scores,
      methodology: performance.methodology,
    },
    competitors: competitors.competitors.map(c => ({
      id: c.id,
      name: c.name,
      relationship: c.relationship,
      shareOfVoice: c.shareOfVoice,
      totalMentions: c.totalMentions,
      avgPosition: c.avgPosition,
      avgSentiment: c.avgSentiment,
    })),
    sources: sources.sources.map(s => ({
      id: s.id,
      domain: s.domain,
      type: s.type,
      isSelf: s.isSelf,
      totalMentions: s.totalMentions,
      rate: s.rate,
      avgPosition: s.avgPosition,
    })),
  };
}

def fetch_all_data(client, brand_id: str, period: int = 30) -> dict:
    """Fetch all data for export."""
    from concurrent.futures import ThreadPoolExecutor
    from datetime import datetime

    with ThreadPoolExecutor(max_workers=3) as executor:
        perf_future = executor.submit(client.get_performance, brand_id, period=period)
        comp_future = executor.submit(client.get_competitors, brand_id, period=period, limit=100)
        src_future = executor.submit(client.get_source_domains, brand_id, period=period, limit=100)

        performance = perf_future.result()
        competitors = comp_future.result()
        sources = src_future.result()

    return {
        'exported_at': datetime.now().isoformat(),
        'brand_id': brand_id,
        'period': period,
        'performance': {
            'scores': performance['scores'],
            'methodology': performance['methodology'],
        },
        'competitors': [
            {
                'id': c['id'],
                'name': c['name'],
                'relationship': c['relationship'],
                'share_of_voice': c['shareOfVoice'],
                'total_mentions': c['totalMentions'],
                'avg_position': c['avgPosition'],
                'avg_sentiment': c['avgSentiment'],
            }
            for c in competitors['competitors']
        ],
        'sources': [
            {
                'id': s['id'],
                'domain': s['domain'],
                'type': s['type'],
                'is_self': s['isSelf'],
                'total_mentions': s['totalMentions'],
                'rate': s['rate'],
                'avg_position': s['avgPosition'],
            }
            for s in sources['sources']
        ],
    }

Export to CSV

const fs = require('fs');

function exportToCSV(data, outputDir = './exports') {
  if (!fs.existsSync(outputDir)) {
    fs.mkdirSync(outputDir, { recursive: true });
  }

  const timestamp = data.exportedAt.split('T')[0];

  // Competitors CSV
  const competitorHeaders = ['id', 'name', 'relationship', 'shareOfVoice', 'totalMentions', 'avgPosition', 'avgSentiment'];
  const competitorRows = data.competitors.map(c => competitorHeaders.map(h => c[h]).join(','));
  fs.writeFileSync(
    `${outputDir}/competitors_${timestamp}.csv`,
    [competitorHeaders.join(','), ...competitorRows].join('\n')
  );

  // Sources CSV
  const sourceHeaders = ['id', 'domain', 'type', 'isSelf', 'totalMentions', 'rate', 'avgPosition'];
  const sourceRows = data.sources.map(s => sourceHeaders.map(h => s[h]).join(','));
  fs.writeFileSync(
    `${outputDir}/sources_${timestamp}.csv`,
    [sourceHeaders.join(','), ...sourceRows].join('\n')
  );

  // Performance CSV (single row)
  const perfHeaders = ['mentionRate', 'sourceRate', 'coverage', 'shareOfVoice', 'sentiment'];
  const perfValues = [
    data.performance.scores.mentionRate,
    data.performance.scores.sourceRate,
    data.performance.scores.coverage,
    data.performance.scores.shareOfVoice,
    data.performance.scores.sentiment,
  ];
  fs.writeFileSync(
    `${outputDir}/performance_${timestamp}.csv`,
    [perfHeaders.join(','), perfValues.join(',')].join('\n')
  );

  return {
    files: [
      `${outputDir}/competitors_${timestamp}.csv`,
      `${outputDir}/sources_${timestamp}.csv`,
      `${outputDir}/performance_${timestamp}.csv`,
    ],
  };
}

import csv
import os


def export_to_csv(data: dict, output_dir: str = './exports') -> dict:
    """Export data to CSV files."""
    os.makedirs(output_dir, exist_ok=True)
    timestamp = data['exported_at'].split('T')[0]
    files = []

    # Competitors CSV
    comp_file = f'{output_dir}/competitors_{timestamp}.csv'
    with open(comp_file, 'w', newline='') as f:
        if data['competitors']:
            writer = csv.DictWriter(f, fieldnames=data['competitors'][0].keys())
            writer.writeheader()
            writer.writerows(data['competitors'])
    files.append(comp_file)

    # Sources CSV
    src_file = f'{output_dir}/sources_{timestamp}.csv'
    with open(src_file, 'w', newline='') as f:
        if data['sources']:
            writer = csv.DictWriter(f, fieldnames=data['sources'][0].keys())
            writer.writeheader()
            writer.writerows(data['sources'])
    files.append(src_file)

    # Performance CSV
    perf_file = f'{output_dir}/performance_{timestamp}.csv'
    scores = data['performance']['scores']
    with open(perf_file, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=['mention_rate', 'source_rate', 'coverage', 'share_of_voice', 'sentiment'])
        writer.writeheader()
        writer.writerow({
            'mention_rate': scores['mentionRate'],
            'source_rate': scores['sourceRate'],
            'coverage': scores['coverage'],
            'share_of_voice': scores['shareOfVoice'],
            'sentiment': scores['sentiment'],
        })
    files.append(perf_file)

    return {'files': files}

Export to JSON

const fs = require('fs');

function exportToJSON(data, outputDir = './exports') {
  if (!fs.existsSync(outputDir)) {
    fs.mkdirSync(outputDir, { recursive: true });
  }

  const timestamp = data.exportedAt.split('T')[0];
  const filepath = `${outputDir}/qwairy_export_${timestamp}.json`;

  fs.writeFileSync(filepath, JSON.stringify(data, null, 2));

  return { file: filepath, size: fs.statSync(filepath).size };
}

import json
import os


def export_to_json(data: dict, output_dir: str = './exports') -> dict:
    """Export data to JSON file."""
    os.makedirs(output_dir, exist_ok=True)
    timestamp = data['exported_at'].split('T')[0]
    filepath = f'{output_dir}/qwairy_export_{timestamp}.json'

    with open(filepath, 'w') as f:
        json.dump(data, f, indent=2)

    return {'file': filepath, 'size': os.path.getsize(filepath)}

Database Schemas

Create these tables before exporting to BigQuery or Snowflake.

-- Create dataset
CREATE SCHEMA IF NOT EXISTS qwairy;

-- Competitors table
CREATE TABLE IF NOT EXISTS qwairy.competitors (
  id STRING NOT NULL,
  name STRING NOT NULL,
  relationship STRING,
  share_of_voice FLOAT64,
  total_mentions INT64,
  avg_position FLOAT64,
  avg_sentiment FLOAT64,
  exported_at TIMESTAMP NOT NULL,
  brand_id STRING NOT NULL
);

-- Sources table
CREATE TABLE IF NOT EXISTS qwairy.sources (
  id STRING NOT NULL,
  domain STRING NOT NULL,
  type STRING,
  is_self BOOL,
  total_mentions INT64,
  rate FLOAT64,
  avg_position FLOAT64,
  exported_at TIMESTAMP NOT NULL,
  brand_id STRING NOT NULL
);

-- Performance table
CREATE TABLE IF NOT EXISTS qwairy.performance (
  mentionRate FLOAT64,
  sourceRate FLOAT64,
  coverage FLOAT64,
  shareOfVoice FLOAT64,
  sentiment FLOAT64,
  methodology STRING,
  exported_at TIMESTAMP NOT NULL,
  brand_id STRING NOT NULL
);

-- Create schema
CREATE SCHEMA IF NOT EXISTS QWAIRY;

-- Competitors table
CREATE TABLE IF NOT EXISTS QWAIRY.COMPETITORS (
  ID VARCHAR(50) NOT NULL,
  NAME VARCHAR(255) NOT NULL,
  RELATIONSHIP VARCHAR(50),
  SHARE_OF_VOICE FLOAT,
  TOTAL_MENTIONS INTEGER,
  AVG_POSITION FLOAT,
  AVG_SENTIMENT FLOAT,
  EXPORTED_AT TIMESTAMP_NTZ NOT NULL,
  BRAND_ID VARCHAR(50) NOT NULL
);

-- Sources table
CREATE TABLE IF NOT EXISTS QWAIRY.SOURCES (
  ID VARCHAR(50) NOT NULL,
  DOMAIN VARCHAR(255) NOT NULL,
  TYPE VARCHAR(50),
  IS_SELF BOOLEAN,
  TOTAL_MENTIONS INTEGER,
  RATE FLOAT,
  AVG_POSITION FLOAT,
  EXPORTED_AT TIMESTAMP_NTZ NOT NULL,
  BRAND_ID VARCHAR(50) NOT NULL
);

-- Performance table
CREATE TABLE IF NOT EXISTS QWAIRY.PERFORMANCE (
  MENTION_RATE FLOAT,
  SOURCE_RATE FLOAT,
  COVERAGE FLOAT,
  SHARE_OF_VOICE FLOAT,
  SENTIMENT FLOAT,
  METHODOLOGY VARCHAR(5000),
  EXPORTED_AT TIMESTAMP_NTZ NOT NULL,
  BRAND_ID VARCHAR(50) NOT NULL
);

Export to BigQuery

const { BigQuery } = require('@google-cloud/bigquery');

async function exportToBigQuery(data, datasetId = 'qwairy', projectId = process.env.GCP_PROJECT_ID) {
  const bigquery = new BigQuery({ projectId });

  // Create dataset if not exists
  const [datasets] = await bigquery.getDatasets();
  if (!datasets.find(d => d.id === datasetId)) {
    await bigquery.createDataset(datasetId);
  }

  const dataset = bigquery.dataset(datasetId);

  // Insert competitors
  await dataset.table('competitors').insert(
    data.competitors.map(c => ({
      ...c,
      exportedAt: data.exportedAt,
      brandId: data.brandId,
    }))
  );

  // Insert sources
  await dataset.table('sources').insert(
    data.sources.map(s => ({
      ...s,
      exportedAt: data.exportedAt,
      brandId: data.brandId,
    }))
  );

  // Insert performance
  await dataset.table('performance').insert([{
    ...data.performance.scores,
    methodology: JSON.stringify(data.performance.methodology),
    exportedAt: data.exportedAt,
    brandId: data.brandId,
  }]);

  return { dataset: datasetId, tables: ['competitors', 'sources', 'performance'] };
}

from google.cloud import bigquery


def export_to_bigquery(data: dict, dataset_id: str = 'qwairy', project_id: str = None) -> dict:
    """Export data to BigQuery."""
    import os
    import json

    project_id = project_id or os.environ.get('GCP_PROJECT_ID')
    client = bigquery.Client(project=project_id)

    # Create dataset if not exists
    dataset_ref = client.dataset(dataset_id)
    try:
        client.get_dataset(dataset_ref)
    except Exception:
        client.create_dataset(dataset_ref)

    # Insert competitors
    competitors_table = f'{project_id}.{dataset_id}.competitors'
    rows = [
        {**c, 'exported_at': data['exported_at'], 'brand_id': data['brand_id']}
        for c in data['competitors']
    ]
    if rows:
        client.insert_rows_json(competitors_table, rows)

    # Insert sources
    sources_table = f'{project_id}.{dataset_id}.sources'
    rows = [
        {**s, 'exported_at': data['exported_at'], 'brand_id': data['brand_id']}
        for s in data['sources']
    ]
    if rows:
        client.insert_rows_json(sources_table, rows)

    # Insert performance
    perf_table = f'{project_id}.{dataset_id}.performance'
    perf_row = {
        **data['performance']['scores'],
        'methodology': json.dumps(data['performance']['methodology']),
        'exported_at': data['exported_at'],
        'brand_id': data['brand_id'],
    }
    client.insert_rows_json(perf_table, [perf_row])

    return {'dataset': dataset_id, 'tables': ['competitors', 'sources', 'performance']}

Export to Snowflake

const snowflake = require('snowflake-sdk');

async function exportToSnowflake(data, config) {
  const connection = snowflake.createConnection({
    account: config.account,
    username: config.username,
    password: config.password,
    warehouse: config.warehouse,
    database: config.database,
    schema: config.schema || 'QWAIRY',
  });

  await new Promise((resolve, reject) => {
    connection.connect((err) => err ? reject(err) : resolve());
  });

  const execute = (sql, binds = []) => new Promise((resolve, reject) => {
    connection.execute({ sqlText: sql, binds, complete: (err, stmt, rows) => err ? reject(err) : resolve(rows) });
  });

  // Insert competitors
  for (const c of data.competitors) {
    await execute(
      `INSERT INTO competitors (id, name, relationship, share_of_voice, total_mentions, avg_position, avg_sentiment, exported_at, brand_id)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [c.id, c.name, c.relationship, c.shareOfVoice, c.totalMentions, c.avgPosition, c.avgSentiment, data.exportedAt, data.brandId]
    );
  }

  // Insert sources
  for (const s of data.sources) {
    await execute(
      `INSERT INTO sources (id, domain, type, is_self, total_mentions, rate, avg_position, exported_at, brand_id)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      [s.id, s.domain, s.type, s.isSelf, s.totalMentions, s.rate, s.avgPosition, data.exportedAt, data.brandId]
    );
  }

  connection.destroy();
  return { schema: config.schema || 'QWAIRY', tables: ['competitors', 'sources'] };
}

import snowflake.connector


def export_to_snowflake(data: dict, config: dict) -> dict:
    """Export data to Snowflake."""
    conn = snowflake.connector.connect(
        account=config['account'],
        user=config['username'],
        password=config['password'],
        warehouse=config['warehouse'],
        database=config['database'],
        schema=config.get('schema', 'QWAIRY'),
    )
    cursor = conn.cursor()

    # Insert competitors
    for c in data['competitors']:
        cursor.execute(
            '''INSERT INTO competitors (id, name, relationship, share_of_voice, total_mentions, avg_position, avg_sentiment, exported_at, brand_id)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)''',
            (c['id'], c['name'], c['relationship'], c['share_of_voice'], c['total_mentions'],
             c['avg_position'], c['avg_sentiment'], data['exported_at'], data['brand_id'])
        )

    # Insert sources
    for s in data['sources']:
        cursor.execute(
            '''INSERT INTO sources (id, domain, type, is_self, total_mentions, rate, avg_position, exported_at, brand_id)
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)''',
            (s['id'], s['domain'], s['type'], s['is_self'], s['total_mentions'],
             s['rate'], s['avg_position'], data['exported_at'], data['brand_id'])
        )

    conn.commit()
    cursor.close()
    conn.close()

    return {'schema': config.get('schema', 'QWAIRY'), 'tables': ['competitors', 'sources']}

Usage

const client = new QwairyClient(process.env.QWAIRY_API_TOKEN);

// Fetch data
const data = await fetchAllData(client, 'your-brand-id', 30);

// Export to files
const csvResult = exportToCSV(data);
console.log('CSV files:', csvResult.files);

const jsonResult = exportToJSON(data);
console.log('JSON file:', jsonResult.file, `(${jsonResult.size} bytes)`);

// Export to BigQuery (requires @google-cloud/bigquery)
// const bqResult = await exportToBigQuery(data);
// console.log('BigQuery tables:', bqResult.tables);

client = QwairyClient()

# Fetch data
data = fetch_all_data(client, 'your-brand-id', period=30)

# Export to files
csv_result = export_to_csv(data)
print(f"CSV files: {csv_result['files']}")

json_result = export_to_json(data)
print(f"JSON file: {json_result['file']} ({json_result['size']} bytes)")

# Export to BigQuery (requires google-cloud-bigquery)
# bq_result = export_to_bigquery(data)
# print(f"BigQuery tables: {bq_result['tables']}")

Scheduling Exports

Automate daily or weekly exports:

Platform	Configuration
Cron	`0 1 * * *` (daily at 1am)
GitHub Actions	`schedule: cron: '0 1 * * *'`
AWS Lambda	EventBridge scheduled rule
Google Cloud	Cloud Scheduler + Cloud Functions

Next Steps

Create a custom dashboard for real-time monitoring
Set up weekly reports for stakeholders
Add competitive analysis to your exports

​What You’ll Build

​Fetch All Data

​Export to CSV

​Export to JSON

​Database Schemas

​Export to BigQuery

​Export to Snowflake

​Usage

​Scheduling Exports

​Next Steps

What You’ll Build

Fetch All Data

Export to CSV

Export to JSON

Database Schemas

Export to BigQuery

Export to Snowflake

Usage

Scheduling Exports

Next Steps