GitHub - rob-blackbourn/jetblack-ksql-dbapi: A python dbapi interface for ksqldb · GitHub
/" data-turbo-transient="true" />
Skip to content
Type / to search
Sign in<br>Sign upAppearance settings
You signed in with another tab or window. Reload to refresh your session.<br>You signed out in another tab or window. Reload to refresh your session.<br>You switched accounts on another tab or window. Reload to refresh your session.
Dismiss alert
{{ message }}
rob-blackbourn
jetblack-ksql-dbapi
Public
Notifications<br>You must be signed in to change notification settings
Fork
Star
main
BranchesTags
Go to file
CodeOpen more actions menu
Folders and files<br>NameNameLast commit message<br>Last commit date<br>Latest commit
History<br>77 Commits<br>77 Commits
examples
examples
scripts
scripts
src/jetblack_ksql_dbapi
src/jetblack_ksql_dbapi
tests
tests
.gitignore
.gitignore
LICENSE
LICENSE
README.md
README.md
pyproject.toml
pyproject.toml
View all files
Repository files navigation
jetblack-ksql-dbapi
A vanilla and asyncio ksql<br>DBAPI interface for Python >= 3.12.
Status
This is work in progress.
Installation
The package uses either httpx or httpx2. This can either<br>be installed separately or specified as an extra.
pip install jetblack-ksql-dbapi[httpx2]
Usage
In the source repo there is a docker compose file in the scripts folder which will bring<br>up a local instance of ksql.
Here is an example using the async client connecting to a local instance of ksql.
None:
conn = connect("http://localhost:8088")
cur = conn.cursor()
# Drop the tables if they exist.<br>await cur.execute(<br>"DROP TABLE IF EXISTS user_view DELETE TOPIC;"<br>await cur.execute(<br>"DROP TABLE IF EXISTS user DELETE TOPIC;"
# Create the tables.<br>await cur.execute(<br>"""\<br>CREATE TABLE user<br>user_id BIGINT PRIMARY KEY,<br>username STRING,<br>created TIMESTAMP,<br>age DECIMAL(3, 0)<br>) WITH (<br>kafka_topic='user',<br>value_format='json',<br>key_format='json',<br>partitions=1<br>);<br>"""
await cur.execute(<br>"""\<br>CREATE TABLE user_view AS SELECT * FROM user;<br>"""
# Insert some data.<br>await cur.executemany(<br>"""\<br>INSERT INTO user(user_id, username, created, age)<br>VALUES (?, ?, ?, ?);<br>""",<br>(1, 'tom', '2026-07-28T12:03:24', 42),<br>(2, 'dick', '2026-07-28T12:03:24', 42),<br>(3, 'harry', '2026-07-28T12:03:24', 42)
await cur.execute("SELECT * FROM user_view;")<br>async for row in cur:<br>print(row)
if __name__ == "__main__":<br>asyncio.run(main())">import asyncio
from jetblack_ksql_dbapi.aio import connect
async def main() -> None:
conn = connect("http://localhost:8088")
cur = conn.cursor()
# Drop the tables if they exist.<br>await cur.execute(<br>"DROP TABLE IF EXISTS user_view DELETE TOPIC;"<br>await cur.execute(<br>"DROP TABLE IF EXISTS user DELETE TOPIC;"
# Create the tables.<br>await cur.execute(<br>"""\<br>CREATE TABLE user<br>user_id BIGINT PRIMARY KEY,<br>username STRING,<br>created TIMESTAMP,<br>age DECIMAL(3, 0)<br>) WITH (<br>kafka_topic='user',<br>value_format='json',<br>key_format='json',<br>partitions=1<br>);<br>"""
await cur.execute(<br>"""\<br>CREATE TABLE user_view AS SELECT * FROM user;<br>"""
# Insert some data.<br>await cur.executemany(<br>"""\<br>INSERT INTO user(user_id, username, created, age)<br>VALUES (?, ?, ?, ?);<br>""",<br>(1, 'tom', '2026-07-28T12:03:24', 42),<br>(2, 'dick', '2026-07-28T12:03:24', 42),<br>(3, 'harry', '2026-07-28T12:03:24', 42)
await cur.execute("SELECT * FROM user_view;")<br>async for row in cur:<br>print(row)
if __name__ == "__main__":<br>asyncio.run(main())
Things to do
Figure out what to do with the paramstyle global.
How to handle timeouts.
Do something useful with the output of commands like SHOW TABLES;
Handle multiple commands.
Tidy up cursors with multiple executions.
About<br>A python dbapi interface for ksqldb<br>Topics<br>asyncioasyncio-pythondbapidbapi2kafkaksqlksqldbksqldb-clientpython<br>Resources<br>Readme<br>Apache-2.0 license<br>Activity<br>Stars<br>0 stars<br>Watchers<br>0 watching<br>Forks<br>0 forks<br>Report repository
Releases
Packages
Contributors
Languages
You can’t perform that action at this time.