dev api

data_watchtower/api/__init__.py

@@ -0,0 +1,2 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-

data_watchtower/api/handlers/__init__.py

@@ -0,0 +1,2 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-

data_watchtower/api/handlers/base.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import tornado.web
+from data_watchtower.utils import json_dumps
+
+
+class BaseHandler(tornado.web.RequestHandler):
+    def initialize(self, *args, **kwargs):
+        self.database = self.settings.get('database')
+
+    @staticmethod
+    def json_dumps(data):
+        return json_dumps(data)
+
+    def json(self, data=None, error=None):
+        if data is None:
+            data = []
+        if error is None:
+            error = dict(
+                err_code=0,
+                err_msg="",
+            )
+        data = {
+            'data': data,
+            'error': error,
+        }
+        self.write(self.json_dumps(data))

data_watchtower/api/handlers/watchtower.py

@@ -0,0 +1,20 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from .base import BaseHandler
+
+
+class WatchtowerHandler(BaseHandler):
+    def get(self):
+        name = self.get_argument('name')
+        wt = self.database.get_watchtower(name)
+        if not wt:
+            self.json(error={'err_code': 1001, 'err_msg': 'watchtower not found'})
+            return
+        return self.json(wt.to_dict())
+
+
+class WatchtowerListHandler(BaseHandler):
+    def get(self):
+        data = self.database.get_watchtowers()
+        self.json(data)
+        return

data_watchtower/api/server.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import os
+import asyncio
+
+import tornado.web
+
+from tornado.options import define, options
+from data_watchtower import DbServices
+from data_watchtower.api.url import URLS
+
+default_db_url = os.getenv('DW_DB_URL', "sqlite:///data.db")
+define("port", default=8888, type=int, help="port to listen on")
+define("db_url", default=default_db_url, help="db url")
+
+
+def make_app():
+    database = DbServices(options.db_url)
+    app = tornado.web.Application(
+        URLS,
+        database=database
+    )
+    return app
+
+
+async def main():
+    options.parse_command_line()
+    app = make_app()
+    app.listen(options.port)
+    await asyncio.Event().wait()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

data_watchtower/api/url.py

@@ -0,0 +1,7 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+from .handlers import watchtower
+
+URLS = [
+    (r"/", watchtower.WatchtowerHandler),
+]

data_watchtower/core/base.py

@@ -53,6 +53,9 @@ def __init__(self, params: Params):
         self.result = None
         self.params = params
 
+    def params_to_dict(self):
+        return to_dict(self.params)
+
     def _validation(self):
         raise NotImplementedError
 
@@ -77,6 +80,7 @@ def module_path(cls):
     def to_dict(self):
         result = dict()
         result['params'] = to_dict(self.params)
+        # result['params']['__class__'] = "%s:%s" % (self.params.__class__.__module__, self.params.__class__.__name__)
         result['__class__'] = "%s:%s" % (self.__class__.__module__, self.__class__.__name__)
         return result
 

data_watchtower/core/watchtower.py

@@ -22,8 +22,8 @@ def __init__(self, name, data_loader, **params):
         success_method = params.get('success_method', None)
         extra = params.get('extra', {})
         self.params = params
-        self._data_loader = data_loader
-        self._data_loader_meta = data_loader.to_dict()
+        self._data_loader = data_loader  # 原本的loader,参数可能包含宏
+        self._data_loader_meta = data_loader.to_dict()  # 原本的loader,参数可能包含宏
         self._validators = []
         self._validators_meta = []
         self.name = name
@@ -125,11 +125,13 @@ def run(self):
         wt_name = macro_template.apply_string(self.name)
         # self.metrics['raw_name'] = self.name
         # self.metrics['name'] = wt_name
-
+        data_loader_meta = {}
         for k, v in self._data_loader_meta.items():
             if isinstance(v, str):
-                self._data_loader_meta[k] = macro_template.apply_string(v)
-        self._data_loader = spawn_data_loader_from_dict(self._data_loader_meta)
+                data_loader_meta[k] = macro_template.apply_string(v)
+            else:
+                data_loader_meta[k] = v
+        self._data_loader = spawn_data_loader_from_dict(data_loader_meta)
         data = self._data_loader.load()
         validators_result = self.run_validators(data, macro_template)
         self.gen_metrics(data, validators_result)

data_watchtower/utils.py

@@ -63,6 +63,7 @@ def spawn_validator_from_dict(item):
 
 
 def spawn_data_loader_from_dict(item):
+    item = item.copy()
     cls_path = item.pop('__class__')
     cls = load_object(cls_path)
     return cls.from_dict(item)

tests/demo.py

@@ -79,6 +79,7 @@ def main():
     custom_macro_map = {
         'today': {'impl': lambda: datetime.datetime.today().strftime("%Y-%m-%d")},
         'start_date': '2024-04-01',
+        'column': 'name',
     }
     # 设置数据加载器,用来加载需要校验的数据
     query = "SELECT * FROM score where date='${today}'"
@@ -90,7 +91,7 @@ def main():
     params = ExpectRowCountToBeBetween.Params(min_value=NUM_OF_STUDENTS, max_value=None)
     wt.add_validator(ExpectRowCountToBeBetween(params))
 
-    params = ExpectColumnValuesToNotBeNull.Params(column='name')
+    params = ExpectColumnValuesToNotBeNull.Params(column='${column}')
     wt.add_validator(ExpectColumnValuesToNotBeNull(params))
 
     result = wt.run()