FirstCommit_Avatar

apps/helpers.py

@@ -0,0 +1,133 @@
+import datetime
+import logging
+import socket
+import os
+
+import psycopg2
+from flask import current_app
+from psycopg2.extras import LoggingConnection
+from minio import Minio
+
+
+CORS_HEADERS = {
+    "Content-Type": "application/json",
+}
+SECRET_NAME = "fission-ailbl-tag-env"
+CONFIG_NAME = "fission-eom-notification-config"
+K8S_NAMESPACE = "default"
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
+
+def init_db_connection():
+    db_host = get_secret("PG_HOST", "127.0.0.1")
+    db_port = int(get_secret("PG_PORT", 5432))
+
+    if not check_port_open(ip=db_host, port=db_port):
+        raise Exception(f"Establishing A Database Connection. `{db_host}:{db_port}`")
+
+    # options = get_secret("PG_DBSCHEMA")
+    # if options:
+    #     options = f"-c search_path={options}"  # if specific db schema
+    conn = psycopg2.connect(
+        database=get_secret("PG_DB", "postgres"),
+        user=get_secret("PG_USER", "postgres"),
+        password=get_secret("PG_PASS", "secret"),
+        host=get_secret("PG_HOST", "127.0.0.1"),
+        port=int(get_secret("PG_PORT", 5432)),
+        # options=options,
+        # cursor_factory=NamedTupleCursor,
+        connection_factory=LoggingConnection,
+    )
+    conn.initialize(logger)
+    return conn
+
+
+def db_row_to_dict(cursor, row):
+    record = {}
+    for i, column in enumerate(cursor.description):
+        data = row[i]
+        if isinstance(data, datetime.datetime):
+            data = data.isoformat()
+        record[column.name] = data
+    return record
+
+
+def db_rows_to_array(cursor, rows):
+    return [db_row_to_dict(cursor, row) for row in rows]
+
+
+def get_current_namespace() -> str: # Lấy config từ env fission kubernetes => Vì có nhiều môi trường khác nhau
+    try:
+        with open("/var/run/secrets/kubernetes.io/serviceaccount/namespace", "r") as f:
+            namespace = f.read()
+    except Exception as err:
+        current_app.logger.error(err)
+        namespace = K8S_NAMESPACE
+    return str(namespace)
+
+
+def get_secret(key: str, default=None):
+    namespace = get_current_namespace()
+    path = f"/secrets/{namespace}/{SECRET_NAME}/{key}"
+    try:
+        with open(path, "r") as f:
+            return f.read()
+    except Exception as err:
+        current_app.logger.error(path, err)
+        return default
+
+S3_BUCKET = get_secret("S3_BUCKET")
+def get_minio_client():
+    client = Minio(
+        os.getenv("MINIO_ENDPOINT", "localhost:9000"),
+        access_key=os.getenv("MINIO_ACCESS_KEY", "minioadmin"),
+        secret_key=os.getenv("MINIO_SECRET_KEY", "minioadmin"),
+        secure=False
+    )
+    return client
+
+
+def get_config(key: str, default=None):
+    namespace = get_current_namespace()
+    path = f"/configs/{namespace}/{CONFIG_NAME}/{key}"
+    try:
+        with open(path, "r") as f:
+            return f.read()
+    except Exception as err:
+        current_app.logger.error(path, err)
+        return default
+
+
+def str_to_bool(input: str | None) -> bool:
+    input = input or ""
+    # Dictionary to map string values to boolean
+    BOOL_MAP = {"true": True, "false": False}
+    return BOOL_MAP.get(input.strip().lower(), None)
+
+
+def check_port_open(ip: str, port: int, timeout: int = 30):
+    try:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.settimeout(timeout)
+            result = s.connect_ex((ip, port))
+            return result == 0
+    except Exception as err:
+        current_app.logger.err(f"Check port open error: {err}")
+        return False
+
+# # Get DB connection
+# def get_db_connection(cursor_factory=None): # Hàm truy cập đến database,  Thông tin database sẽ nằm ở trong này
+#     try:
+#         conn = psycopg2.connect(
+#             host=os.getenv("DB_HOST"),
+#             database=os.getenv("DB_NAME"), #biến env(môi trường)
+#             user=os.getenv("DB_USERNAME"),
+#             password=os.getenv("DB_PASSWORD")
+#         )
+#         print("✅ Connected to DB successfully")
+#         return conn
+#     except Exception as e:
+#         print("❌ Database connection failed:", e)
+#         raise