py-ailbl-user-address/apps/helpers.py

import datetime
import logging
import socket
import typing
import psycopg2
from flask import current_app
from psycopg2.extras import LoggingConnection


CORS_HEADERS = {
    "Content-Type": "application/json",
}
SECRET_NAME = "fission-ailbl-user-address-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", "locahost")
    db_port = int(get_secret("PG_PORT", 55432))

    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_row_to_dict(cursor, row):
    record = {}
    for i, column in enumerate(cursor.description):
        data = row[i]
        if isinstance(data, (datetime.datetime, datetime.date)):
            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:
    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


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


def str_to_bool(value: str | None) -> typing.Optional[bool]:
    if value is None:
        return None
    val = value.strip().lower()
    if val in ("true", "1", "yes"):
        return True
    if val in ("false", "0", "no"):
        return False
    return None