New example for db_pre_save: write_event

Post to api/v1/event with info about inserted and updated objects. This currently works on Postgres only, as the callback needs to open a second write TX which isn't possible with Sqlite.

Co-Authored-By: Philipp Hempel <phempel@programmfabrik.de>

manifest.master.yml

@@ -295,6 +295,14 @@ callbacks:
           objecttypes:
             - object
             - objects
+      - name: "write event with info about inserted and updated objects"
+        callback: write_event
+        filter:
+          type: objecttype
+          objecttypes:
+            - object
+            - objects
+            - images
       - name: "bounce json back for 'bounce'"
         callback: bounce
         filter:
@@ -333,6 +341,20 @@ callbacks:
               args:
                 - type: "value"
                   value: "%_exec.pluginDir%/server/db_pre_save+webhook/set_comment.js"
+      write_event:
+        exec:
+          service: python3
+          commands:
+            - prog: python3
+              stdin:
+                type: body
+              stderr:
+                type: body
+              stdout:
+                type: body
+              args:
+                - type: value
+                  value: "%_exec.pluginDir%/server/db_pre_save+write_event/write_event.py"
       check:
         exec:
           service: "node"
@@ -437,5 +459,7 @@ fas_config:
     - fas_config/cookbook_grayscale.yml
 
 custom_events:
+  EXAMPLE_PLUGIN_OBJECT_STATISTICS:
+    user: true
   FYLR_EXAMPLE_EVENT:
     user: true

server/db_pre_save+write_event/util.py

@@ -0,0 +1,95 @@
+# encoding: utf-8
+
+
+import json
+import sys
+import requests
+
+
+def get_json_value(js, path, expected=False, split_char='.'):
+
+    current = js
+    path_parts = []
+    current_part = ''
+
+    for i in range(len(path)):
+        if path[i] != split_char:
+            current_part += path[i]
+            if i == len(path) - 1:
+                path_parts.append(current_part)
+            continue
+
+        if i > 0 and path[i - 1] == '\\':
+            current_part += path[i]
+            continue
+
+        if len(current_part) > 0:
+            path_parts.append(current_part)
+            current_part = ''
+
+    for path_part in path_parts:
+        path_part = path_part.replace('\\' + split_char, split_char)
+
+        if not isinstance(current, dict) or path_part not in current:
+            if expected:
+                raise Exception('expected: ' + path_part)
+            else:
+                return None
+
+        current = current[path_part]
+
+    return current
+
+
+def dumpjs(js, indent=4):
+    return json.dumps(js, indent=indent)
+
+
+# plugin response functions
+
+
+def stdout(line):
+    sys.stdout.write(line)
+    sys.stdout.write('\n')
+
+
+def stderr(line):
+    sys.stderr.write(line)
+    sys.stderr.write('\n')
+
+
+def return_response(response):
+    stdout(dumpjs(response))
+    exit(0)
+
+
+def return_error_response(error):
+    stderr(error)
+    exit(1)
+
+# fylr api functions
+
+
+def fylr_api_headers(access_token):
+    return {
+        'authorization': 'Bearer ' + access_token,
+    }
+
+
+def get_from_api(api_url, path, access_token):
+
+    resp = requests.get(
+        api_url + '/' + path,
+        headers=fylr_api_headers(access_token))
+
+    return resp.text, resp.status_code
+
+
+def post_to_api(api_url, path, access_token, payload=None):
+
+    resp = requests.post(
+        api_url + '/' + path,
+        headers=fylr_api_headers(access_token),
+        data=payload)
+
+    return resp.text, resp.status_code

server/db_pre_save+write_event/write_event.py

@@ -0,0 +1,70 @@
+# encoding: utf-8
+
+
+import sys
+import json
+import util
+
+
+def main():
+
+    orig_data = json.loads(sys.stdin.read())
+
+    # get the server url
+    api_url = util.get_json_value(orig_data, 'info.api_url')
+    if api_url is None:
+        util.return_error_response('info.api_url missing!')
+    api_url += '/api/v1'
+
+    # get a session token
+    access_token = util.get_json_value(
+        orig_data, 'info.api_user_access_token')
+    if access_token is None:
+        util.return_error_response('info.api_user_access_token missing!')
+
+    objects = util.get_json_value(orig_data, 'objects')
+    if not isinstance(objects, list):
+        util.return_response(orig_data)
+
+    objecttype_count = {}
+
+    # iterate over the objects and count how many were inserted/updated
+    for obj in objects:
+
+        if not isinstance(obj, dict):
+            continue
+
+        objecttype = util.get_json_value(obj, '_objecttype')
+
+        if not objecttype in objecttype_count:
+            objecttype_count[objecttype] = {
+                'inserted': 0,
+                'updated': 0,
+            }
+
+        version = util.get_json_value(obj, objecttype + '._version')
+        if version == 1:
+            objecttype_count[objecttype]['inserted'] += 1
+        else:
+            objecttype_count[objecttype]['updated'] += 1
+
+    # write events
+    for objecttype in objecttype_count:
+        util.post_to_api(
+            api_url=api_url,
+            path='event',
+            access_token=access_token,
+            payload=util.dumpjs({
+                'event': {
+                    'type': 'EXAMPLE_PLUGIN_OBJECT_STATISTICS',
+                    'objecttype': objecttype,
+                    'info': objecttype_count[objecttype]
+                }
+            })
+        )
+
+    util.return_response(orig_data)
+
+
+if __name__ == '__main__':
+    main()