ingest_data_tinybird.py

import requests
import urllib.parse
import json


def jprint(data: str):
    """Prints JSON-like string (data) nicely"""
    print(json.dumps(data, indent=4, ensure_ascii=False))

def ingest_data(datasource: str,
                token: str,
                csv_url: str = None,
                csv_file: str = None,
                mode='append'):
    print(f'Appending {csv_url or csv_file} to the Tinybird Data Source {datasource}')
    headers = {'Authorization': f'Bearer {token}'}
    base_url = 'https://api.tinybird.co/v0/datasources'
    url = f'{base_url}?name={datasource}&mode={mode}'
    if csv_file:
        with open(csv_file, 'rb') as f:
            files = {'csv': f}  # the key has to be *exactly* 'csv' or it'll fail silently (0 rows will be imported)
            response = requests.post(url, headers=headers, files=files)
    elif csv_url:
        url += f'&url={urllib.parse.quote(csv_url)}'
        response = requests.post(url, headers=headers)
    return jprint(response.json())