API Reference¶
questdb.ingress¶
API for fast data ingestion into QuestDB.
- class questdb.ingress.Buffer¶
Bases:
object
Construct QuestDB-flavored InfluxDB Line Protocol (ILP) messages.
The
Buffer.row()
method is used to add a row to the buffer.You can call this many times.
from questdb.ingress import Buffer buf = Buffer() buf.row( 'table_name1', symbols={'s1', 'v1', 's2', 'v2'}, columns={'c1': True, 'c2': 0.5}) buf.row( 'table_name2', symbols={'questdb': '❤️'}, columns={'like': 100000}) # Append any additional rows then, once ready, call sender.flush(buffer) # a `Sender` instance. # The sender auto-cleared the buffer, ready for reuse. buf.row( 'table_name1', symbols={'s1', 'v1', 's2', 'v2'}, columns={'c1': True, 'c2': 0.5}) # etc.
- Buffer Constructor Arguments:
init_capacity
(int
): Initial capacity of the buffer in bytes. Defaults to65536
(64KiB).max_name_len
(int
): Maximum length of a column name. Defaults to127
which is the same default value as QuestDB. This should match thecairo.max.file.name.length
setting of the QuestDB instance you’re connecting to.
# These two buffer constructions are equivalent. buf1 = Buffer() buf2 = Buffer(init_capacity=65536, max_name_len=127)
To avoid having to manually set these arguments every time, you can call the sender’s
new_buffer()
method instead.from questdb.ingress import Sender, Buffer sender = Sender(host='localhost', port=9009, init_capacity=16384, max_name_len=64) buf = sender.new_buffer() assert buf.init_capacity == 16384 assert buf.max_name_len == 64
- __str__()¶
Return the constructed buffer as a string. Use for debugging.
- capacity() int ¶
The current buffer capacity.
- clear()¶
Reset the buffer.
Note that flushing a buffer will (unless otherwise specified) also automatically clear it.
This method is designed to be called only in conjunction with
sender.flush(buffer, clear=False)
.
- dataframe(df, *, table_name: str | None = None, table_name_col: None | int | str = None, symbols: str | bool | List[int] | List[str] = 'auto', at: None | int | str | TimestampNanos | datetime = None)¶
Add a pandas DataFrame to the buffer.
Also see the
Sender.dataframe()
method if you’re not using the buffer explicitly. It supports the same parameters and also supports auto-flushing.This feature requires the
pandas
,numpy
andpyarrow
package to be installed.- Parameters:
df (pandas.DataFrame) – The pandas DataFrame to serialize to the buffer.
table_name (str or None) –
The name of the table to which the rows belong.
If
None
, the table name is taken from thetable_name_col
parameter. If bothtable_name
andtable_name_col
areNone
, the table name is taken from the DataFrame’s index name (df.index.name
attribute).table_name_col (str or int or None) –
The name or index of the column in the DataFrame that contains the table name.
If
None
, the table name is taken from thetable_name
parameter. If bothtable_name
andtable_name_col
areNone
, the table name is taken from the DataFrame’s index name (df.index.name
attribute).If
table_name_col
is an integer, it is interpreted as the index of the column starting from0
. The index of the column can be negative, in which case it is interpreted as an offset from the end of the DataFrame. E.g.-1
is the last column.symbols (str or bool or list of str or list of int) –
The columns to be serialized as symbols.
If
'auto'
(default), all columns of dtype'categorical'
are serialized as symbols. IfTrue
, allstr
columns are serialized as symbols. IfFalse
, no columns are serialized as symbols.The list of symbols can also be specified explicitly as a
list
of column names (str
) or indices (int
). Integer indices start at0
and can be negative, offset from the end of the DataFrame. E.g.-1
is the last column.Only columns containing strings can be serialized as symbols.
at (TimestampNanos, datetime.datetime, int or str or None) –
The designated timestamp of the rows.
You can specify a single value for all rows or column name or index. If
None
, timestamp is assigned by the server for all rows. To pass in a timestamp explicity as an integer use theTimestampNanos
wrapper type. To get the current timestamp, useTimestampNanos.now()
. When passing adatetime.datetime
object, the timestamp is converted to nanoseconds. Adatetime
object is assumed to be in the local timezone unless one is specified explicitly (so calldatetime.datetime.now(tz=datetime.timezone.utc)
instead ofdatetime.datetime.utcnow()
for the current timestamp to avoid bugs).To specify a different timestamp for each row, pass in a column name (
str
) or index (int
, 0-based index, negative index supported): In this case, the column needs to be of dtypedatetime64[ns]
(assumed to be in the UTC timezone and not local, due to differences in Pandas and Python datetime handling) ordatetime64[ns, tz]
. When a timezone is specified in the column, it is converted to UTC automatically.A timestamp column can also contain
None
values. The server will assign the current timestamp to those rows.Note: All timestamps are always converted to nanoseconds and in the UTC timezone. Timezone information is dropped before sending and QuestDB will not store any timezone information.
Note: It is an error to specify both
table_name
andtable_name_col
.Note: The “index” column of the DataFrame is never serialized, even if it is named.
Example:
import pandas as pd import questdb.ingress as qi buf = qi.Buffer() # ... df = pd.DataFrame({ 'location': ['London', 'Managua', 'London'], 'temperature': [24.5, 35.0, 25.5], 'humidity': [0.5, 0.6, 0.45], 'ts': pd.date_range('2021-07-01', periods=3)}) buf.dataframe( df, table_name='weather', at='ts', symbols=['location']) # ... sender.flush(buf)
Pandas to ILP datatype mappings
Pandas Mappings¶ Pandas
dtype
Nulls
ILP Datatype
'bool'
N
BOOLEAN
'boolean'
N α
BOOLEAN
'object'
(bool
objects)N α
BOOLEAN
'uint8'
N
INTEGER
'int8'
N
INTEGER
'uint16'
N
INTEGER
'int16'
N
INTEGER
'uint32'
N
INTEGER
'int32'
N
INTEGER
'uint64'
N
INTEGER
β'int64'
N
INTEGER
'UInt8'
Y
INTEGER
'Int8'
Y
INTEGER
'UInt16'
Y
INTEGER
'Int16'
Y
INTEGER
'UInt32'
Y
INTEGER
'Int32'
Y
INTEGER
'UInt64'
Y
INTEGER
β'Int64'
Y
INTEGER
'object'
(int
objects)Y
INTEGER
β'float32'
γY (
NaN
)FLOAT
'float64'
Y (
NaN
)FLOAT
'object'
(float
objects)Y (
NaN
)FLOAT
'string'
(str
objects)Y
STRING
(default),SYMBOL
viasymbols
arg. δ'string[pyarrow]'
Y
STRING
(default),SYMBOL
viasymbols
arg. δ'category'
(str
objects) εY
SYMBOL
(default),STRING
viasymbols
arg. δ'object'
(str
objects)Y
STRING
(default),SYMBOL
viasymbols
arg. δ'datetime64[ns]'
Y
TIMESTAMP
ζ'datetime64[ns, tz]'
Y
TIMESTAMP
ζNote
α: Note some pandas dtypes allow nulls (e.g.
'boolean'
), where the QuestDB database does not.β: The valid range for integer values is -2^63 to 2^63-1. Any
'uint64'
,'UInt64'
or pythonint
object values outside this range will raise an error during serialization.γ: Upcast to 64-bit float during serialization.
δ: Columns containing strings can also be used to specify the table name. See
table_name_col
.ε: We only support categories containing strings. If the category contains non-string values, an error will be raised.
ζ: The ‘.dataframe()’ method only supports datetimes with nanosecond precision. The designated timestamp column (see
at
parameter) maintains the nanosecond precision, whilst values stored as columns have their precision truncated to microseconds. All dates are sent as UTC and any additional timezone information is dropped. If no timezone is specified, we follow the pandas convention of assuming the timezone is UTC. Datetimes before 1970-01-01 00:00:00 UTC are not supported. If a datetime value is specified asNone
(NaT
), it is interpreted as the current QuestDB server time set on receipt of message.
Error Handling and Recovery
In case an exception is raised during dataframe serialization, the buffer is left in its previous state. The buffer remains in a valid state and can be used for further calls even after an error.
For clarification, as an example, if an invalid
None
value appears at the 3rd row for abool
column, neither the 3rd nor the preceding rows are added to the buffer.Note: This differs from the
Sender.dataframe()
method, which modifies this guarantee due to itsauto_flush
logic.Performance Considerations
The Python GIL is released during serialization if it is not needed. If any column requires the GIL, the entire serialization is done whilst holding the GIL.
Column types that require the GIL are:
Columns of
str
,float
orint
orfloat
Python objects.The
'string[python]'
dtype.
- init_capacity¶
The initial capacity of the buffer when first created.
This may grow over time, see
capacity()
.
- max_name_len¶
Maximum length of a table or column name.
- reserve(additional: int)¶
Ensure the buffer has at least additional bytes of future capacity.
- Parameters:
additional (int) – Additional bytes to reserve.
- row(table_name: str, *, symbols: Dict[str, str | None] | None = None, columns: Dict[str, None | bool | int | float | str | TimestampMicros | datetime] | None = None, at: None | TimestampNanos | datetime = None)¶
Add a single row (line) to the buffer.
# All fields specified. buffer.row( 'table_name', symbols={'sym1': 'abc', 'sym2': 'def', 'sym3': None}, columns={ 'col1': True, 'col2': 123, 'col3': 3.14, 'col4': 'xyz', 'col5': TimestampMicros(123456789), 'col6': datetime(2019, 1, 1, 12, 0, 0), 'col7': None}, at=TimestampNanos(123456789)) # Only symbols specified. Designated timestamp assigned by the db. buffer.row( 'table_name', symbols={'sym1': 'abc', 'sym2': 'def'}) # Float columns and timestamp specified as `datetime.datetime`. # Pay special attention to the timezone, which if unspecified is # assumed to be the local timezone (and not UTC). buffer.row( 'sensor data', columns={ 'temperature': 24.5, 'humidity': 0.5}, at=datetime.datetime.now(tz=datetime.timezone.utc))
Python strings passed as values to
symbols
are going to be encoded as theSYMBOL
type in QuestDB, whilst Python strings passed as values tocolumns
are going to be encoded as theSTRING
type.Refer to the QuestDB documentation to understand the difference between the
SYMBOL
andSTRING
types (TL;DR: symbols are interned strings).Column values can be specified with Python types directly and map as so:
Python type
Serialized as ILP type
bool
int
float
str
datetime.datetime
andTimestampMicros
None
Column is skipped and not serialized.
If the destination table was already created, then the columns types will be cast to the types of the existing columns whenever possible (Refer to the QuestDB documentation pages linked above).
- Parameters:
table_name – The name of the table to which the row belongs.
symbols – A dictionary of symbol column names to
str
values. As a convenience, you can also pass aNone
value which will have the same effect as skipping the key: If the column already existed, it will be recorded asNULL
, otherwise it will not be created.columns – A dictionary of column names to
bool
,int
,float
,str
,TimestampMicros
ordatetime
values. As a convenience, you can also pass aNone
value which will have the same effect as skipping the key: If the column already existed, it will be recorded asNULL
, otherwise it will not be created.at – The timestamp of the row. If
None
, timestamp is assigned by the server. Ifdatetime
, the timestamp is converted to nanoseconds. A nanosecond unix epoch timestamp can be passed explicitly as aTimestampNanos
object.
- exception questdb.ingress.IngressError(code, msg)¶
Bases:
Exception
An error whilst using the
Sender
or constructing itsBuffer
.- __init__(code, msg)¶
- property code: IngressErrorCode¶
Return the error code.
- class questdb.ingress.IngressErrorCode(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)¶
Bases:
Enum
Category of Error.
- AuthError = 6¶
- BadDataFrame = 8¶
- CouldNotResolveAddr = 0¶
- InvalidApiCall = 1¶
- InvalidName = 4¶
- InvalidTimestamp = 5¶
- InvalidUtf8 = 3¶
- SocketError = 2¶
- TlsError = 7¶
- class questdb.ingress.Sender¶
Bases:
object
A sender is a client that inserts rows into QuestDB via the ILP protocol.
Inserting two rows
In this example, data will be flushed and sent at the end of the
with
block.with Sender('localhost', 9009) as sender: sender.row( 'weather_sensor', symbols={'id': 'toronto1'}, columns={'temperature': 23.5, 'humidity': 0.49}, at=TimestampNanos.now()) sensor.row( 'weather_sensor', symbols={'id': 'dubai2'}, columns={'temperature': 41.2, 'humidity': 0.34}, at=TimestampNanos.now())
The
Sender
object holds an internal buffer. The call to.row()
simply forwards all arguments to theBuffer.row()
method.Explicit flushing
An explicit call to
Sender.flush()
will send any pending data immediately.with Sender('localhost', 9009) as sender: sender.row( 'weather_sensor', symbols={'id': 'toronto1'}, columns={'temperature': 23.5, 'humidity': 0.49}, at=TimestampNanos.now()) sender.flush() sender.row( 'weather_sensor', symbols={'id': 'dubai2'}, columns={'temperature': 41.2, 'humidity': 0.34}, at=TimestampNanos.now()) sender.flush()
Auto-flushing (on by default, watermark at 63KiB)
To avoid accumulating very large buffers, the sender will flush the buffer automatically once its buffer reaches a certain byte-size watermark.
You can control this behavior by setting the
auto_flush
argument.# Never flushes automatically. sender = Sender('localhost', 9009, auto_flush=False) sender = Sender('localhost', 9009, auto_flush=None) # Ditto. sender = Sender('localhost', 9009, auto_flush=0) # Ditto. # Flushes automatically when the buffer reaches 1KiB. sender = Sender('localhost', 9009, auto_flush=1024) # Flushes automatically after every row. sender = Sender('localhost', 9009, auto_flush=True) sender = Sender('localhost', 9009, auto_flush=1) # Ditto.
Authentication and TLS Encryption
This implementation supports authentication and TLS full-connection encryption.
The
Sender(.., auth=..)
argument is a tuple of(kid, d, x, y)
as documented on the QuestDB ILP authentication documentation. Authentication is optional and disabled by default.The
Sender(.., tls=..)
argument is one of:False
: No TLS encryption (default).True
: TLS encryption, accepting all common certificates as recognized by either the webpki-roots Rust crate (which in turn relies on https://mkcert.org/), or the OS-provided certificate store.A
str
orpathlib.Path
: Path to a PEM-encoded certificate authority file. This is useful for testing with self-signed certificates.The special
'os_roots'
string: Use the OS-provided certificate store.The special
'webpki_roots'
string: Use the webpki-roots Rust crate to recognize certificates.The special
'webpki_and_os_roots'
string: Use both the webpki-roots Rust crate and the OS-provided certificate store to recognize certificates. (equivalent to True).The special
'insecure_skip_verify'
string: Dangerously disable all TLS certificate verification (do NOT use in production environments).
Positional constructor arguments for the Sender(..)
host
: Hostname or IP address of the QuestDB server.port
: Port number of the QuestDB server.
Keyword-only constructor arguments for the Sender(..)
interface
(str
): Network interface to bind to. Set this if you have an accelerated network interface (e.g. Solarflare) and want to use it.auth
(tuple
): Authentication tuple orNone
(default). See above for details.tls
(bool
,pathlib.Path
orstr
): TLS configuration orFalse
(default). See above for details.read_timeout
(int
): How long to wait for messages from the QuestDB server during the TLS handshake or authentication process. This field is expressed in milliseconds. The default is 15 seconds.init_capacity
(int
): Initial buffer capacity of the internal buffer. Default: 65536 (64KiB). See Buffer’s constructor for more details.max_name_length
(int
): Maximum length of a table or column name. See Buffer’s constructor for more details.auto_flush
(bool
orint
): Whether to automatically flush the buffer when it reaches a certain byte-size watermark. Default: 64512 (63KiB). See above for details.
- __enter__() Sender ¶
Call
Sender.connect()
at the start of awith
block.
- __exit__(exc_type, _exc_val, _exc_tb)¶
Flush pending and disconnect at the end of a
with
block.If the
with
block raises an exception, any pending data will NOT be flushed.This is implemented by calling
Sender.close()
.
- __str__()¶
Inspect the contents of the internal buffer.
The
str
value returned represents the unsent data.Also see
Sender.__len__()
.
- close(flush=True)¶
Disconnect.
This method is idempotent and can be called repeatedly.
Once a sender is closed, it can’t be re-used.
- Parameters:
flush (bool) – If
True
, flush the internal buffer before closing.
- connect()¶
Connect to the QuestDB server.
This method is synchronous and will block until the connection is established.
If the connection is set up with authentication and/or TLS, this method will return only after the handshake(s) is/are complete.
- dataframe(df, *, table_name: str | None = None, table_name_col: None | int | str = None, symbols: str | bool | List[int] | List[str] = 'auto', at: None | int | str | TimestampNanos | datetime = None)¶
Write a Pandas DataFrame to the internal buffer.
Example:
import pandas as pd import questdb.ingress as qi df = pd.DataFrame({ 'car': pd.Categorical(['Nic 42', 'Eddi', 'Nic 42', 'Eddi']), 'position': [1, 2, 1, 2], 'speed': [89.3, 98.2, 3, 4], 'lat_gforce': [0.1, -0.2, -0.6, 0.4], 'accelleration': [0.1, -0.2, 0.6, 4.4], 'tyre_pressure': [2.6, 2.5, 2.6, 2.5], 'ts': [ pd.Timestamp('2022-08-09 13:56:00'), pd.Timestamp('2022-08-09 13:56:01'), pd.Timestamp('2022-08-09 13:56:02'), pd.Timestamp('2022-08-09 13:56:03')]}) with qi.Sender('localhost', 9000) as sender: sender.dataframe(df, table_name='race_metrics', at='ts')
This method builds on top of the
Buffer.dataframe()
method. See its documentation for details on arguments.Additionally, this method also supports auto-flushing the buffer as specified in the
Sender
’sauto_flush
constructor argument. Auto-flushing is implemented incrementally, meanting that when callingsender.dataframe(df)
with a largedf
, the sender may have sent some of the rows to the server already whist the rest of the rows are going to be sent at the next auto-flush or next explicit call toSender.flush()
.In case of data errors with auto-flushing enabled, some of the rows may have been transmitted to the server already.
- flush(buffer=None, clear=True)¶
If called with no arguments, immediately flushes the internal buffer.
Alternatively you can flush a buffer that was constructed explicitly by passing
buffer
.The buffer will be cleared by default, unless
clear
is set toFalse
.This method does nothing if the provided or internal buffer is empty.
- Parameters:
buffer – The buffer to flush. If
None
, the internal buffer is flushed.clear – If
True
, the flushed buffer is cleared (default). IfFalse
, the flushed buffer is left in the internal buffer. Note thatclear=False
is only supported ifbuffer
is also specified.
The Python GIL is released during the network IO operation.
- init_capacity¶
The initial capacity of the sender’s internal buffer.
- max_name_len¶
Maximum length of a table or column name.
- new_buffer()¶
Make a new configured buffer.
The buffer is set up with the configured init_capacity and max_name_len.
- row(table_name: str, *, symbols: Dict[str, str] | None = None, columns: Dict[str, bool | int | float | str | TimestampMicros | datetime] | None = None, at: None | TimestampNanos | datetime = None)¶
Write a row to the internal buffer.
This may be sent automatically depending on the
auto_flush
setting in the constructor.Refer to the
Buffer.row()
documentation for details on arguments.
- class questdb.ingress.TimestampMicros¶
Bases:
object
A timestamp in microseconds since the UNIX epoch (UTC).
You may construct a
TimestampMicros
from an integer or adatetime.datetime
, or simply call theTimestampMicros.now()
method.# Recommended way to get the current timestamp. TimestampMicros.now() # The above is equivalent to: TimestampMicros(time.time_ns() // 1000) # You can provide a numeric timestamp too. It can't be negative. TimestampMicros(1657888365426838)
TimestampMicros
can also be constructed from adatetime.datetime
object.TimestampMicros.from_datetime( datetime.datetime.now(tz=datetime.timezone.utc))
We recommend that when using
datetime
objects, you explicitly pass in the timezone to use. This is becausedatetime
objects without an associated timezone are assumed to be in the local timezone and it is easy to make mistakes (e.g. passingdatetime.datetime.utcnow()
is a likely bug).- classmethod from_datetime(dt: datetime)¶
Construct a
TimestampMicros
from adatetime.datetime
object.
- classmethod now()¶
Construct a
TimestampMicros
from the current time as UTC.
- value¶
Number of microseconds (Unix epoch timestamp, UTC).
- class questdb.ingress.TimestampNanos¶
Bases:
object
A timestamp in nanoseconds since the UNIX epoch (UTC).
You may construct a
TimestampNanos
from an integer or adatetime.datetime
, or simply call theTimestampNanos.now()
method.# Recommended way to get the current timestamp. TimestampNanos.now() # The above is equivalent to: TimestampNanos(time.time_ns()) # You can provide a numeric timestamp too. It can't be negative. TimestampNanos(1657888365426838016)
TimestampNanos
can also be constructed from adatetime
object.TimestampNanos.from_datetime( datetime.datetime.now(tz=datetime.timezone.utc))
We recommend that when using
datetime
objects, you explicitly pass in the timezone to use. This is becausedatetime
objects without an associated timezone are assumed to be in the local timezone and it is easy to make mistakes (e.g. passingdatetime.datetime.utcnow()
is a likely bug).- classmethod from_datetime(dt: datetime)¶
Construct a
TimestampNanos
from adatetime.datetime
object.
- classmethod now()¶
Construct a
TimestampNanos
from the current time as UTC.
- value¶
Number of nanoseconds (Unix epoch timestamp, UTC).