init

.gitignore

@@ -0,0 +1,101 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule
+
+# SageMath parsed files
+*.sage.py
+
+# dotenv
+.env
+
+# virtualenv
+.venv
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/

.travis.yml

@@ -0,0 +1,13 @@
+language: python
+python:
+  - "2.7"
+  - "3.3"
+  - "3.4"
+  - "3.5"
+  - "3.6"
+# command to install dependencies
+install:
+  - pip install nose vcrpy
+  - pip install -r requirements.txt
+# command to run tests
+script: nosetests

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2017 Bryan Yang
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

ksql/__init__.py

@@ -0,0 +1,6 @@
+
+__package_name__ = "ksql"
+__version__ = "0.2"
+
+
+from ksql.client import KSqlAPI

ksql/client.py

@@ -0,0 +1,76 @@
+import json
+import requests
+
+class KSqlAPI(object):
+    """ API Class """
+
+    def __init__(self, url, **kwargs):
+        self.url = url
+        self.timeout = kwargs.get("timeout", 5)
+
+    def get_url(self):
+        return self.url
+
+    def get_ksql_version(self):
+        r = requests.get(self.url)
+        if r.status_code == 200:
+            info = r.json().get('KSQL Server Info')
+            version = info.get('version')
+            return version
+        else:
+            raise ValueError('Status Code: {}.\nMessage: {}'.format(r.status_code, r.content))
+
+    def __request(self, endpoint, method='post', sql_string=''):
+        url = '{}/{}'.format(self.url, endpoint)
+
+        sql_string = self.__validate_sql_string(sql_string) 
+        print(sql_string)
+        data = json.dumps({
+            "ksql": sql_string
+        })
+
+        headers = {
+            "Content-Type": "application/json"
+        }
+
+        if endpoint == 'query':
+            stream = True
+        else:
+            stream = False
+
+        r = requests.request(
+            method=method,
+            url=url,
+            data=data,
+            timeout=self.timeout,
+            headers=headers,
+            stream=stream)  
+
+        return r  
+
+    @staticmethod
+    def __validate_sql_string(sql_string):
+        if len(sql_string) > 0:
+            if sql_string[-1] != ';':
+                sql_string += ';'
+        return sql_string
+
+    def ksql(self, ksql_string):
+        r = self.__request(endpoint='ksql', sql_string=ksql_string)
+
+        if r.status_code == 200:
+            r = r.json()
+            return r
+        else: 
+            raise ValueError('Status Code: {}.\nMessage: {}'.format(r.status_code, r.content))
+
+    def query(self, query_string, encoding='utf-8', chunk_size=128):
+        """  
+        Process streaming incoming data.
+
+        """
+        r = self.__request(endpoint='query', sql_string=query_string)
+
+        for chunk in r.iter_content(chunk_size=chunk_size):
+            if chunk != b'\n':
+                print(chunk.decode(encoding))

requirements.txt

@@ -0,0 +1,3 @@
+requests
+six
+urllib3

setup.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+""" Setup module """
+import os
+import re
+
+from setuptools import setup
+from pip.req import parse_requirements
+
+# Get version from __init__.py file
+VERSION = ""
+with open("ksql/__init__.py", "r") as fd:
+    VERSION = re.search(r"^__version__\s*=\s*['\"]([^\"]*)['\"]", fd.read(), re.MULTILINE).group(1)
+
+if not VERSION:
+    raise RuntimeError("Cannot find version information")
+
+here = os.path.dirname(__file__)
+
+# Get long description
+README = open(os.path.join(here, "README.md")).read()
+
+reqs = [str(x.req) for x in parse_requirements(os.path.join(here,'requirements.txt'), session='hack')]
+
+# allow setup.py to be run from any path
+os.chdir(os.path.normpath(os.path.join(os.path.abspath(__file__), os.pardir)))
+
+setup(
+    name="ksql",
+    version=VERSION,
+    description="A Python wrapper for the KSql REST API",
+    long_description=README,
+    author="Bryan Yang @ Vpon",
+    url="https://github.com/bryanyang0528/ksql-python",
+    license="MIT License",
+    packages=[
+        "ksql"
+    ],
+    include_package_data=True,
+    platforms=['any'],
+    install_requires=reqs,
+)

tests/test_client.py

@@ -0,0 +1,48 @@
+import json
+import unittest
+import requests
+
+import vcr
+
+from ksql import KSqlAPI
+import ksql
+
+class TestKSqlAPI(unittest.TestCase):
+    """Test case for the client methods."""
+
+    def setUp(self):
+        self.url = "http://ksql-server:8080"
+        self.api_client = KSqlAPI(url=self.url)
+
+    def test_with_timeout(self):
+        api_client = KSqlAPI(url='http://foo', timeout=10)
+        self.assertEquals(api_client.timeout, 10)
+
+    @vcr.use_cassette('tests/vcr_cassettes/get_ksql_server.yml')
+    def test_get_ksql_version_success(self):
+        """ Test GET requests """
+        version = self.api_client.get_ksql_version()
+        self.assertEqual(version, '0.2')
+
+    @vcr.use_cassette('tests/vcr_cassettes/ksql_show_table.yml')
+    def test_ksql_show_tables(self):
+        """ Test GET requests """
+        ksql_string = "show tables;"
+        r = self.api_client.ksql(ksql_string)
+        self.assertEqual(r, [{'tables': {'statementText': 'show tables;', 'tables': []}}])
+
+    @vcr.use_cassette('tests/vcr_cassettes/ksql_show_table.yml')
+    def test_ksql_show_tables_with_no_semicolon(self):
+        """ Test GET requests """
+        ksql_string = "show tables"
+        r = self.api_client.ksql(ksql_string)
+        self.assertEqual(r, [{'tables': {'statementText': 'show tables;', 'tables': []}}])
+
+    @vcr.use_cassette('tests/vcr_cassettes/ksql_create_table.yml')
+    def test_ksql_create_table(self):
+        """ Test GET requests """
+        ksql_string = "CREATE STREAM test_table (viewtime bigint, userid varchar, pageid varchar) \
+                       WITH (kafka_topic='t1', value_format='DELIMITED');"
+        r = self.api_client.ksql(ksql_string)
+        self.assertEqual(r[0]['currentStatus']['commandStatus']['status'], 'SUCCESS')
+

tests/vcr_cassettes/get_ksql_server.yml

@@ -0,0 +1,18 @@
+interactions:
+- request:
+    body: null
+    headers:
+      Accept: ['*/*']
+      Accept-Encoding: ['gzip, deflate']
+      Connection: [keep-alive]
+      User-Agent: [python-requests/2.18.4]
+    method: GET
+    uri: http://ksql-server:8080/
+  response:
+    body: {string: '{"KSQL Server Info":{"version":"0.2"}}'}
+    headers:
+      Content-Type: [application/json]
+      Date: ['Fri, 08 Dec 2017 11:36:49 GMT']
+      Server: [Jetty(9.2.z-SNAPSHOT)]
+    status: {code: 200, message: OK}
+version: 1

tests/vcr_cassettes/ksql_create_table.yml

@@ -0,0 +1,24 @@
+interactions:
+- request:
+    body: '{"ksql": "CREATE STREAM test_table (viewtime bigint, userid varchar, pageid
+      varchar)                        WITH (kafka_topic=''t1'', value_format=''DELIMITED'');"}'
+    headers:
+      Accept: ['*/*']
+      Accept-Encoding: ['gzip, deflate']
+      Connection: [keep-alive]
+      Content-Length: ['160']
+      Content-Type: [application/json]
+      User-Agent: [python-requests/2.18.4]
+    method: POST
+    uri: http://ksql-server:8080/ksql
+  response:
+    body: {string: '[{"currentStatus":{"statementText":"CREATE STREAM test_table (viewtime
+        bigint, userid varchar, pageid varchar)                        WITH (kafka_topic=''t1'',
+        value_format=''DELIMITED'');","commandId":"stream/TEST_TABLE","commandStatus":{"status":"SUCCESS","message":"Stream
+        created"}}}]'}
+    headers:
+      Content-Type: [application/json]
+      Date: ['Fri, 08 Dec 2017 12:14:41 GMT']
+      Server: [Jetty(9.2.z-SNAPSHOT)]
+    status: {code: 200, message: OK}
+version: 1

tests/vcr_cassettes/ksql_show_table.yml

@@ -0,0 +1,20 @@
+interactions:
+- request:
+    body: '{"ksql": "show tables;"}'
+    headers:
+      Accept: ['*/*']
+      Accept-Encoding: ['gzip, deflate']
+      Connection: [keep-alive]
+      Content-Length: ['24']
+      Content-Type: [application/json]
+      User-Agent: [python-requests/2.18.4]
+    method: POST
+    uri: http://ksql-server:8080/ksql
+  response:
+    body: {string: '[{"tables":{"statementText":"show tables;","tables":[]}}]'}
+    headers:
+      Content-Type: [application/json]
+      Date: ['Fri, 08 Dec 2017 12:14:41 GMT']
+      Server: [Jetty(9.2.z-SNAPSHOT)]
+    status: {code: 200, message: OK}
+version: 1