02dc4943f6
Some checks failed
K8S Fission Deployment / Deployment fission functions (push) Failing after 12s
133 lines
4.0 KiB
Python
133 lines
4.0 KiB
Python
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-user-avatar-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 |