Python Flight SQL DBAPI client

The Python flightsql-dbapi Flight SQL DBAPI library integrates with Python applications using SQL to query data stored in an InfluxDB Cloud Serverless bucket. The flightsql-dbapi library uses the Flight SQL protocol to query and retrieve data.

Installation

The flightsql-dbapi Flight SQL library for Python provides a DB API 2 interface and SQLAlchemy dialect for Flight SQL. Installing flightsql-dbapi also installs the pyarrow library that you’ll use for working with Arrow data.

In your terminal, use pip to install flightsql-dbapi:

pip install flightsql-dbapi

Importing the module

The flightsql-dbapi package provides the flightsql module. From the module, import the FlightSQLClient class method:

from flightsql import FlightSQLClient

• flightsql.FlightSQLClient class: an interface for initializing a client and interacting with a Flight SQL server.

API reference

• Class FlightSQLClient
  • Syntax
• Initialize a client
  • Instance methods
  • FlightSQLClient.execute
    • Syntax {#execute-query-syntax}
    • Example {#execute-query-example}
  • FlightSQLClient.do_get
    • Syntax {#retrieve-data-syntax}
    • Example {#retrieve-data-example}

Class FlightSQLClient

Provides an interface for initializing a client and interacting with a Flight SQL server.

Syntax

__init__(self, host=None, token=None, metadata=None, features=None)

Initializes and returns a FlightSQLClient instance for interacting with the server.

Initialize a client

The following example shows how to use Python with flightsql-dbapi and the DB API 2 interface to instantiate a Flight SQL client configured for an InfluxDB database.

from flightsql import FlightSQLClient

# Instantiate a FlightSQLClient configured for a database
client = FlightSQLClient(host='cloud2.influxdata.com',
                        token='
API_TOKEN
',
                        metadata={'database': '
BUCKET_NAME
'},
                        features={'metadata-reflection': 'true'})

Replace the following:

• API_TOKEN: an InfluxDB Cloud Serverless API token with read permissions on the buckets you want to query
• BUCKET_NAME: the name of your InfluxDB Cloud Serverless bucket

Instance methods

FlightSQLClient.execute

Sends a Flight SQL RPC request to execute the specified SQL Query.

Syntax

execute(query: str, call_options: Optional[FlightSQLCallOptions] = None)

Example

# Execute the query
info = client.execute("SELECT * FROM home")

The response contains a flight.FlightInfo object that contains metadata and an endpoints: [...] list. Each endpoint contains the following:

• A list of addresses where you can retrieve query result data.
• A ticket value that identifies the data to retrieve.

FlightSQLClient.do_get

Passes a Flight ticket (obtained from a FlightSQLClient.execute response) and retrieves Arrow data identified by the ticket. Returns a pyarrow.flight.FlightStreamReader for streaming the data.

Syntax

 do_get(ticket, call_options: Optional[FlightSQLCallOptions] = None)

Example

The following sample shows how to use Python with flightsql-dbapi and pyarrow to query InfluxDB and retrieve data.

from flightsql import FlightSQLClient

# Instantiate a FlightSQLClient configured for a database
client = FlightSQLClient(host='cloud2.influxdata.com',
    token='API_TOKEN',
    metadata={'database': 'BUCKET_NAME'},
    features={'metadata-reflection': 'true'})

# Execute the query to retrieve FlightInfo
info = client.execute("SELECT * FROM home")

# Extract the token for retrieving data
ticket = info.endpoints[0].ticket

# Use the ticket to request the Arrow data stream.
# Return a FlightStreamReader for streaming the results.
reader = client.do_get(ticket)

# Read all data to a pyarrow.Table
table = reader.read_all()

print(table)

do_get(ticket) returns a pyarrow.flight.FlightStreamReader for streaming Arrow record batches.

To read data from the stream, call one of the following FlightStreamReader methods:

• read_all(): Read all record batches as a pyarrow.Table.
• read_chunk(): Read the next RecordBatch and metadata.
• read_pandas(): Read all record batches and convert them to a pandas.DataFrame.