Query with SDKs

npm install @parable/flight-sql-client apache-arrow

pip install adbc-driver-flightsql pyarrow

go get github.com/apache/arrow-go/v18/arrow/flight/flightsql

cargo add arrow-flight futures tonic

import { tableFromIPC } from "apache-arrow";
import { createFlightSqlClient } from "@parable/flight-sql-client";

const endpoint = new URL(process.env.PARABLE_FLIGHT_URL!);
const token = process.env.PARABLE_API_TOKEN!;
const workspace = process.env.PARABLE_WORKSPACE!;

const client = await createFlightSqlClient({
  host: endpoint.hostname,
  port: Number(endpoint.port || "443"),
  tls: endpoint.protocol === "https:",
  headers: [
    ["authorization", `Bearer ${token}`],
    ["x-tenant", workspace],
  ],
});

try {
  const ipc = await client.query(`
    SELECT id, primaryemail, orgunitpath
    FROM providers.google.users
    WHERE suspended = false
    LIMIT 10
  `);
  console.log(tableFromIPC(ipc).toArray());
} finally {
  await client.close();
}

import os

import adbc_driver_flightsql.dbapi
from adbc_driver_flightsql import DatabaseOptions

with adbc_driver_flightsql.dbapi.connect(
    os.environ["PARABLE_FLIGHT_URL"],
    db_kwargs={
        DatabaseOptions.AUTHORIZATION_HEADER.value:
            f"Bearer {os.environ['PARABLE_API_TOKEN']}",
        f"{DatabaseOptions.RPC_CALL_HEADER_PREFIX.value}x-tenant":
            os.environ["PARABLE_WORKSPACE"],
    },
) as connection:
    with connection.cursor() as cursor:
        cursor.execute("""
            SELECT id, primaryemail, orgunitpath
            FROM providers.google.users
            WHERE suspended = false
            LIMIT 10
        """)
        print(cursor.fetch_arrow_table())

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/apache/arrow-go/v18/arrow/flight/flightsql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

ctx := metadata.AppendToOutgoingContext(
	context.Background(),
	"authorization", "Bearer "+os.Getenv("PARABLE_API_TOKEN"),
	"x-tenant", os.Getenv("PARABLE_WORKSPACE"),
)

client, err := flightsql.NewClient(
	os.Getenv("PARABLE_FLIGHT_ENDPOINT"),
	nil,
	nil,
	grpc.WithTransportCredentials(credentials.NewTLS(nil)),
)
if err != nil {
	log.Fatal(err)
}
defer client.Close()

info, err := client.Execute(ctx, `
  SELECT id, primaryemail, orgunitpath
  FROM providers.google.users
  WHERE suspended = false
  LIMIT 10
`)
if err != nil {
	log.Fatal(err)
}

reader, err := client.DoGet(ctx, info.Endpoint[0].Ticket)
if err != nil {
	log.Fatal(err)
}
defer reader.Release()

for reader.Next() {
	fmt.Println(reader.Record())
}

let token = std::env::var("PARABLE_API_TOKEN")?;
let workspace = std::env::var("PARABLE_WORKSPACE")?;
let channel = tonic::transport::Channel::from_shared(
    std::env::var("PARABLE_FLIGHT_URL")?,
)?
.connect()
.await?;

let mut client = arrow_flight::sql::client::FlightSqlServiceClient::new(channel);
client.set_header("authorization", format!("Bearer {token}"));
client.set_header("x-tenant", workspace);

let info = client
    .execute(
        r#"
        SELECT id, primaryemail, orgunitpath
        FROM providers.google.users
        WHERE suspended = false
        LIMIT 10
        "#
        .to_string(),
        None,
    )
    .await?;

let ticket = info.endpoint[0].ticket.clone().expect("ticket");
let mut stream = client.do_get(ticket.into()).await?;
while let Some(batch) = futures::TryStreamExt::try_next(&mut stream).await? {
    println!("{batch:?}");
}