ksql 0.10.2
A Python wrapper for the KSQL REST API
License: MIT License

ksql-python
===========

A python wrapper for the KSQL REST API. Easily interact with the KSQL REST API using this library.

Supported KSQLDB version: 0.10.1+
Supported Python version: 3.5+

.. image:: https://travis-ci.org/bryanyang0528/ksql-python.svg?branch=master
  :target: https://travis-ci.org/bryanyang0528/ksql-python

.. image:: https://codecov.io/gh/bryanyang0528/ksql-python/branch/master/graph/badge.svg
  :target: https://codecov.io/gh/bryanyang0528/ksql-python

.. image:: https://pepy.tech/badge/ksql
  :target: https://pepy.tech/project/ksql

.. image:: https://pepy.tech/badge/ksql/month
  :target: https://pepy.tech/project/ksql/month
  
.. image:: https://img.shields.io/badge/license-MIT-yellow.svg
  :target: https://github.com/bryanyang0528/ksql-python/blob/master/LICENSE  
  
Installation
------------

.. code:: bash

    pip install ksql

Or

.. code:: bash

    git clone https://github.com/bryanyang0528/ksql-python
    cd ksql-python
    python setup.py install

Getting Started
---------------

Setup for KSQL
~~~~~~~~~~~~~~~

This is the GITHUB page of KSQL. https://github.com/confluentinc/ksql

If you have installed open source Confluent CLI (e.g. by installing Confluent Open Source or Enterprise Platform), you can start KSQL and its dependencies with one single command:

.. code:: bash

    confluent start ksql-server

Setup for ksql-python API
~~~~~~~~~~~~~~~~~~~~~~~~~

-  Setup for the KSQL API:

.. code:: python

    from ksql import KSQLAPI
    client = KSQLAPI('http://ksql-server:8088')

- Setup for KSQl API with logging enabled:

.. code:: python

    import logging
    from ksql import KSQLAPI
    logging.basicConfig(level=logging.DEBUG)
    client = KSQLAPI('http://ksql-server:8088')

- Setup for KSQL API with Basic Authentication

.. code:: python

    from ksql import KSQLAPI
    client = KSQLAPI('http://ksql-server:8088', api_key="your_key", secret="your_secret")

Options
~~~~~~~

+---------------+-----------+------------+--------------------------------------------------------------+
| Option        | Type      | Required   | Description                                                  |
+===============+===========+============+==============================================================+
| ``url``       | string    | yes        | Your ksql-server url. Example: ``http://ksql-server:8080``   |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``timeout``   | integer   | no         | Timout for Requests. Default: ``5``                          |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``api_key``   | string    | no         | API Key to use on the requests                               |
+---------------+-----------+------------+--------------------------------------------------------------+
| ``secret``    | string    | no         | Secret to use on the requests                                |
+---------------+-----------+------------+--------------------------------------------------------------+

Main Methods
~~~~~~~~~~~~

ksql
^^^^

This method can be used for some KSQL features which are not supported via other specific methods like ``query``, ``create_stream`` or ``create_stream_as``.
The following example shows how to execute the ``show tables`` statement:

.. code:: python

    client.ksql('show tables')

-  Example Response ``[{'tables': {'statementText': 'show tables;', 'tables': []}}]``

query
^^^^^

It will execute sql query and keep listening streaming data.

.. code:: python

    client.query('select * from table1')

This command returns a generator. It can be printed e.g. by reading its values via `next(query)` or a for loop. Here is a complete example:

.. code:: python
    
  from ksql import KSQLAPI
  client = KSQLAPI('http://localhost:8088')
  query = client.query('select * from table1')
  for item in query: print(item)

-  Example Response

   ::

       {"row":{"columns":[1512787743388,"key1",1,2,3]},"errorMessage":null}
       {"row":{"columns":[1512787753200,"key1",1,2,3]},"errorMessage":null}
       {"row":{"columns":[1512787753488,"key1",1,2,3]},"errorMessage":null}
       {"row":{"columns":[1512787753888,"key1",1,2,3]},"errorMessage":null}

Query with HTTP/2
^^^^^^^^^^^^^^^^^
Execute queries with the new ``/query-stream`` endpoint. Documented `here <https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/streaming-endpoint/#executing-pull-or-push-queries>`_

To execute a sql query use the same syntax as the regular query, with the additional ``use_http2=True`` parameter.

.. code:: python

    client.query('select * from table1', use_http2=True)

A generator is returned with the following example response

   ::

       {"queryId":"44d8413c-0018-423d-b58f-3f2064b9a312","columnNames":["ORDER_ID","TOTAL_AMOUNT","CUSTOMER_NAME"],"columnTypes":["INTEGER","DOUBLE","STRING"]}
       [3,43.0,"Palo Alto"]
       [3,43.0,"Palo Alto"]
       [3,43.0,"Palo Alto"]

To terminate the query above use the ``close_query`` call.
Provide the ``queryId`` returned from the ``query`` call.

.. code:: python

    client.close_query("44d8413c-0018-423d-b58f-3f2064b9a312")

Insert rows into a Stream with HTTP/2
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Uses the new ``/inserts-stream`` endpoint. See `documentation <https://docs.ksqldb.io/en/0.10.0-ksqldb/developer-guide/ksqldb-rest-api/streaming-endpoint/#inserting-rows-into-an-existing-stream>`_

.. code:: python

    rows = [
            {"ORDER_ID": 1, "TOTAL_AMOUNT": 23.5, "CUSTOMER_NAME": "abc"},
            {"ORDER_ID": 2, "TOTAL_AMOUNT": 3.7, "CUSTOMER_NAME": "xyz"}
        ]

    results = self.api_client.inserts_stream("my_stream_name", rows)

An array of object will be returned on success, with the status of each row inserted.


Simplified API
~~~~~~~~~~~~~~

create_stream/ create_table
^^^^^^^^^^^^^^^^^^^^^^^^^^^

.. code:: python

    client.create_stream(table_name=table_name,
                         columns_type=columns_type,
                         topic=topic,
                         value_format=value_format)

Options
^^^^^^^

+-----------------+-----------+----------+--------------------------------------------------------------+
| Option          | Type      | Required | Description                                                  |
+=================+===========+==========+==============================================================+
| ``table_name``  | string    | yes      | name of stream/table                                         |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``columns_type``| list      | yes      | ex:``['viewtime bigint','userid varchar','pageid varchar']`` |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``topic``       | string    | yes      | Kafka topic                                                  |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``value_format``| string    | no       | ``JSON`` (Default) or ``DELIMITED`` or ``AVRO``              |
+-----------------+-----------+----------+--------------------------------------------------------------+
| ``key``         | string    | for Table| Key (used for JOINs)                                         |
+-----------------+-----------+----------+--------------------------------------------------------------+


-  Responses

:If create table/stream succeed:
  return True

:If failed:
  raise a CreateError(respose_from_ksql_server)

create_stream_as
^^^^^^^^^^^^^^^^

a simplified api for creating stream as select

.. code:: python

    client.create_stream_as(table_name=table_name,
                            select_columns=select_columns,
                            src_table=src_table,
                            kafka_topic=kafka_topic,
                            value_format=value_format,
                            conditions=conditions,
                            partition_by=partition_by,
                            **kwargs)


.. code:: sql

  CREATE STREAM <table_name>
  [WITH ( kafka_topic=<kafka_topic>, value_format=<value_format>, property_name=expression ... )]
  AS SELECT  <select_columns>
  FROM <src_table>
  [WHERE <conditions>]
  PARTITION BY <partition_by>];

Options
^^^^^^^

+-------------------+-----------+----------+--------------------------------------------------------------+
| Option            | Type      | Required | Description                                                  |
+===================+===========+==========+==============================================================+
| ``table_name``    | string    | yes      | name of stream/table                                         |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``select_columns``| list      | yes      | you can select ``[*]`` or ``['columnA', 'columnB']``         |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``src_table``     | string    | yes      | name of source table                                         |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``kafka_topic``   | string    | no       | The name of the Kafka topic of this new stream(table).       |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``value_format``  | string    | no       | ``DELIMITED``, ``JSON``(Default) or ``AVRO``                 |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``conditions``    | string    | no       | The conditions in the where clause.                          |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``partition_by``  | string    | no       | Data will be distributed across partitions by this column.   |
+-------------------+-----------+----------+--------------------------------------------------------------+
| ``kwargs``        | pair      | no       | please provide ``key=value`` pairs. Please see more options. |
+-------------------+-----------+----------+--------------------------------------------------------------+

KSQL JOINs
~~~~~~~~~~~~~~

KSQL JOINs between Streams and Tables are not supported yet via explicit methods, but you can use the ``ksql`` method for this like the following:

.. code:: python

    client.ksql("CREATE STREAM join_per_user WITH (VALUE_FORMAT='AVRO', KAFKA_TOPIC='join_per_user') AS SELECT Time, Amount FROM source c INNER JOIN users u on c.user = u.userid WHERE u.USERID = 1")

FileUpload
~~~~~~~~~~~~~~

upload
^^^^^^^^^^^^^^^^^^^^^^^^^^^
Run commands from a .ksql file. Can only support ksql commands and not streaming queries.

.. code:: python

     from ksql import FileUpload
     pointer = FileUpload('http://ksql-server:8080')
     pointer.upload('rules.ksql')


Options
^^^^^^^

+-----------------+-----------+----------+--------------------------------------------------------------+
| Option          | Type      | Required | Description                                                  |
+=================+===========+==========+==============================================================+
| ``ksqlfile``    | string    | yes      | name of file containing the rules                            |
+-----------------+-----------+----------+--------------------------------------------------------------+


-  Responses

:If ksql-commands succesfully executed:
  return (List of server response for all commands)

:If failed:
  raise the appropriate error

More Options
^^^^^^^^^^^^

There are more properties (partitions, replicas, etc...) in the official document.

`KSQL Syntax Reference <https://github.com/confluentinc/ksql/blob/0.1.x/docs/syntax-reference.md#syntax-reference>`_

-  Responses

:If create table/stream succeed:
  return True

:If failed:
  raise a CreatError(respose_from_ksql_server)