semvec/pypi-packages huggingface dataset:
confluent-ksqldb 0.1.1
Python client for ksqlDB
License: Apache-2.0
Requires Python: >=3.7,<4.0

# confluent-ksqldb-python [![](https://img.shields.io/badge/python-3.7+-blue.svg)](https://www.python.org/downloads/) [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)

A Python client for [ksqlDB](https://ksqldb.io/).

## Installation

Use the package manager [pip](https://pip.pypa.io/en/stable/) to install `confluent-ksqldb`.

```bash
pip install confluent-ksqldb
```

## Usage

#### Streamed Query:

```python
import asyncio

from ksqldb.client import Client
from ksqldb.row import Row
from ksqldb.streamed_query_result import StreamedQueryResult


async def main():
    # Initialize the client with the host and port
    client: Client = Client(host='localhost', port=8088)

    """
    Execute a Push query.
     
    client.stream_query_async() is appropriate for queries which may not run 
    to immediate completion, so this works well for both pull and push queries.
    """
    result: StreamedQueryResult = await client.stream_query_async('select * from RATINGS EMIT CHANGES;')
    print(f'Got result query id: {result.query_id}')
    print(f'Got result schema: {result.schema}')

    while not result.is_complete():
        # Poll for a row
        row: Row = await result.poll_async()
        print(f'Row: {row}')

    # Close the client and dispose of any resources that it uses
    await client.close()


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

#### Batched Query:

```python
import asyncio

from ksqldb.client import Client
from ksqldb.batched_query_result import BatchedQueryResult


async def main():
    # Initialize the client with the host and port
    client: Client = Client(host='localhost', port=8088)

    """
    Execute a Pull query. 
    
    client.execute_query_async() is appropriate for queries which can run to 
    completion in a batch, namely pull queries and push queries with LIMIT statements.
    """
    result: BatchedQueryResult = await client.execute_query_async('select * from RATINGS LIMIT 3;')
    print(f'Got result query id: {result.query_id}')
    print(f'Got result schema: {result.schema}')

    # Print the rows received
    for row in result.rows:
        print(f'Row: {row}')

    # Close the client and dispose of any resources that it uses
    await client.close()


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

## Contributing
Pull requests are welcome. For major changes, please open an issue first to discuss what you would like to change.

Please make sure to update tests as appropriate.

## License
[Apache License 2.0](https://choosealicense.com/licenses/apache-2.0/)