Python client for DB Service

The DB Service Python client provides a thin wrapper over the DB Service APIs, making it easier to connect to a running service and perform client operations from Python.

Use this client to create a session, run queries, manage tables, and interact with DB Service from Python.

Examples of using the client are included below. For a comprehensive list of parameters and their meanings, refer to the KDB-X DB Service API spec.

Requirements

• Python 3.x
• A running DB Service instance

Install

# Install the client
pip install --pre --index https://portal.dl.kx.com/assets/pypi kdbx-db-service-client

For KDB-X Python installation and environment setup, see the KDB-X Python install guide.

Connect

Create a session to connect to DB Service:

import dbservice_client as dbs
import pandas as pd  # Query examples return dataframes
from datetime import datetime as dt, timezone, timedelta

# Default endpoint: localhost:8080
session = dbs.Session()

# Optional explicit endpoint
rest_session = dbs.Session(endpoint="localhost:8080")

Selecting an assembly

Table, import, delete and configuration export calls are routed to an assembly with the assembly query parameter. A service configured with a single assembly routes there by default, so assembly only needs to be given when more than one assembly is configured. Set it once for the session, or per call to override the session default:

# Route every table, import, delete and configuration export call in this session to 'fxdb'
session = dbs.Session(assembly="fxdb")

# Route a single call elsewhere
session.list_tables(assembly="ratesdb")

Query calls (query_simple, query_sql, query_q, query_preview) do not take an assembly.

Examples

This section shows common DB Service Python client workflows, including table management, data import, querying, configuration export, deleting data, and deleting tables.

Managing Tables

Use these calls to define and inspect table schemas in DB Service.

# List tables (empty to begin with)
session.list_tables()

# Create partitioned table ('fxquote')
session.create_table(
    table="fxquote",
    type="partitioned",
    prtnCol="ts",
    sortColsDisk=["sym"],
    sortColsOrd=["sym"],
    columns=[
        {"name": "trddate", "type": "date"},
        {"name": "ts", "type": "timestamp"},
        {"name": "sym", "type": "symbol", "attrMem": "grouped", "attrDisk": "parted", "attrOrd": "parted"},
        {"name": "bid", "type": "float"},
        {"name": "ask", "type": "float"},
    ]
)

# List tables ('fxquote' table returned)
session.list_tables()

# Describe the 'fxquote' table
session.describe_table(table="fxquote")

Importing Data

DB Service supports both file-based and in-memory ingest. Any file you want to import must first be copied into the DB Service imports staging directory, for example: ~/.kx/db-service/data/imports/

Import CSV

# Import a CSV file into the existing 'fxquote' table
job = session.import_files(table="fxquote", path="fxquote.csv.gz")

# Check the status of the above import job
session.get_import(job_id=job["jobId"])

# Import a parquet file into the existing 'fxquote' table
session.import_files(table='fxquote', path='fxquote.parquet')

# Import a CSV file and create the 'instruments' table automatically if it does not exist
session.import_files(table="instruments", path="instruments.csv", createTable=True)

Import Kdb database

# Import the 'fxquote' HDB table from the root kdb+ database directory 'fxquote-hdb'
session.import_database(table="fxquote", path="fxquote-hdb")

Import Data

Users can import data directly from Python without file staging. By default, pandas DataFrames and PyKX table-like data use binary REST transport, while plain Python row/object payloads use JSON. Pass transport="json" or transport="binary" to force a specific transport.

# Objects payload imported to 'instruments' table
job = session.import_data(
    table="instruments",
    data=[
        {"instrumentid": 77, "sym": "USDBRL", "category": "EM", "decimals": 4, "pipdecimals": 4},
        {"instrumentid": 78, "sym": "USDKRW", "category": "EM", "decimals": 2, "pipdecimals": 2},
    ],
    insert_as="objects",
)

# Rows payload imported to the 'fxquote' table
session.import_data(
    table="fxquote",
    data=[
        ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
        ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
    ],
    columnNames=["trddate", "ts", "sym", "bid", "ask"],
    insert_as="rows",
)
# Note: for rows payload, columnNames are required.

Querying Tables

Run structured, SQL, or q queries against DB Service.

# Structured query
session.query_simple(
    table="fxquote",
    startTS="2026.03.02D00:00:00.000",
    endTS="2026.03.03D00:00:00.000",
    sortCols=["ts"],
    limit=5,
    return_as="json",
)

# SQL query
session.query_sql(
    query="SELECT * FROM instruments WHERE category LIKE 'EM'",
    return_as="pandas",
)

# QSQL query
session.query_q(
    query='select o:first bid,h:max bid,l:min bid,c:last bid by trddate,sym from fxquote',
    return_as="pandas",
)

# Preview (lightweight table sample)
session.query_preview(
    table="fxquote",
    limit=5,
    return_as="json",
)

Return format: return_as may be json, pandas, or pykx. If omitted, it defaults to json.

Exporting Configuration

Export the active assembly configuration as YAML, for single-node DB Service deployments.

# Return the assembly YAML as a string
assembly_yaml = session.export_assembly()
print(assembly_yaml)

# Save the assembly YAML to a file
session.export_assembly(filename="assembly.yaml")

# Export a specific assembly's configuration
session.export_assembly(assembly="ratesdb")

Deleting Data

Delete rows from a table over an optional time window and filter. Deletion is asynchronous: delete_rows returns a pending job, and the final outcome is read back with get_delete.

# Delete matching rows within a time window
job = session.delete_rows(
    table="fxquote",
    startTS="2026.03.02D00:00:00.000",
    endTS="2026.03.03D00:00:00.000",
    filter=[["=", "sym", "EURUSD"]],
)

# Check the status of the above delete job
session.get_delete(job["jobId"])

# Delete every row in the time window
session.delete_rows(
    table="fxquote",
    startTS="2026.03.02D00:00:00.000",
    endTS="2026.03.03D00:00:00.000",
    filter=[],
)

# Delete from a specific assembly
session.delete_rows(
    table="fxquote",
    startTS="2026.03.02D00:00:00.000",
    endTS="2026.03.03D00:00:00.000",
    filter=[["=", "sym", "EURUSD"]],
    assembly="ratesdb",
)

# Clear a delete job's tracked status
session.cancel_delete(job["jobId"])

filter is mandatory. Pass filter=[] to delete every row in the window. The explicit empty list is a safety catch, so a filter left off by accident can never widen a delete.

Omitting startTS/endTS deletes the whole table. The window defaults to all of time, so bound it unless you intend to remove every row. startTS is inclusive and endTS is exclusive.

Timestamps are wall-clock. startTS, endTS and any timestamp inside filter accept q literals or timezone-naive datetime objects. An aware datetime is rejected rather than silently shifted, because the delete API has no inputTZ field to carry the offset. Convert first, e.g. ts.astimezone(timezone.utc).replace(tzinfo=None).

cancel_delete does not roll back. It only clears the job's tracked status. A delete already applying to disk runs to completion.

Deleting Tables

# List tables (expected: 'fxquote' and 'instruments')
session.list_tables()

# Drop the 'fxquote' table
session.drop_table(table="fxquote")

# Drop the 'instruments' table
session.drop_table(table="instruments")

# List tables (expected: no tables)
session.list_tables()