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.Categoricalcolumn and each group is bulk-loaded with its ownQuestDB.dataframecall.Columns of type
pd.Categoricalare sent asSYMBOLtypes.
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.
"""
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()
UUID, IPV4, BINARY, CHAR, DATE, LONG256, GEOHASH (QuestDB 10.0.0+)¶
The following example writes all seven column types row by row. UUID,
IPV4 and BINARY take the Python types that already mean them; the
other four take the Char,
DateMillis,
Long256 and Geohash
wrappers.
import datetime
import ipaddress
import sys
import time
import uuid
import questdb
from questdb import (
Char,
DateMillis,
Geohash,
Long256,
QuestDBError,
TimestampNanos,
)
def _query_when_applied(db, sql, attempts=100):
"""Rows land through the WAL, so a read issued straight after the
write acknowledges can still be ahead of them."""
frame = db.query(sql).to_pandas()
for _ in range(attempts):
if not frame.empty:
break
time.sleep(0.05)
frame = db.query(sql).to_pandas()
return frame
def example(
host: str = 'localhost',
port: int = 9000,
table_name: str = 'devices'):
with questdb.connect(f'ws::addr={host}:{port};') as db:
# These column types are auto-created like any other, but a
# GEOHASH column's precision is fixed when the column is
# created, so this table states it up front.
db.execute(
f'CREATE TABLE IF NOT EXISTS {table_name} ('
' device_id UUID,'
' address IPV4,'
' payload BINARY,'
' grade CHAR,'
' last_seen DATE,'
' checksum LONG256,'
' location GEOHASH(5c),'
' timestamp TIMESTAMP_NS'
') TIMESTAMP(timestamp) PARTITION BY DAY WAL')
with db.sender() as sender:
sender.row(
table_name,
columns={
# UUID, IPV4 and BINARY are written with the
# ordinary Python types.
'device_id': uuid.UUID(
'123e4567-e89b-12d3-a456-426614174000'),
'address': ipaddress.IPv4Address('10.0.0.7'),
'payload': b'\x00\x01\x02\x03',
# The rest need a wrapper, because the Python
# type they would otherwise arrive as already
# means another QuestDB column type.
'grade': Char('A'),
# A DATE is a millisecond timestamp rather than
# a civil date; `DateMillis.now()` and
# `DateMillis(millis)` build one too.
'last_seen': DateMillis.from_datetime(
datetime.datetime(
2024, 1, 2, 3, 4, 5, 678000,
tzinfo=datetime.timezone.utc)),
'checksum': Long256(0xdeadbeef),
# 5 characters of base32, so 25 bits of
# precision -- the GEOHASH(5c) the column holds.
'location': Geohash.from_string('u33d8'),
},
at=TimestampNanos.now())
sender.flush(wait=True)
print(_query_when_applied(
db,
'SELECT device_id, address, grade, last_seen, checksum, '
f'location FROM {table_name} LIMIT -1'))
if __name__ == '__main__':
try:
example()
except QuestDBError as e:
sys.stderr.write(
f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')
sys.exit(1)
The same types from a DataFrame, stated through the column’s Arrow type
and through schema_overrides, and then read back and written out
again on the strength of df.attrs['questdb'] alone:
import sys
import time
import uuid
import pandas as pd
import pyarrow as pa
import questdb
from questdb import QuestDBError, TimestampNanos
def _query_when_applied(db, sql, attempts=100):
"""Rows land through the WAL, so a read issued straight after the
write acknowledges can still be ahead of them."""
frame = db.query(sql).to_pandas()
for _ in range(attempts):
if not frame.empty:
break
time.sleep(0.05)
frame = db.query(sql).to_pandas()
return frame
def example(
host: str = 'localhost',
port: int = 9000,
table_name: str = 'devices_df'):
with questdb.connect(f'ws::addr={host}:{port};') as db:
db.execute(
f'CREATE TABLE IF NOT EXISTS {table_name} ('
' device_id UUID,'
' address IPV4,'
' grade CHAR,'
' last_seen DATE,'
' checksum LONG256,'
' location GEOHASH(5c),'
' timestamp TIMESTAMP_NS'
') TIMESTAMP(timestamp) PARTITION BY DAY WAL')
device_ids = [
uuid.UUID('123e4567-e89b-12d3-a456-426614174000'),
uuid.UUID('00000000-0000-0000-0000-0000000000ff')]
frame = pd.DataFrame({
# A DataFrame states a type through the column's own
# Arrow type where the type is specific enough to say
# so, and through `schema_overrides` where it is not.
# Only `pa.uuid()` names a type outright here: a
# fixed-size binary of 16 or 32 bytes does not say
# whether it holds a UUID, a LONG256 or plain bytes, and
# a DataFrame refuses one that nothing labels; uint32 and
# uint16 are plain integers. Each of those is named
# below.
'device_id': pd.Series(
[value.bytes for value in device_ids],
dtype=pd.ArrowDtype(pa.binary(16))),
'address': pd.Series(
[0x0A000007, 0x0A000008],
dtype=pd.ArrowDtype(pa.uint32())),
'grade': pd.Series(
[ord('A'), ord('B')],
dtype=pd.ArrowDtype(pa.uint16())),
# A millisecond Arrow timestamp is itself the DATE
# claim; `pa.date32()` and `pa.date64()` work too.
'last_seen': pd.Series(
[1704164645678, 1704164645679],
dtype=pd.ArrowDtype(pa.timestamp('ms'))),
'checksum': pd.Series(
[(0xdeadbeef).to_bytes(32, 'little'),
(1).to_bytes(32, 'little')],
dtype=pd.ArrowDtype(pa.binary(32))),
# A GEOHASH rides on a signed integer wide enough for
# its precision, and `schema_overrides` names the bits.
# 27364744 is the base32 string 'u33d8' packed into 25
# bits -- `Geohash.from_string` does that conversion for
# the row API.
'location': pd.Series(
[27364744, 27364745],
dtype=pd.ArrowDtype(pa.int32())),
})
db.dataframe(
frame,
table_name=table_name,
schema_overrides={
'device_id': 'uuid',
'address': 'ipv4',
'grade': 'char',
'checksum': 'long256',
'location': ('geohash', 25),
},
at=TimestampNanos.now())
# Reading a result back and writing it out again keeps
# every one of these types. A pandas dtype holds an Arrow
# type and no field, so the claim travels in
# `df.attrs['questdb']`, which `dataframe()` reads. It works
# the same on all three `to_pandas` backends, and needs no
# `schema_overrides` second time round.
read_back = _query_when_applied(
db, f'SELECT * FROM {table_name} LIMIT -2')
print(read_back.attrs['questdb'])
# Written back into the table it came from: the server
# refuses a column whose type does not match the one the
# table holds, so this append landing is the claim doing its
# job. Written to a new table name instead, the same frame
# auto-creates it with these types rather than with BINARY
# and plain integers.
db.dataframe(read_back, table_name=table_name, at='timestamp')
print(_query_when_applied(
db,
'SELECT device_id, address, grade, last_seen, checksum, '
f'location FROM {table_name}'))
if __name__ == '__main__':
try:
example()
except QuestDBError as e:
sys.stderr.write(
f'Got error: {e} (code={e.code}, in_doubt={e.in_doubt})\n')
sys.exit(1)
Note
These types need QuestDB server 10.0.0 or later – the first
production QWP implementation – and a QWP sender (udp::,
ws:: or wss::). The legacy ILP transports have no
such column types: over them a uuid.UUID or an
ipaddress.IPv4Address is rejected, and so are the four wrappers.
The DataFrame route is narrower still: it runs on ws:: / wss::
only, because it is
QuestDB.dataframe that writes these
columns and questdb.connect() takes no other scheme. On udp::
they go one row at a time through
Sender.row.
See UUID, IPV4, BINARY, CHAR, DATE, LONG256 and GEOHASH Columns.
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.