Merge pull request #2 from cr0wg4n/develop

update readme and examples

README.md

@@ -1,5 +1,10 @@
 # Kibana API Mapping Library
 
+
+[![Supported Versions](https://img.shields.io/pypi/pyversions/kibana-api.svg)](https://pypi.org/project/kibana-api/)
+[![Downloads](https://pepy.tech/badge/kibana-api/month)](https://pepy.tech/project/kibana-api/month)
+
+
 ## Development Requirements
 I only use `requests` to perform HTTP requests and pure logic for 
 all behaviour.
@@ -9,13 +14,14 @@ all behaviour.
 pip install kibana-api
 ```
 ## Usage and Examples
+If you going to test every example, you shold run the `docker-compose.yml` example(development section).
 
-Configure Kibana Object:
+### Configure Kibana Object:
 ```python
 URL = "http://localhost:5601"
-USERNAME = "XXXX"
+USERNAME = "XXXX" 
 PASSWORD = "XXXX"
-
+# username and password are optional fields
 kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
 ```
 
@@ -168,4 +174,12 @@ python -m unittest tests.tests
 ```
 
 ## Contributing
-Yes fella, you know how ;)
+Yes fella, you know how ;)
+
+## Contact Me
+
+My blog: [cr0wg4n](https://cr0wg4n.medium.com/)
+
+Twitter: [cr0wg4n](https://twitter.com/cr0wg4n)
+
+Linkedin: [cr0wg4n](https://www.linkedin.com/in/cr0wg4n/)

kibana_api/kibana.py

@@ -2,20 +2,20 @@
 from urllib.parse import urljoin
 from .objects import Space, Object
 
-spaces = "http://localhost:5601/api/spaces/space"
-host = 'http://localhost:5601/s/demo/api/saved_objects/index-pattern'
-
 class Kibana:
     def __init__(self, base_url, username=None, password=None) -> None:
         self.base_url = base_url
         self.username = username
         self.password = password
 
+    def ping(self):
+        status_code = self.requester(method="get", url=self.base_url).status_code
+        return True if status_code == 200 else False
+
     def url(self, *args) -> str:
         url = self.base_url
         for arg in args:
-            if arg:
-                url = urljoin(url, arg + "/")
+            url = urljoin(url, arg)+ "/"
         return  url[:-1] if url[-1:]=="/" else url
 
     def requester(self, **kwargs): 

kibana_api/objects.py

@@ -22,27 +22,35 @@ def create(self):
         return response
 
 class Object(BaseModel):
-    def __init__(self, space_id=None, kibana=None, attribs={}, type=None, references={}) -> None:
+    def __init__(self, space_id=None, kibana=None, attribs={}, type="", references={}) -> None:
         super().__init__([], space_id=space_id, kibana=kibana)
         self.types =["visualization", "dashboard", "search", "index-pattern", 
         "config", "timelion-sheet", "url", "query", "canvas-element", "canvas-workpad", "lens",
         "infrastructure-ui-source", "metrics-explorer-view", "inventory-view"]
         self.create_url = "api/saved_objects"
         self.import_url = "api/saved_objects/_import"
         self.all_url = "api/saved_objects/_find"
+        self.get_url = "api/saved_objects"
         self.attribs = attribs
         self.references = references
-        self.type = type
+        self.type = type.lower()
 
-    def all(self, type=None):
+    def get(self, id, type=""):
+        type = self.type if not type else type.lower()
+        if not self.validate_type(type, types=self.types):
+            return None
+        url = self.url(self.get_url, type, id)
+        return self.requester(url=url, method="get")
+
+    def all(self, type=""):
         params = {
             "type": self.types if not type else type
         }
         url = self.url(self.all_url)
         return self.requester(url=url, method="get", params=params)
 
-    def create(self, type=None, attribs={}, references={}, body={}):
-        type = (self.type.lower() if self.type else type.lower())
+    def create(self, type="", attribs={}, references={}, body={}):
+        type = (self.type if not type else type.lower())
         attribs = (self.attribs if not attribs else attribs)
         references = (self.references if not references else references)
         if not self.validate_type(type, types=self.types):
@@ -137,7 +145,7 @@ def __querier(self):
 
 class Visualization(Utils):
 
-    def __init__(self, index_pattern_id:str, title:str="", query:str="", mappings_dir_path:str=None, type:str=None) -> None:
+    def __init__(self, index_pattern_id:str, title:str="", query:str="", mappings_dir_path:str=None, type:str="") -> None:
         CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
         self.primitive_type="visualization"
         self.types=["area", "heatmap", "line", "metric", "pie", "table", "tagcloud"]
@@ -160,7 +168,6 @@ def create(self, index_pattern_id:str=None, title:str="", body={}, query:str="")
         visualization_state = self.__templater(title) if not body else self.__templater_json(title, body) 
         search_state = self.__querier(self.query if not query else query)
         index_pattern_id = self.index_pattern_id if not index_pattern_id else index_pattern_id
-        print(visualization_state)
         return {
             "attributes": {
                 "title": title,

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="kibana-api",
-    version="0.0.2",
+    version="0.0.3",
     author="Mauricio Matias Conde",
     author_email="mcm.crw@gmail.com",
     description="This is an API mapping library for Kibana API to generate visualizations and dashboards automatically",

tests/tests.py

@@ -1,25 +1,57 @@
-from kibana_api import Dashboard, Panel, Visualization, Kibana
+import time
+import json
+from kibana_api import Dashboard, Panel, Visualization, Kibana, kibana
 import random
 import unittest
 import os
 
 URL = "http://localhost:5601"
+ELASTIC_DEMO_INDEX_URL = "http://localhost:9200/index_demo/_doc"
 USERNAME = "elastic"
 PASSWORD = "elastic"
+
+
+class mock:
+    index_pattern_id = ""
+    space_id = ""
+    visualization_id = ""
+
 class TestStringMethods(unittest.TestCase):
 
+    def test_ping(self):
+        # return True
+        kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
+        test = True
+        ping = kibana.ping() # True or False
+        self.assertEqual(ping, test)
+
     def test_url_parser(self):
-        pass
+        # return True
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
         url = kibana.url(URL, "1", "2", "3")
         self.assertEqual("http://localhost:5601/1/2/3", url)
 
-    def test_create_space(self):
-        pass
+    def test_generate_data(self):
+        # return True
+        data = {
+           "@timestamp": "2021-05-26T13:40:15.000Z",
+            "user": {
+                "name": "matias max",
+                "email": "mcm.12@asd.com",
+                "age": 22
+            }
+        }
+        response = Kibana(base_url=URL, username=USERNAME, password=PASSWORD).requester(url=ELASTIC_DEMO_INDEX_URL, method="post", data=json.dumps(data))
+        print("generated data:", response)
+        self.assertEqual(201, response.status_code)
+
+
+    def test1_create_space(self):
+        # return True
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        id = f"test-{int(random.randint(0,100)*0.33)}"
-        name = "test" + id
-        description = "descripcion del espacio de pruebas"
+        id = "test-{}".format(int(random.randint(0,100)*0.33))
+        name = id
+        description = "space description"
         color = "#000000"
         response = kibana.space(id=id, name=name, description=description, color=color).create()
         response_json = {
@@ -29,80 +61,73 @@ def test_create_space(self):
             "color": color,
             "disabledFeatures": []
         }
+        mock.space_id = id
+        print("space created: ", mock.space_id)
         self.assertEqual(response.json(), response_json)
 
-    def test_create_index_pattern(self):
-        pass
+    def test2_create_index_pattern(self):
+        # return True
         pattern_json = {
-            "title":"demo*",
+            "title":"index*",
             "timeFieldName": "@timestamp",
             "fields":"[]"
         }
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        res = kibana.object(space_id="demo").create('index-pattern', attribs=pattern_json)
-        self.assertEqual(res.json()["attributes"], pattern_json)
+        response = kibana.object(space_id=mock.space_id).create('index-pattern', attribs=pattern_json).json()
+        mock.index_pattern_id = response["id"]
+        print("index created: ", mock.index_pattern_id)
+        self.assertEqual(response["attributes"], pattern_json)
 
-    def test_import(self):
-        pass
+    def test3_import(self):
+        # return True
         CURRENT_DIR = os.path.dirname(os.path.realpath(__file__))
         FILE_PATH = os.path.join(CURRENT_DIR, 'exported_data.ndjson')
+
         file = open(FILE_PATH, 'r')
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        response = kibana.object().loads(file=file)
+        response = kibana.object(space_id=mock.space_id).loads(file=file)
         file.close()
-        print(response.json())
+        print(response)
 
-    def test_get_all_objects(self):
-        pass
+    def test4_get_all_objects(self):
+        # return True
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        response = kibana.object(space_id="demo").all(type="index-pattern")
-        print(response.json())
-
-    def test_create_panel(self):
-        pass
-        test = {'version': '7.8.0', 'gridData': {'x': 0, 'y': 12, 'w': 48, 'h': 12, 'i': 'holamundo'}, 'panelIndex': 'holamundo', 'embeddableConfig': {}, 'panelRefName': 'panel_0'}
-        result = Panel("panel_0", 48, 12, 0, 12, id="holamundo", visualization_id="XXXXXXXXXXXX")
-        references = result.get_reference()
-        print(references)
+        response = kibana.object(space_id=mock.space_id).all(type="index-pattern")
+        print(response)
+
+    def test5_create_panel(self):
+        # return True
+        panel_id = "XXXXX"
+        test = {'version': '7.8.0', 'gridData': {'x': 0, 'y': 12, 'w': 48, 'h': 12, 'i': panel_id}, 'panelIndex': panel_id, 'embeddableConfig': {}, 'panelRefName': 'panel_0'}
+        result = Panel("panel_0", 48, 12, 0, 12, id=panel_id)
         self.assertEqual(test, result.create())
 
-    def test_create_visualization(self):
-        pass
-        pattern_json = {
-            "title":"demo*",
-            "timeFieldName": "@timestamp",
-            "fields":"[]"
-        }
-        kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        res = kibana.object(space_id="demo", attribs=pattern_json).create('index-pattern').json()
-        index_pattern = res["id"]
+    def test6_create_visualization(self):
+        # return True
         type = "line"
         title = "hello this is a visualization :D 2"
-        visualization = Visualization(type=type, title=title, index_pattern_id=index_pattern).create()
-        res = kibana.object(space_id="demo").create('visualization', body=visualization).json()
-        print(res)
-
-    def test_create_dashboard(self):
-        pass
-        pattern_json = {
-            "title":"de*",
-            "timeFieldName": "@timestamp",
-            "fields":"[]"
-        }
         kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
-        res = kibana.object(space_id="demo", attribs=pattern_json).create('index-pattern').json()
-        index_pattern = res["id"]
-        type = "line"
-        title = "hello this is a visualization :D 3"
-        visualization = Visualization(type=type, title=title, index_pattern_id=index_pattern).create()
-        res = kibana.object(space_id="demo").create('visualization', body=visualization).json()
-        visualization_id = res["id"]
-        panel = Panel("panel_0", 48, 12, 0, 2, visualization_id=visualization_id)
+        visualization_obj = Visualization(type=type, title=title, index_pattern_id=mock.index_pattern_id).create()
+        response = kibana.object(space_id=mock.space_id).create('visualization', body=visualization_obj).json()
+        mock.visualization_id = response["id"]
+        print("visualization created: ", mock.visualization_id)
+
+    def test7_create_dashboard(self):
+        # return True
+        panel = Panel("panel_0", 48, 12, 0, 2, visualization_id=mock.visualization_id)
         panels = [panel.create()]
         references = [panel.get_reference()]
         dasboard = Dashboard(title="hola mundo", panels=panels, references=references, query="user.name: mat*").create()
-        res = kibana.object(space_id="demo").create('dashboard', body=dasboard).json()
-        print(res)
+        kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
+        response = kibana.object(space_id=mock.space_id).create('dashboard', body=dasboard).json()
+        dashboard_id = response["id"]
+        print("dashboard created: ", dashboard_id)
+
+    def test8_object_by_id(self):
+        # return True
+        kibana = Kibana(base_url=URL, username=USERNAME, password=PASSWORD)
+        reponse = kibana.object(space_id=mock.space_id).get(id=mock.visualization_id, type="visualization")
+        print(reponse)
 
 if __name__ == "__main__":
     unittest.main()