Metadata-Version: 2.4
Name: jetblack-ksql-dbapi
Version: 0.1.0
Summary: A vanilla and asyncio DBAPI interface for ksql
Author-email: Rob Blackbourn <rob.blackbourn@gmail.com>
Project-URL: Repository, https://github.com/rob-blackbourn/jetblack-ksql-dbapi
Project-URL: Issues, https://github.com/rob-blackbourn/jetblack-ksql-dbapi/issues
Description-Content-Type: text/markdown
Requires-Dist: lark<2,>=1
Provides-Extra: httpx
Requires-Dist: httpx[http2]<1,>=0.28.1; extra == "httpx"
Provides-Extra: httpx2
Requires-Dist: httpx2[http2]<3,>=2.0.0; extra == "httpx2"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pylint; extra == "dev"
Requires-Dist: mypy; extra == "dev"

# jetblack-ksql-dbapi

A vanilla and asyncio ksql DBAPI interface for Python >= 3.12.

## Status

This is work in progress.

## Installation

The package uses either httpx or httpx2. This can either
be installed separately or specified.

```bash
pip install jetblack-ksql-dbapi[httpx2]
```

## Usage

There is a docker compose file in the scripts folder which will bring
up a local instance of ksql.

Here is an example using the async client connecting to a local instance of ksql.

```python
import asyncio

import jetblack_ksql_dbapi.aio as ksql
from jetblack_ksql_dbapi.aio import Connection


async def main() -> None:

    conn = ksql.connect("http://localhost:8088")

    cur = conn.cursor()

    # Drop the tables if they exist.
    await cur.execute(
        "DROP TABLE IF EXISTS user_view DELETE TOPIC;"
    )
    await cur.execute(
        "DROP TABLE IF EXISTS user DELETE TOPIC;"
    )

    # Create the tables.
    await cur.execute(
        """\
CREATE TABLE user
(
    user_id BIGINT  PRIMARY KEY,
    username        STRING,
    created         TIMESTAMP,
    age             DECIMAL(3, 0)
) WITH (
    kafka_topic='user',
    value_format='json',
    key_format='json',
    partitions=1
);
"""
    )

    await cur.execute(
        """\
CREATE TABLE user_view AS SELECT * FROM user;
"""
    )

    # Insert some data.
    await cur.executemany(
        """\
INSERT INTO user(user_id, username, created, age)
VALUES (?, ?, ?, ?);
""",
        (
            (1, 'tom', '2026-07-28T12:03:24', 42),
            (2, 'dick', '2026-07-28T12:03:24', 42),
            (3, 'harry', '2026-07-28T12:03:24', 42)
        )
    )

    await cur.execute(
        "SELECT * FROM user_view;"
    )
    async for row in cur:
        print(row)


if __name__ == "__main__":
    asyncio.run(main())
```
