add log backend; update clients; add pytest.ini for log config; add t… (#108)

* add log backend; update clients; add pytest.ini for log config
Signed-off-by: SreeV <[email protected]>

Author: sreev

marquez_client/client_wo.py

@@ -46,7 +46,7 @@ def create_namespace(self, namespace_name, owner_name, description=None):
         return self._backend.put(
             self._path('/namespaces/{0}', namespace_name),
             headers=_HEADERS,
-            json=payload
+            json_payload=payload
         )
 
     # Source API
@@ -67,7 +67,7 @@ def create_source(self, source_name, source_type, connection_url,
         return self._backend.put(
             self._path('/sources/{0}', source_name),
             headers=_HEADERS,
-            json=payload)
+            json_payload=payload)
 
     # Datasets API
     def create_dataset(self, namespace_name, dataset_name, dataset_type,
@@ -109,7 +109,7 @@ def create_dataset(self, namespace_name, dataset_name, dataset_type,
             self._path('/namespaces/{0}/datasets/{1}', namespace_name,
                        dataset_name),
             headers=_HEADERS,
-            json=payload
+            json_payload=payload
         )
 
     # Job API
@@ -138,7 +138,7 @@ def create_job(self, namespace_name, job_name, job_type,
         return self._backend.put(
             self._path('/namespaces/{0}/jobs/{1}', namespace_name, job_name),
             headers=_HEADERS,
-            json=payload
+            json_payload=payload
         )
 
     def create_job_run(self, namespace_name, job_name, run_id,
@@ -165,7 +165,7 @@ def create_job_run(self, namespace_name, job_name, run_id,
             self._path('/namespaces/{0}/jobs/{1}/runs',
                        namespace_name, job_name),
             headers=_HEADERS,
-            json=payload)
+            json_payload=payload)
 
         if mark_as_running:
             response = self.mark_job_run_as_started(
@@ -192,7 +192,7 @@ def __mark_job_run_as(self, run_id, action, action_at=None):
             self._path('/jobs/runs/{0}/{1}?at={2}', run_id, action,
                        action_at if action_at else Utils.utc_now()),
             headers=_HEADERS,
-            json={}
+            json_payload={}
         )
 
     # Common

marquez_client/clients.py

@@ -12,14 +12,15 @@
 
 import logging
 import os
-import marquez_client
 
-from marquez_client.file_backend import FileBackend
-from marquez_client.http_backend import HttpBackend
+import marquez_client
 from marquez_client.constants import (DEFAULT_MARQUEZ_BACKEND,
                                       DEFAULT_MARQUEZ_URL,
                                       DEFAULT_MARQUEZ_FILE,
                                       DEFAULT_TIMEOUT_MS)
+from marquez_client.file_backend import FileBackend
+from marquez_client.http_backend import HttpBackend
+from marquez_client.log_backend import LogBackend
 from marquez_client.utils import Utils
 
 log = logging.getLogger(__name__)
@@ -52,3 +53,5 @@ def from_env():
         elif backend_env == 'file':
             file = os.environ.get('MARQUEZ_FILE', DEFAULT_MARQUEZ_FILE)
             return FileBackend(file)
+        elif backend_env == 'log':
+            return LogBackend()

marquez_client/file_backend.py

@@ -29,36 +29,39 @@ def _mkdir_if_not_exists(file):
         os.makedirs(path, exist_ok=True)
         return file
 
-    def put(self, path, headers, json):
+    def put(self, path, headers, json_payload):
         log.debug("_put()")
 
         put_details = {}
 
         put_details['method'] = 'PUT'
         put_details['path'] = path
         put_details['headers'] = headers
-        put_details['payload'] = json
+        put_details['payload'] = json_payload
 
-        log.info(put_details)
+        put_json = json.dumps(put_details)
+        log.info(put_json)
 
-        self._sync_file(put_details)
+        self._sync_file(put_json)
 
-    def post(self, path, headers, json=None):
+    def post(self, path, headers, json_payload=None):
         log.debug("_post()")
 
         post_details = {}
 
         post_details['method'] = 'POST'
         post_details['path'] = path
         post_details['headers'] = headers
-        post_details['payload'] = json
+        if json_payload:
+            post_details['payload'] = json_payload
 
-        log.info(post_details)
+        post_json = json.dumps(post_details)
+        log.info(post_json)
 
-        self._sync_file(post_details)
+        self._sync_file(post_json)
 
     def _sync_file(self, json_data):
-        self._file.write(json.dumps(json_data))
+        self._file.write(json_data)
         self._file.write(os.linesep)
         self._file.flush()
         os.fsync(self._file.fileno())

marquez_client/log_backend.py

@@ -0,0 +1,50 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+
+from marquez_client.backend import Backend
+
+log = logging.getLogger(__name__)
+
+
+class LogBackend(Backend):
+    def __init__(self):
+        log.debug("LogBackend.init")
+
+    def put(self, path, headers, json_payload):
+        log.debug("_put()")
+
+        put_details = {}
+
+        put_details['method'] = 'PUT'
+        put_details['path'] = path
+        put_details['headers'] = headers
+        put_details['payload'] = json_payload
+
+        log_details = json.dumps(put_details)
+        log.info(log_details)
+
+    def post(self, path, headers, json_payload=None):
+        log.debug("_post()")
+
+        post_details = {}
+
+        post_details['method'] = 'POST'
+        post_details['path'] = path
+        post_details['headers'] = headers
+        if json_payload:
+            post_details['payload'] = json_payload
+
+        log_details = json.dumps(post_details)
+        log.info(log_details)

marquez_client/utils.py

@@ -10,13 +10,15 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import logging
 import uuid
+from datetime import datetime
+
 import pytz
 import time
-
-from datetime import datetime
 from pyrfc3339 import generate
+
 from marquez_client.models import (DatasetFieldType, DatasetType)
 
 log = logging.getLogger(__name__)
@@ -92,3 +94,8 @@ def now_ms():
     def utc_now():
         return str(generate(datetime.utcnow().replace(tzinfo=pytz.utc),
                             microseconds=True))
+
+    @staticmethod
+    def get_json(file):
+        with open(file) as json_file:
+            return json.load(json_file)

tests/pytest.ini

@@ -0,0 +1,17 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[pytest]
+log_cli_level = DEBUG
+log_cli = True
+log_cli_format = %(asctime)s %(levelname)s %(message)s
+log_cli_date_format = %Y-%m-%d %H:%M:%S

tests/resources/post.json

@@ -0,0 +1,16 @@
+{
+  "method": "POST",
+  "path": "/api/v1/namespaces/my-namespace/jobs/my-job/runs",
+  "headers": {
+    "User-Agent": "marquez-python/0.7.6"
+  },
+  "payload": {
+    "id": "8c78b02b-86b8-40c7-91bf-e5dd7899ed3b",
+    "args": {
+      "email": "[email protected]",
+      "emailOnFailure": "true",
+      "emailOnRetry": "true",
+      "retries": "1"
+    }
+  }
+}

tests/resources/put.json

@@ -0,0 +1,24 @@
+{
+  "method": "PUT",
+  "path": "/api/v1/namespaces/my-namespace/jobs/my-job",
+  "headers": {
+    "User-Agent": "marquez-python/0.7.6"
+  },
+  "payload": {
+    "inputs": [
+      {
+        "namespace": "my-namespace",
+        "name": "public.mytable"
+      }
+    ],
+    "outputs": {
+      "namespace": "my-namespace",
+      "name": "public.mytable"
+    },
+    "type": "BATCH",
+    "context": {
+      "sql": "SELECT * FROM public.mytable;"
+    },
+    "location": "https://github.com/my-jobs/blob/07f3d2dfc8186cadae9146719e70294a4c7a8ee8"
+  }
+}

tests/test_log_backend.py

@@ -0,0 +1,64 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import json
+import logging
+import unittest
+
+import mock
+
+from marquez_client.log_backend import LogBackend
+from marquez_client.utils import Utils
+
+log = logging.getLogger(__name__)
+
+
+class TestLogBackend(unittest.TestCase):
+    def setUp(self):
+        log.debug("TestLogBackend.setup(): ")
+        self.log_backend = LogBackend()
+        self.put_json = Utils.get_json("tests/resources/put.json")
+        self.post_json = Utils.get_json("tests/resources/post.json")
+
+    @mock.patch("json.dumps")
+    def test_put(self, mock_put):
+        log.debug("TestLogBackend.test_put(): ")
+
+        path = self.put_json["path"]
+        headers = self.put_json["headers"]
+        json_payload = self.put_json["payload"]
+
+        self.log_backend.put(path, headers, json)
+        mock_put.return_value = self.put_json
+
+        assert path == mock_put.return_value["path"]
+        assert headers == mock_put.return_value["headers"]
+        assert json_payload == mock_put.return_value["payload"]
+
+    @mock.patch("json.dumps")
+    def test_post(self, mock_post):
+        log.debug("TestLogBackend.test_post(): ")
+
+        path = self.post_json["path"]
+        headers = self.post_json["headers"]
+        json_payload = self.post_json["payload"]
+
+        self.log_backend.post(path, headers, json)
+        mock_post.return_value = self.post_json
+
+        assert path == mock_post.return_value["path"]
+        assert headers == mock_post.return_value["headers"]
+        assert json_payload == mock_post.return_value["payload"]
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_marquez_wo_client_log.py

@@ -0,0 +1,35 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging.config
+import os
+import unittest
+
+from marquez_client import Clients
+from marquez_client.log_backend import LogBackend
+
+_NAMESPACE = "my-namespace"
+log = logging.getLogger(__name__)
+
+
+class TestMarquezWriteOnlyClientLog(unittest.TestCase):
+    def test_log_backend(self):
+        log.debug("MarquezWriteOnlyClient.setup(): ")
+        os.environ['MARQUEZ_BACKEND'] = 'log'
+        self.client_wo_log = Clients.new_write_only_client()
+        log.info("created marquez_client_wo_log.")
+        if not isinstance(self.client_wo_log._backend, LogBackend):
+            raise Exception("Not a LogBackend.")
+
+
+if __name__ == '__main__':
+    unittest.main()