Examples¶

Getting started¶

connect() quick start¶

The following example leases a pooled sender from a QuestDB handle, writes a row over QWP/WebSocket, waits for the server acknowledgement, then runs a bind-parameter query into pandas.

import sys

import questdb
from questdb import QuestDBError, TimestampNanos


def example(host: str = 'localhost', port: int = 9000):
    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            # Lease a pooled sender for row-by-row ingestion.
            # Take one lease per thread; leaving the `with` block
            # returns it to the pool.
            with db.sender() as sender:
                sender.row(
                    'trades',
                    symbols={
                        'symbol': 'ETH-USD',
                        'side': 'sell'},
                    columns={
                        'price': 2615.54,
                        'amount': 0.00044},
                    at=TimestampNanos.now())

                # Pooled rows auto-publish at the configured thresholds.
                # `wait=True` also publishes any partial batch, then blocks
                # until the server acknowledges everything on this lease.
                sender.flush(wait=True)

            # Query with positional bind parameters ($1..$N) and read the
            # result straight into pandas. Fully draining the result
            # returns its connection to the pool.
            frame = db.query(
                'SELECT symbol, price, amount FROM trades '
                'WHERE symbol = $1 LIMIT 5',
                ['ETH-USD'],
            ).to_pandas()
            print(frame)

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    example()

Queries¶

The following example runs DDL through execute(), binds positional parameters, streams a large result batch by batch, and leases one reader connection for a run of queries.

import sys

import questdb
from questdb import QuestDBError


def example(host: str = 'localhost', port: int = 9000):
    try:
        with questdb.connect(f'ws::addr={host}:{port};') as db:
            # execute() runs DDL and DML statements: it drains the
            # (empty) result and returns the pooled connection in one
            # call.
            db.execute(
                'CREATE TABLE IF NOT EXISTS trades ('
                '  timestamp TIMESTAMP, symbol SYMBOL,'
                '  price DOUBLE, amount DOUBLE'
                ') TIMESTAMP(timestamp) PARTITION BY DAY WAL')

            # Bind positional parameters to $1..$N placeholders instead of
            # interpolating values into the SQL.
            frame = db.query(
                'SELECT timestamp, price, amount FROM trades '
                'WHERE symbol = $1 AND price > $2 LIMIT 10',
                ['ETH-USD', 1000.0],
            ).to_pandas()
            print(frame)

            # Stream a large result batch by batch instead of
            # materializing all of it; fully draining the iterator
            # returns the connection to the pool.
            total = 0
            with db.query('SELECT * FROM trades') as result:
                for batch in result.iter_pandas():
                    total += len(batch)
            print('rows streamed:', total)

            # Lease one reader connection for a run of queries.
            with db.reader() as r:
                latest = r.query(
                    'SELECT * FROM trades LIMIT -5').to_pandas()
                per_symbol = r.query(
                    'SELECT symbol, count() FROM trades').to_pandas()
            print(latest)
            print(per_symbol)

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    example()

DataFrame bulk load and reader leases¶

The following example bulk-loads a pandas DataFrame through QuestDB.dataframe, then leases one reader connection with db.reader() and runs two queries on it.

import sys

import pandas as pd

import questdb
from questdb import QuestDBError


def example(host: str = 'localhost', port: int = 9000):
    df = pd.DataFrame({
        'symbol': pd.Categorical(['ETH-USD', 'BTC-USD']),
        'side': pd.Categorical(['sell', 'sell']),
        'price': [2615.54, 39269.98],
        'amount': [0.00044, 0.001],
        'timestamp': pd.to_datetime(
            ['2021-01-01T00:00:00Z', '2021-01-01T00:00:01Z'])})
    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            # Bulk-load the whole frame over the direct columnar path.
            # Categorical columns become SYMBOL automatically
            # (symbols='auto'); the call blocks until the server
            # acknowledges the final batch.
            db.dataframe(df, table_name='trades', at='timestamp')

            # Lease one reader connection and run queries on it
            # sequentially.
            with db.reader() as q:
                recent = q.query('SELECT * FROM trades LIMIT 5').to_pandas()
                print(recent)

                by_symbol = q.query(
                    'SELECT symbol, count() FROM trades'
                ).to_pandas()
                print(by_symbol)

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    example()

Ticking data and flush cadence¶

The following example mimics an application loop producing rows at random intervals. It disables pooled auto-flush and publishes on an explicit row-count / elapsed-time cadence.

import random
import sys
import time
import uuid

import questdb
from questdb import QuestDBError, TimestampNanos

FLUSH_ROWS = 100        # publish once this many rows are buffered ...
FLUSH_INTERVAL = 5.0    # ... or once this many seconds have passed


def example(host: str = 'localhost', port: int = 9000, total_rows=None):
    table_name = str(uuid.uuid1())
    try:
        with questdb.connect(
                f'ws::addr={host}:{port};auto_flush=off;') as db:
            with db.sender() as sender:
                # Auto-flush is disabled above so this example can implement
                # its own row-count / elapsed-time cadence.
                sent = 0
                last_flush = time.monotonic()
                print('Ctrl^C to terminate...')
                while total_rows is None or sent < total_rows:
                    time.sleep(random.randint(0, 750) / 1000)

                    sender.row(
                        table_name,
                        symbols={
                            'src': random.choice(('ALPHA', 'BETA', 'OMEGA')),
                            'dst': random.choice(('ALPHA', 'BETA', 'OMEGA'))},
                        columns={
                            'price': random.randint(200, 500),
                            'qty': random.randint(1, 5)},
                        at=TimestampNanos.now())
                    sent += 1

                    if (len(sender) >= FLUSH_ROWS or
                            time.monotonic() - last_flush >= FLUSH_INTERVAL):
                        print(f'Flushing {len(sender)} rows...')
                        sender.flush()
                        last_flush = time.monotonic()

                # Closing the lease publishes any remaining rows.
                print(f'table: {table_name}, total rows sent: {sent}')

    except KeyboardInterrupt:
        print('bye!')
    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    # Bounded run so the example terminates; pass total_rows=None to run
    # until Ctrl^C, as a real ticking-data loop would.
    example(total_rows=25)

Data Frames¶

Pandas Basics¶

The following example shows how to insert data from a Pandas DataFrame to the 'trades' table and run a generated query into Pandas.

import questdb
from questdb import QuestDBError

import sys
import pandas as pd


def example(host: str = 'localhost', port: int = 9000):
    df = pd.DataFrame({
            'symbol': pd.Categorical(['ETH-USD', 'BTC-USD']),
            'side': pd.Categorical(['sell', 'sell']),
            'price': [2615.54, 39269.98],
            'amount': [0.00044, 0.001],
            'timestamp': pd.to_datetime(['2021-01-01', '2021-01-02'])})
    try:
        with questdb.connect(f"ws::addr={host}:{port};") as db:
            # Ingress: publish a Pandas DataFrame into QuestDB.
            db.dataframe(
                df,
                table_name='trades',  # Table name to insert into.
                symbols=['symbol', 'side'],  # Columns to be inserted as SYMBOL types.
                at='timestamp')  # Column containing the designated timestamps.

            # Egress: query QuestDB and materialise the result as Pandas.
            with db.query(
                    "SELECT x AS trade_id, "
                    "x * 10.0 AS price "
                    "FROM long_sequence(3)") as result:
                queried = result.to_pandas()
            print(queried)

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

For details on all options, see the QuestDB.dataframe method.

`pd.Categorical` and multiple tables¶

The next example shows some more advanced features inserting data from Pandas and running a generated query into Pandas.

The data is sent to multiple tables: the columnar path loads one table per call, so the frame is split by its table-naming pd.Categorical column and each group is bulk-loaded with its own QuestDB.dataframe call.
Columns of type pd.Categorical are sent as SYMBOL types.

import questdb
from questdb import QuestDBError

import sys
import pandas as pd


def example(host: str = 'localhost', port: int = 9000):
    df = pd.DataFrame({
            'metric': pd.Categorical(
                ['humidity', 'temp_c', 'voc_index', 'temp_c']),
            'sensor': pd.Categorical(
                ['paris-01', 'london-02', 'london-01', 'paris-01']),
            'value': [
                0.83, 22.62, 100.0, 23.62],
            'ts': [
                pd.Timestamp('2022-08-06 07:35:23.189062'),
                pd.Timestamp('2022-08-06 07:35:23.189062'),
                pd.Timestamp('2022-08-06 07:35:23.189062'),
                pd.Timestamp('2022-08-06 07:35:23.189062')]})
    try:
        with questdb.connect(f"ws::addr={host}:{port};") as db:
            # Ingress: one DataFrame destined for multiple tables.
            # The columnar path loads one table per call, so split by the
            # table-naming column and bulk-load each group.
            for metric, group in df.groupby('metric', observed=True):
                db.dataframe(
                    group.drop(columns=['metric']),
                    table_name=str(metric),
                    symbols='auto',  # Category columns as SYMBOL. (Default)
                    at='ts')

            # Egress: query QuestDB and materialise the result as Pandas.
            with db.query(
                    "SELECT x AS sample_id, "
                    "x / 10.0 AS value "
                    "FROM long_sequence(3)") as result:
                queried = result.to_pandas()
            print(queried)

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

After running this example, the rows will be split across the 'humidity', 'temp_c' and 'voc_index' tables.

For details on all options, see the QuestDB.dataframe method.

Loading Pandas from a Parquet File¶

The following example shows how to load a Pandas DataFrame from a Parquet file and run a generated query into Pandas.

The example also relies on the dataframe’s index name to determine the table name.

import questdb
import pandas as pd


def write_parquet_file():
    df = pd.DataFrame({
        'location': pd.Categorical(
            ['BP-5541', 'UB-3355', 'SL-0995', 'BP-6653']),
        'provider': pd.Categorical(
            ['BP Pulse', 'Ubitricity', 'Source London', 'BP Pulse']),
        'speed_kwh': pd.Categorical(
            [50, 7, 7, 120]),
        'connector_type': pd.Categorical(
            ['Type 2 & 2+CCS', 'Type 1 & 2', 'Type 1 & 2', 'Type 2 & 2+CCS']),
        'current_type': pd.Categorical(
            ['dc', 'ac', 'ac', 'dc']),
        'price_pence':
            [54, 34, 32, 59],
        'in_use':
            [True, False, False, True],
        'ts': [
            pd.Timestamp('2022-12-30 12:15:00'),
            pd.Timestamp('2022-12-30 12:16:00'),
            pd.Timestamp('2022-12-30 12:18:00'),
            pd.Timestamp('2022-12-30 12:19:00')]})
    name = 'ev_chargers'
    df.index.name = name  # We set the dataframe's index name here!
    filename = f'{name}.parquet'
    df.to_parquet(filename)
    return filename


def example(host: str = 'localhost', port: int = 9000):
    filename = write_parquet_file()

    df = pd.read_parquet(filename)
    with questdb.connect(f"ws::addr={host}:{port};") as db:
        # Ingress: bulk-load a Pandas DataFrame into QuestDB.
        # Note: Table name is looked up from the dataframe's index name.
        db.dataframe(df, at='ts')

        # Egress: query QuestDB and materialise the result as Pandas.
        with db.query(
                "SELECT x AS charger_id, "
                "x * 25 AS speed_kwh "
                "FROM long_sequence(3)") as result:
            queried = result.to_pandas()
        print(queried)


if __name__ == '__main__':
    example()

For details on all options, see the QuestDB.dataframe method.

Polars¶

The following example ingests a Polars DataFrame over QWP/WebSocket via QuestDB.dataframe, runs a query into Polars, and includes a schema_overrides variant.

"""Polars DataFrame ingest and query example.

`QuestDB.dataframe()` accepts polars `DataFrame` and `LazyFrame` directly,
riding the Arrow PyCapsule Interface (`__arrow_c_stream__`) straight into
`column_sender_flush_arrow_batch`. `QuestDB.query()` can materialise query
results as a polars `DataFrame` with `QueryResult.to_polars()`.
"""

import questdb
from questdb import QuestDBError
import datetime
import sys


def example(host: str = 'localhost', port: int = 9000):
    import polars as pl

    df = pl.DataFrame({
        'symbol': ['ETH-USD', 'BTC-USD', 'ETH-USD'],
        'side':   ['sell', 'buy', 'buy'],
        'price':  [2615.54, 67234.12, 2620.88],
        'amount': [0.00044, 0.0012, 0.00033],
        'ts': [
            datetime.datetime(2025, 1, 1, 12, 0, 0),
            datetime.datetime(2025, 1, 1, 12, 0, 1),
            datetime.datetime(2025, 1, 1, 12, 0, 2),
        ],
    })

    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            # Ingress: publish a Polars DataFrame into QuestDB.
            db.dataframe(df, table_name='trades', at='ts')

            db.dataframe(
                df,
                table_name='trades_chunked',
                at='ts',
                max_rows_per_batch=2)

            # Egress: query QuestDB and materialise the result as Polars.
            with db.query(
                    "SELECT x AS trade_id, "
                    "x * 10.0 AS price, "
                    "timestamp_sequence("
                    "'2025-01-01T12:00:00.000000Z', 1000000) AS ts "
                    "FROM long_sequence(3)") as result:
                queried = result.to_polars()
            print(queried)
    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


def schema_overrides_example(host: str = 'localhost', port: int = 9000):
    """B-class wire types (IPv4 / Geohash / etc.) need an explicit hint
    because `arr.dtype` alone cannot disambiguate them from plain
    integer columns. `schema_overrides` injects the corresponding
    `questdb.*` Arrow Field metadata. Requires pyarrow.
    """
    import polars as pl

    df = pl.DataFrame({
        'addr': [0x0A000001, 0xC0A80101, 0x7F000001],
        'price': [100, 200, 300],
        'ts': [
            datetime.datetime(2025, 1, 1, 12, 0, 0),
            datetime.datetime(2025, 1, 1, 12, 0, 1),
            datetime.datetime(2025, 1, 1, 12, 0, 2),
        ],
    }, schema={'addr': pl.UInt32, 'price': pl.Int64, 'ts': pl.Datetime('us')})

    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            db.dataframe(
                df,
                table_name='ipv4_log',
                at='ts',
                schema_overrides={'addr': 'ipv4'})
    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()
    schema_overrides_example()

PyArrow¶

The following example ingests a PyArrow table over QWP/WebSocket via QuestDB.dataframe and runs a query into PyArrow.

"""PyArrow Table / RecordBatch ingest and query example.

`QuestDB.dataframe()` accepts any object exposing the Arrow PyCapsule
Interface (`__arrow_c_stream__`) — pyarrow Table, RecordBatch, DuckDB
relations, cudf, etc. — and routes through
`column_sender_flush_arrow_batch` one-shot. No per-column Cython
dispatch, no chunk lifecycle. `QuestDB.query()` can materialise query
results as a pyarrow `Table` with `QueryResult.to_arrow()`.
"""

import questdb
from questdb import QuestDBError
import sys


def example(host: str = 'localhost', port: int = 9000):
    import pyarrow as pa

    schema = pa.schema([
        pa.field('symbol', pa.string()),
        pa.field('side', pa.string()),
        pa.field('price', pa.float64()),
        pa.field('amount', pa.float64()),
        pa.field('ts', pa.timestamp('us')),
    ])
    table = pa.Table.from_pydict({
        'symbol': ['ETH-USD', 'BTC-USD'],
        'side':   ['sell', 'buy'],
        'price':  [2615.54, 67234.12],
        'amount': [0.00044, 0.0012],
        'ts':     [1735732800_000_000, 1735732801_000_000],
    }, schema=schema)

    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            # Ingress: publish a PyArrow Table into QuestDB.
            db.dataframe(table, table_name='trades', at='ts')

            # Egress: query QuestDB and materialise the result as PyArrow.
            with db.query(
                    "SELECT x AS trade_id, "
                    "x * 10.0 AS price, "
                    "timestamp_sequence("
                    "'2025-01-01T12:00:00.000000Z', 1000000) AS ts "
                    "FROM long_sequence(3)") as result:
                queried = result.to_arrow()
            print(queried)
    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


def schema_metadata_example(host: str = 'localhost', port: int = 9000):
    """B-class wire types can be selected either via `schema_overrides`
    (wrapper injects metadata for you) or by attaching the metadata
    directly on the pyarrow Field. Both are equivalent; this example
    shows the direct-attach form.
    """
    import pyarrow as pa

    schema = pa.schema([
        pa.field('addr', pa.uint32(),
                 metadata={b'questdb.column_type': b'ipv4'}),
        pa.field('loc', pa.int32(),
                 metadata={b'questdb.geohash_bits': b'20'}),
        pa.field('ts', pa.timestamp('us')),
    ])
    table = pa.Table.from_pydict({
        'addr': [0x0A000001, 0xC0A80101],
        'loc':  [0x12345, 0x67890],
        'ts':   [1735732800_000_000, 1735732801_000_000],
    }, schema=schema)

    try:
        conf = f'ws::addr={host}:{port};'
        with questdb.connect(conf) as db:
            db.dataframe(table, table_name='locations', at='ts')
    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()
    schema_metadata_example()

Decimal Types (QuestDB 9.2.0+)¶

The following example shows how to insert data with decimal precision using Python’s decimal.Decimal type, both row by row and as an object-dtype DataFrame column.

Requires QuestDB server 9.2.0+. DECIMAL columns must be pre-created (auto-creation not supported); the examples create them through query().

import sys
from decimal import Decimal

import pandas as pd

import questdb
from questdb import QuestDBError, TimestampNanos


def example(host: str = 'localhost', port: int = 9000):
    try:
        with questdb.connect(f'ws::addr={host}:{port};') as db:
            # DECIMAL columns must be created ahead of time; the server
            # does not auto-create them.
            db.execute(
                'CREATE TABLE IF NOT EXISTS financial_data ('
                '  symbol SYMBOL,'
                '  price DECIMAL(18, 6),'
                '  quantity DECIMAL(12, 4),'
                '  timestamp TIMESTAMP_NS'
                ') TIMESTAMP(timestamp) PARTITION BY DAY WAL')

            # Row-by-row ingestion with Python Decimal values.
            with db.sender() as sender:
                sender.row(
                    'financial_data',
                    symbols={'symbol': 'BTC-USD'},
                    columns={
                        'price': Decimal('50123.456789'),
                        'quantity': Decimal('1.2345')},
                    at=TimestampNanos.now())
                sender.flush(wait=True)

            # Bulk load with object-dtype Decimal columns.
            df = pd.DataFrame({
                'symbol': pd.Categorical(['BTC-USD', 'ETH-USD']),
                'price': [Decimal('50123.456789'), Decimal('2615.123456')],
                'quantity': [Decimal('1.2345'), Decimal('10.5678')],
            })
            db.dataframe(
                df, table_name='financial_data', symbols=['symbol'],
                at=TimestampNanos.now())

            frame = db.query(
                'SELECT symbol, price, quantity FROM financial_data '
                'LIMIT -3').to_pandas()
            print(frame)

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    example()

For better performance with DataFrames, use PyArrow decimal types, which carry width and scale in the column type:

import sys

import pandas as pd
import pyarrow as pa

import questdb
from questdb import QuestDBError, TimestampNanos


def example(host: str = 'localhost', port: int = 9000):
    # Arrow decimal columns carry their width and scale in the type,
    # so they map onto DECIMAL columns without per-cell inspection.
    df = pd.DataFrame({
        'symbol': pd.Categorical(['BTC-USD', 'ETH-USD']),
        'price': pd.Series(
            [50123.456789, 2615.123456],
            dtype=pd.ArrowDtype(pa.decimal128(18, 6))),
        'quantity': pd.Series(
            [1.2345, 10.5678],
            dtype=pd.ArrowDtype(pa.decimal128(12, 4))),
    })
    try:
        with questdb.connect(f'ws::addr={host}:{port};') as db:
            db.execute(
                'CREATE TABLE IF NOT EXISTS financial_data ('
                '  symbol SYMBOL,'
                '  price DECIMAL(18, 6),'
                '  quantity DECIMAL(12, 4),'
                '  timestamp TIMESTAMP_NS'
                ') TIMESTAMP(timestamp) PARTITION BY DAY WAL')

            db.dataframe(
                df, table_name='financial_data', symbols=['symbol'],
                at=TimestampNanos.now())

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')


if __name__ == '__main__':
    example()

Note

Over ws:: / wss:: decimal support is negotiated automatically. On the legacy ILP transports, HTTP/HTTPS auto-negotiates protocol version 3 while TCP/TCPS must configure it explicitly: tcp::addr=localhost:9009;protocol_version=3;.

For more details, see the QuestDB DECIMAL documentation.

Production¶

TLS, token auth, multi-host¶

The following production-shaped example combines TLS with OS trust roots, a bearer token, a multi-node addr list, a disk-backed store-and-forward spool, and a connection-event listener.

import sys

import questdb
from questdb import QuestDBError, TimestampNanos


def on_connection_event(event):
    print(event.kind, event.host, event.port, event.cause_msg,
          file=sys.stderr)


def example():
    # A production-shaped configuration: TLS with OS trust roots, a
    # bearer token, every node of the deployment in one addr list, and
    # a disk-backed store-and-forward spool that replays
    # unacknowledged frames after a process restart with the same
    # sender_id. For HTTP basic auth, replace the token with
    # 'username=...;password=...;'.
    conf = (
        'wss::addr=db-primary.example.com:9000,db-replica.example.com:9000;'
        'token=YOUR_BEARER_TOKEN;'
        'tls_ca=os_roots;'
        'sf_dir=/var/spool/questdb;'
        'sender_id=ingest-01;'
    )
    try:
        with questdb.connect(
                conf, connection_listener=on_connection_event) as db:
            info = db.server_info()
            print('connected to', info.role)

            with db.sender() as sender:
                sender.row(
                    'trades',
                    symbols={'symbol': 'ETH-USD'},
                    columns={'price': 2615.54, 'amount': 0.00044},
                    at=TimestampNanos.now())
                # Block until the server acknowledges everything
                # published on this lease.
                sender.flush(wait=True)

            print('events delivered:', db.connection_events_delivered)

    except QuestDBError as e:
        sys.stderr.write(
            f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')
        sys.exit(1)


if __name__ == '__main__':
    example()

Connection-level transports¶

The examples below use the standalone Sender — the connection-level API, where one sender drives exactly one connection. This layer carries the point-to-point capabilities the deployment-level handle does not: QWP/UDP datagrams, the legacy ILP transports (HTTP/TCP), and HTTP transactions. See Two API layers.

QWP over UDP¶

The following example sends a row using QuestWire Protocol over UDP.

Requires a QuestDB instance with QWP/UDP receiver support enabled. The default listener port is 9007.

from questdb import Sender, Protocol, QuestDBError, TimestampNanos
import sys


def example(
        host: str = 'localhost',
        port: int = 9007,
        table_name: str = 'trades'):
    try:
        with Sender(
                Protocol.Udp,
                host,
                port,
                max_datagram_size=1400) as sender:
            sender.row(
                table_name,
                symbols={
                    'symbol': 'ETH-USD',
                    'side': 'sell'},
                columns={
                    'price': 2615.54,
                    'amount': 0.00044,
                },
                at=TimestampNanos.now())

            # QWP/UDP defaults `auto_flush_bytes` to the datagram size.
            # Flush manually here to send the row immediately.
            sender.flush()

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

Legacy ILP (HTTP/TCP)¶

InfluxDB Line Protocol ingestion is fully supported through the standalone Sender — including HTTP transactions, which are unique to this layer.

HTTP with Token Auth¶

The following example connects to the database and sends two rows (lines).

The connection is made via HTTPS and uses token based authentication.

The data is sent at the end of the with block.

from questdb import Sender, QuestDBError, TimestampNanos
import sys


def example(host: str = 'localhost', port: int = 9000, token: str = None):
    try:
        if token is not None:
            # Bearer-token auth runs over TLS.
            conf = f'https::addr={host}:{port};token={token};'
        else:
            conf = f'http::addr={host}:{port};'
        with Sender.from_conf(conf) as sender:
            # Record with provided designated timestamp (using the 'at' param)
            # Notice the designated timestamp is expected in Nanoseconds,
            # but timestamps in other columns are expected in Microseconds.
            # The API provides convenient functions
            sender.row(
                'trades',
                symbols={
                    'symbol': 'ETH-USD',
                    'side': 'sell'},
                columns={
                    'price': 2615.54,
                    'amount': 0.00044,
                   },
                at=TimestampNanos.now())

            # You can call `sender.row` multiple times inside the same `with`
            # block. The client will buffer the rows and send them in batches.

            # You can flush manually at any point.
            sender.flush()

            # If you don't flush manually, the client will flush automatically
            # when a row is added and either:
            #   * The buffer contains 75000 rows (if HTTP) or 600 rows (if TCP)
            #   * The last flush was more than 1000ms ago.
            # Auto-flushing can be customized via the `auto_flush_..` params.

        # Any remaining pending rows will be sent when the `with` block ends.

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

HTTP Transactions¶

The following example sends rows atomically inside an ILP/HTTP transaction, a capability unique to the HTTP transport.

from questdb import Sender, TimestampNanos
import pandas as pd


def example():
    conf = 'http::addr=localhost:9000;'
    with Sender.from_conf(conf) as sender:
        # Force a whole dataframe to be written in a single transaction.
        # This temporarily suspends auto-flushing:
        # The dataframe-generated buffer must thus fit in memory.
        with sender.transaction('trades') as txn:
            df = pd.DataFrame({
                    'symbol': pd.Categorical(['ETH-USD', 'BTC-USD']),
                    'side': pd.Categorical(['sell', 'sell']),
                    'price': [2615.54, 39269.98],
                    'amount': [0.00044, 0.001],
                    'timestamp': pd.to_datetime(['2021-01-01', '2021-01-02'])})
            txn.dataframe(
                df,
                symbols=['symbol', 'side'],
                at='timestamp')

            # You can write additional dataframes or rows,
            # but they must all be for the same table.
            txn.row(
                symbols={'symbol': 'ETH-USD', 'side': 'sell'},
                columns={'price': 2615.54, 'amount': 0.00044},
                at=TimestampNanos.now())

        # The transaction is flushed when the `with` block ends.


if __name__ == '__main__':
    example()

TCP Basics¶

The following example writes rows over ILP/TCP with auto-flush.

from questdb import Sender, QuestDBError, TimestampNanos
import sys


def example(host: str = 'localhost', port: int = 9009):
    try:
        conf = f'tcp::addr={host}:{port};'
        with Sender.from_conf(conf) as sender:
            # Record with provided designated timestamp (using the 'at' param)
            # Notice the designated timestamp is expected in Nanoseconds,
            # but timestamps in other columns are expected in Microseconds.
            # The API provides convenient functions
            sender.row(
                'trades',
                symbols={
                    'symbol': 'ETH-USD',
                    'side': 'sell'},
                columns={
                    'price': 2615.54,
                    'amount': 0.00044,
                   },
                at=TimestampNanos.now())

            # You can call `sender.row` multiple times inside the same `with`
            # block. The client will buffer the rows and send them in batches.

            # You can flush manually at any point.
            sender.flush()

            # If you don't flush manually, the client will flush automatically
            # when a row is added and either:
            #   * The buffer contains 75000 rows (if HTTP) or 600 rows (if TCP)
            #   * The last flush was more than 1000ms ago.
            # Auto-flushing can be customized via the `auto_flush_..` params.

        # Any remaining pending rows will be sent when the `with` block ends.

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

TCP Authentication and TLS¶

Continuing from the previous example, the connection is authenticated and also uses TLS.

from questdb import Sender, QuestDBError, TimestampNanos
import sys


def example(host: str = 'localhost', port: int = 9009):
    try:
        conf = (
            f"tcps::addr={host}:{port};" +
            "username=testUser1;" +
            "token=5UjEMuA0Pj5pjK8a-fa24dyIf-Es5mYny3oE_Wmus48;" +
            "token_x=fLKYEaoEb9lrn3nkwLDA-M_xnuFOdSt9y0Z7_vWSHLU;" +
            "token_y=Dt5tbS1dEDMSYfym3fgMv0B99szno-dFc1rYF9t0aac;")
        with Sender.from_conf(conf) as sender:
            # Record with provided designated timestamp (using the 'at' param)
            # Notice the designated timestamp is expected in Nanoseconds,
            # but timestamps in other columns are expected in Microseconds.
            # The API provides convenient functions
            sender.row(
                'trades',
                symbols={
                    'symbol': 'ETH-USD',
                    'side': 'sell'},
                columns={
                    'price': 2615.54,
                    'amount': 0.00044,
                   },
                at=TimestampNanos.now())

            # You can call `sender.row` multiple times inside the same `with`
            # block. The client will buffer the rows and send them in batches.

            # You can flush manually at any point.
            sender.flush()

            # If you don't flush manually, the client will flush automatically
            # when a row is added and either:
            #   * The buffer contains 75000 rows (if HTTP) or 600 rows (if TCP)
            #   * The last flush was more than 1000ms ago.
            # Auto-flushing can be customized via the `auto_flush_..` params.

        # Any remaining pending rows will be sent when the `with` block ends.

    except QuestDBError as e:
        sys.stderr.write(f'Got error: {e}\n')


if __name__ == '__main__':
    example()

Further legacy ILP examples — ECDSA auth (tcp_auth.py) and configuration loading (http_from_conf.py) — live in the repository’s examples directory.

Examples¶

Getting started¶

connect() quick start¶

Queries¶

DataFrame bulk load and reader leases¶

Ticking data and flush cadence¶

Data Frames¶

Pandas Basics¶

`pd.Categorical` and multiple tables¶

Loading Pandas from a Parquet File¶

Polars¶

PyArrow¶

Decimal Types (QuestDB 9.2.0+)¶

Production¶

TLS, token auth, multi-host¶

Connection-level transports¶

QWP over UDP¶

Legacy ILP (HTTP/TCP)¶

HTTP with Token Auth¶

HTTP Transactions¶

TCP Basics¶

TCP Authentication and TLS¶

questdb

Table of Contents

This Page

Examples¶

Getting started¶

connect() quick start¶

Queries¶

DataFrame bulk load and reader leases¶

Ticking data and flush cadence¶

Data Frames¶

Pandas Basics¶

pd.Categorical and multiple tables¶

Loading Pandas from a Parquet File¶

Polars¶

PyArrow¶

Decimal Types (QuestDB 9.2.0+)¶

Production¶

TLS, token auth, multi-host¶

Connection-level transports¶

QWP over UDP¶

Legacy ILP (HTTP/TCP)¶

HTTP with Token Auth¶

HTTP Transactions¶

TCP Basics¶

TCP Authentication and TLS¶

`pd.Categorical` and multiple tables¶