[#] init project: add id_allocation controller for postgresql proce…

…ss_id_allocation function

.gitignore

@@ -0,0 +1,7 @@
+.vscode
+venv
+log
+data
+.history
+src/config/*.toml
+!src/config/main.sample.toml

requirements.txt

@@ -0,0 +1,29 @@
+astroid==2.13.5
+certifi==2024.2.2
+charset-normalizer==3.3.2
+confluent-kafka==2.2.0
+dill==0.3.8
+idna==3.7
+isort==5.13.2
+lazy-object-proxy==1.10.0
+mccabe==0.7.0
+numpy==1.22.4
+pandas==1.3.5
+platformdirs==4.2.2
+protobuf==4.23.4
+psycopg2-binary==2.9.9
+pylint==2.15.2
+python-dateutil==2.8.2
+pytz==2024.1
+PyYAML==6.0
+requests==2.32.2
+six==1.16.0
+supervisor==4.2.5
+toml==0.10.2
+tomli==2.0.1
+tomlkit==0.12.5
+tornado==6.3.3
+typing_extensions==4.11.0
+urllib3==2.2.1
+uuid==1.30
+wrapt==1.16.0

src/config/main.sample.toml

@@ -0,0 +1,14 @@
+[server]
+server_name = "id_allocation"
+ip = "127.0.0.1"
+port = 9001
+work_path = "/app"
+thread_count = 8
+process_count = 6
+log_path = "/app/log"
+log_level = 1
+log_max_size = 1600
+
+[id_allocation]
+write = "postgresql://postgres:password@host/dbname"
+read = "postgresql://postgres:password@host/dbname"

src/controller/allocation_controller.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+'''
+Author: Zella Zhong
+Date: 2024-05-23 22:47:04
+LastEditors: Zella Zhong
+LastEditTime: 2024-05-24 04:52:22
+FilePath: /id_allocation/src/controller/allocation_controller.py
+Description: allocation controller
+'''
+import json
+import logging
+import time
+import setting
+
+from httpsvr import httpsvr
+import psycopg2
+from setting import get_write_conn
+
+
+def dict_factory(cursor, row):
+    """
+    Convert query result to a dictionary.
+    """
+    col_names = [col_desc[0] for col_desc in cursor.description]
+    return dict(zip(col_names, row))
+
+
+class AllocationController(httpsvr.BaseController):
+    '''AllocationController'''
+    def __init__(self, obj, param=None):
+        super(AllocationController, self).__init__(obj)
+
+    def allocation(self):
+        '''
+        description:
+        requestbody: {
+            "graph_id": "string",
+            "updated_nanosecond": "int64",
+            "vids": ["string"],
+        }
+        return: {
+            "return_graph_id": "string",
+            "return_updated_nanosecond": "int64",
+        }
+        '''
+        post_data = self.inout.request.body
+        data = json.loads(post_data)
+        graph_id = data.get("graph_id", "")
+        updated_nanosecond = data.get("updated_nanosecond", 0)
+        vids = data.get("vids", [])
+        if graph_id == "" or updated_nanosecond == 0:
+            return httpsvr.Resp(msg="Invalid input body", data={}, code=-1)
+        if len(vids) == 0:
+            return httpsvr.Resp(msg="Invalid input body", data={}, code=-1)
+
+
+        data = {}
+        rows = []
+        code = 0
+        msg = ""
+        try:
+            pg_conn = get_write_conn()
+            cursor = pg_conn.cursor()
+
+            process_vids = "ARRAY[" + ",".join(["'" + x + "'" for x in vids]) + "]"
+            ssql = "SELECT * FROM process_id_allocation(%s, '%s', %d);" % (process_vids, graph_id, updated_nanosecond)
+            cursor.execute(ssql)
+            rows = [dict_factory(cursor, row) for row in cursor.fetchall()]
+            logging.debug("allocation vids: {}, result: {}".format(process_vids, rows))
+            if len(rows) == 0:
+                cursor.close()
+                pg_conn.close()
+                return httpsvr.Resp(msg="allocation ID=null", data={}, code=-1)
+
+            data = rows[0]
+            cursor.close()
+            pg_conn.close()
+        except Exception as e:
+            code = -1
+            msg = repr(e)
+            logging.exception(e)
+
+        return httpsvr.Resp(msg=msg, data=data, code=code)

src/data_server.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+'''
+Author: Zella Zhong
+Date: 2024-05-23 22:34:52
+LastEditors: Zella Zhong
+LastEditTime: 2024-05-23 22:54:59
+FilePath: /id_allocation/src/data_server.py
+Description: main entry point for allocating
+'''
+import os
+import logging
+
+import setting
+import setting.filelogger as logger
+
+from controller.allocation_controller import AllocationController
+
+
+if __name__ == "__main__":
+    config = setting.load_settings(env="development")
+    # config = setting.load_settings(env="production")
+    if not os.path.exists(config["server"]["log_path"]):
+        os.makedirs(config["server"]["log_path"])
+    logger.InitLogger(config)
+    logger.SetLoggerName("id_allocation")
+
+    try:
+        from httpsvr import httpsvr
+        # [path, controller class, method]
+        ctrl_info = [
+            ["/id_allocation/allocation", AllocationController, "allocation"],
+        ]
+        svr = httpsvr.HttpSvr(config, ctrl_info)
+        svr.Start()
+
+    except Exception as e:
+        logging.exception(e)

src/httpsvr/config.py

@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+import os
+import logging
+
+class HttpSvrConfig():
+    """config is an toml object"""
+    def __init__(self, config):
+        self.ip = ""
+        self.port = config["server"]["port"]
+        self.work_path = config["server"]["work_path"]
+        self.server_name = config["server"]["server_name"]
+        try:
+            self.thread_count = config["server"]["thread_count"]
+        except:
+            self.thread_count = 8
+
+        try:
+            self.process_count = config["server"]["process_count"]
+        except:
+            self.process_count = 0
+
+        self.log_path = config["server"]["log_path"]
+        self.log_level = config["server"]["log_level"]
+        self.log_max_size = config["server"]["log_max_size"]
+
+        if self.log_path == "":
+            self.log_path = "./log"
+
+        if self.ip == "":
+            self.ip = "127.0.0.1"
+
+        try:
+            self.max_buffer_size = config["server"]["max_buffer_size"]
+        except:
+            self.max_buffer_size = 0
+