018f267fab
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 22s
107 lines
3.0 KiB
Python
107 lines
3.0 KiB
Python
import datetime
|
|
import logging
|
|
import socket
|
|
|
|
import psycopg2
|
|
from flask import current_app
|
|
from psycopg2.extras import LoggingConnection
|
|
|
|
CORS_HEADERS = {
|
|
"Content-Type": "application/json",
|
|
}
|
|
SECRET_NAME = "fission-ai-work-env"
|
|
CONFIG_NAME = "fission-ai-work-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:
|
|
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 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
|