213 lines
10 KiB
Python
213 lines
10 KiB
Python
import os
|
|
from typing import List
|
|
|
|
# Directory Anchors
|
|
CORE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
APP_DIR = os.path.dirname(CORE_DIR)
|
|
BACKEND_DIR = os.path.dirname(APP_DIR)
|
|
PROJECT_ROOT = os.path.abspath(os.path.join(BACKEND_DIR, ".."))
|
|
|
|
# Load environment variables from .env if present
|
|
ENV_FILE_PATH = os.path.join(PROJECT_ROOT, ".env")
|
|
if not os.path.exists(ENV_FILE_PATH):
|
|
ENV_FILE_PATH = os.path.join(BACKEND_DIR, ".env")
|
|
|
|
if os.path.exists(ENV_FILE_PATH):
|
|
with open(ENV_FILE_PATH, "r", encoding="utf-8") as ef:
|
|
for line in ef:
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
k, v = line.split("=", 1)
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
|
|
def resolve_dir_path(env_var_name: str, relative_subpath: str) -> str:
|
|
"""Resolve directory path with fallback across project root and backend dir."""
|
|
env_val = os.getenv(env_var_name)
|
|
if env_val:
|
|
if os.path.isabs(env_val) and os.path.exists(env_val):
|
|
return os.path.abspath(env_val)
|
|
c1 = os.path.abspath(os.path.join(BACKEND_DIR, env_val.lstrip("./")))
|
|
if os.path.exists(c1):
|
|
return c1
|
|
c2 = os.path.abspath(os.path.join(PROJECT_ROOT, env_val.lstrip("./")))
|
|
if os.path.exists(c2):
|
|
return c2
|
|
|
|
c1 = os.path.abspath(os.path.join(BACKEND_DIR, relative_subpath))
|
|
if os.path.exists(c1):
|
|
return c1
|
|
c2 = os.path.abspath(os.path.join(PROJECT_ROOT, relative_subpath))
|
|
if os.path.exists(c2):
|
|
return c2
|
|
return os.path.abspath(os.path.join(PROJECT_ROOT, relative_subpath))
|
|
|
|
|
|
import glob
|
|
|
|
def resolve_configs_dir() -> str:
|
|
"""Resolve pipeline configs directory with SSOT priority:
|
|
1. Explicit custom AIRFLOW_CONFIGS_DIR (if valid and exists)
|
|
2. Container mount /app/merica-pipelines/configs
|
|
3. Host repo ../release/merica-pipelines/configs
|
|
4. Sibling repo ../merica-pipelines/configs or ../merica-pipeline-bundles/configs
|
|
5. Airflow 3 GitDAGBundle directories under /opt/airflow/dag_bundles/*/tracking_repo/configs
|
|
6. Local fallback directory airflow/configs
|
|
"""
|
|
env_val = os.getenv("AIRFLOW_CONFIGS_DIR")
|
|
if env_val:
|
|
resolved = resolve_dir_path("AIRFLOW_CONFIGS_DIR", env_val)
|
|
if os.path.exists(resolved) and (os.path.isdir(resolved) and any(f.endswith(('.yaml', '.yml')) for f in os.listdir(resolved)) or env_val.strip("./") not in ("airflow/configs", "configs")):
|
|
return resolved
|
|
|
|
# Check container mount
|
|
container_cfg = "/app/merica-pipelines/configs"
|
|
if os.path.exists(container_cfg) and os.path.isdir(container_cfg):
|
|
return container_cfg
|
|
|
|
# Check release/merica-pipelines and sibling repos
|
|
for rel_path in [
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "release", "merica-pipelines", "configs")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "release", "merica-pipelines", "configs")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipelines", "configs")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipeline-bundles", "configs")),
|
|
]:
|
|
if os.path.exists(rel_path) and os.path.isdir(rel_path):
|
|
return rel_path
|
|
|
|
# Auto-detect AIP-66 Git DAG Bundles in standard Airflow 3 bundle storage
|
|
for bundle_cfg in sorted(glob.glob("/opt/airflow/dag_bundles/*/tracking_repo/configs")):
|
|
if os.path.exists(bundle_cfg):
|
|
return bundle_cfg
|
|
|
|
bundle_path = os.getenv("AIRFLOW_DAG_BUNDLE_PATH", "/opt/airflow/dag_bundles/merica-pipeline/tracking_repo/configs")
|
|
if os.path.exists(bundle_path):
|
|
return bundle_path
|
|
|
|
return resolve_dir_path("AIRFLOW_CONFIGS_DIR", os.path.join("airflow", "configs"))
|
|
|
|
|
|
def resolve_metadata_dir() -> str:
|
|
"""Resolve pipeline metadata directory with SSOT priority:
|
|
1. Explicit custom AIRFLOW_METADATA_DIR
|
|
2. Container mount /app/merica-pipelines/metadata
|
|
3. Host repo ../release/merica-pipelines/metadata
|
|
4. Sibling repo ../merica-pipelines/metadata or ../merica-pipeline-bundles/metadata
|
|
5. Airflow 3 GitDAGBundle metadata directories (/opt/airflow/dag_bundles/*/tracking_repo/metadata)
|
|
6. Fallback to CONFIGS_DIR
|
|
"""
|
|
env_val = os.getenv("AIRFLOW_METADATA_DIR")
|
|
if env_val:
|
|
resolved = resolve_dir_path("AIRFLOW_METADATA_DIR", env_val)
|
|
if os.path.exists(resolved):
|
|
return resolved
|
|
|
|
container_meta = "/app/merica-pipelines/metadata"
|
|
if os.path.exists(container_meta) and os.path.isdir(container_meta):
|
|
return container_meta
|
|
|
|
for rel_path in [
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "release", "merica-pipelines", "metadata")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "release", "merica-pipelines", "metadata")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipelines", "metadata")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipeline-bundles", "metadata")),
|
|
]:
|
|
if os.path.exists(rel_path) and os.path.isdir(rel_path):
|
|
return rel_path
|
|
|
|
for bundle_meta in sorted(glob.glob("/opt/airflow/dag_bundles/*/tracking_repo/metadata")):
|
|
if os.path.exists(bundle_meta):
|
|
return bundle_meta
|
|
|
|
bundle_meta = os.path.abspath(os.path.join(os.getenv("AIRFLOW_DAG_BUNDLE_PATH", "/opt/airflow/dag_bundles/merica-pipeline/tracking_repo"), "metadata"))
|
|
if os.path.exists(bundle_meta):
|
|
return bundle_meta
|
|
|
|
def resolve_dbt_dir() -> str:
|
|
"""Resolve dbt transformation project directory with SSOT priority:
|
|
1. Explicit custom AIRFLOW_DBT_DIR (if valid and exists)
|
|
2. Container mount /app/merica-pipelines/dbt
|
|
3. Host repo ../release/merica-pipelines/dbt
|
|
4. Sibling repo ../merica-pipelines/dbt or ../merica-pipeline-bundles/dbt
|
|
5. Local fallback directory airflow/dbt/dlh_transformation
|
|
"""
|
|
env_val = os.getenv("AIRFLOW_DBT_DIR")
|
|
if env_val:
|
|
resolved = resolve_dir_path("AIRFLOW_DBT_DIR", env_val)
|
|
if os.path.exists(resolved) and os.path.isdir(resolved):
|
|
return resolved
|
|
|
|
container_dbt = "/app/merica-pipelines/dbt"
|
|
if os.path.exists(container_dbt) and os.path.isdir(container_dbt):
|
|
return container_dbt
|
|
|
|
for rel_path in [
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "release", "merica-pipelines", "dbt")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "release", "merica-pipelines", "dbt")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipelines", "dbt")),
|
|
os.path.abspath(os.path.join(PROJECT_ROOT, "..", "merica-pipeline-bundles", "dbt")),
|
|
]:
|
|
if os.path.exists(rel_path) and os.path.isdir(rel_path):
|
|
return rel_path
|
|
|
|
return resolve_dir_path("AIRFLOW_DBT_DIR", os.path.join("airflow", "dbt", "dlh_transformation"))
|
|
|
|
|
|
|
|
class Settings:
|
|
PROJECT_NAME: str = "DLH Data Lakehouse - Dynamic Pipeline & Storage API"
|
|
VERSION: str = "2.0.0"
|
|
API_V1_PREFIX: str = "/api/v1"
|
|
|
|
# Path settings (SSOT decoupled with merica-pipeline-bundles)
|
|
CONFIGS_DIR: str = resolve_configs_dir()
|
|
METADATA_DIR: str = resolve_metadata_dir()
|
|
UPLOADS_DIR: str = os.path.abspath(os.path.join(PROJECT_ROOT, "airflow", "configs", "uploads"))
|
|
DOCS_DIR: str = resolve_dir_path("FRONTEND_DOCS_DIR", os.path.join("frontend", "src", "docs"))
|
|
|
|
DEFAULT_DBT_DIR: str = os.path.abspath(os.path.join(PROJECT_ROOT, "airflow", "dbt", "dlh_transformation"))
|
|
DBT_DIR: str = resolve_dbt_dir()
|
|
|
|
# Database
|
|
AIRFLOW_DATABASE_URL: str = os.getenv(
|
|
"AIRFLOW__DATABASE__SQL_ALCHEMY_CONN",
|
|
"postgresql+psycopg2://dlh_user:dlh_password@postgres:5432/dlh_lakehouse"
|
|
)
|
|
|
|
# CORS
|
|
CORS_ORIGINS: List[str] = ["*"]
|
|
|
|
# Security & JWT Auth
|
|
SECRET_KEY: str = os.getenv("SECRET_KEY", "merica-secret-key-dlh-12345678901234567890123456789012")
|
|
ALGORITHM: str = "HS256"
|
|
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440")) # 24 hours
|
|
|
|
# Local Admin Fallback Credentials (Zero-SSO emergency access)
|
|
ADMIN_USERNAME: str = os.getenv("MERICA_ADMIN_USERNAME", "admin")
|
|
ADMIN_PASSWORD: str = os.getenv("MERICA_ADMIN_PASSWORD", "admin")
|
|
|
|
# Keycloak OIDC Settings
|
|
KEYCLOAK_URL: str = os.getenv("KEYCLOAK_URL", "http://localhost:14180")
|
|
KEYCLOAK_INTERNAL_URL: str = os.getenv("KEYCLOAK_INTERNAL_URL", os.getenv("KEYCLOAK_URL", "http://localhost:14180"))
|
|
KEYCLOAK_REALM: str = os.getenv("KEYCLOAK_REALM", "merica")
|
|
KEYCLOAK_CLIENT_ID: str = os.getenv("KEYCLOAK_CLIENT_ID", "merica-app")
|
|
|
|
# Default Dynamic Role & Group Mappings (Fallbacks if not in DB/JSON)
|
|
AUTH_ADMIN_ROLES: List[str] = [r.strip().lower() for r in os.getenv("AUTH_ADMIN_ROLES", "merica_admin,admin,administrator,dlh_admin,role_admin,superuser").split(",") if r.strip()]
|
|
AUTH_ADMIN_GROUPS: List[str] = [g.strip().lower() for g in os.getenv("AUTH_ADMIN_GROUPS", "/platform administrators,/admins,/dlh/admins").split(",") if g.strip()]
|
|
|
|
AUTH_EDITOR_ROLES: List[str] = [r.strip().lower() for r in os.getenv("AUTH_EDITOR_ROLES", "data_engineer,engineer,editor,dlh_engineer,role_data_engineer,pranata_komputer,pipeline_editor,de_bpkp").split(",") if r.strip()]
|
|
AUTH_EDITOR_GROUPS: List[str] = [g.strip().lower() for g in os.getenv("AUTH_EDITOR_GROUPS", "/data engineers,/ingestion engineers,/tim pengembang,/dlh/engineers").split(",") if g.strip()]
|
|
|
|
AUTH_VIEWER_ROLES: List[str] = [r.strip().lower() for r in os.getenv("AUTH_VIEWER_ROLES", "merica_viewer,viewer,auditor,pimpinan,analyst,dlh_auditor,role_viewer,pipeline_viewer,pengawas").split(",") if r.strip()]
|
|
AUTH_VIEWER_GROUPS: List[str] = [g.strip().lower() for g in os.getenv("AUTH_VIEWER_GROUPS", "/pimpinan unit kerja,/auditor internal bpkp,/data analysts,/auditors,/dlh/viewers").split(",") if g.strip()]
|
|
|
|
|
|
settings = Settings()
|
|
for _dir_path in (settings.CONFIGS_DIR, settings.METADATA_DIR, settings.UPLOADS_DIR):
|
|
if not _dir_path.startswith("/opt/airflow/dag_bundles"):
|
|
try:
|
|
os.makedirs(_dir_path, exist_ok=True)
|
|
except Exception:
|
|
pass
|