feat: Implement initial setup for Financial Crime domain
- Added Makefile for orchestrating domain instantiation and management. - Created deploy README for guiding the setup process of the Financial Crime domain. - Introduced domain configuration file for mapping metadata to the deployment template. - Implemented pre-commit hook for validating metadata before commits. - Developed instantiation script to manage the lifecycle of the Financial Crime domain. - Added Salesforce CRM source with associated metadata and transformation files. - Added SAP Fraud Management source with associated metadata and transformation files. - Added Temenos Payment source with associated metadata and transformation files. - Removed obsolete payment event and parties transformation files.
This commit is contained in:
Executable
+339
@@ -0,0 +1,339 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Metadata-driven instantiation of the Financial Crime data domain.
|
||||
|
||||
This driver turns the MD-DDL metadata in this repo into a running, registered
|
||||
domain by orchestrating the semprini-data-domain template scripts. It does not
|
||||
re-implement them — it wires this repo as the template's "data root", ensures
|
||||
secrets, then calls new-domain.sh / docker compose / domain-integration.sh in
|
||||
the right order.
|
||||
|
||||
Usage:
|
||||
python3 deploy/instantiate.py <step> [<step> ...]
|
||||
python3 deploy/instantiate.py all # full stand-up
|
||||
python3 deploy/instantiate.py teardown # reverse it
|
||||
|
||||
Steps (also runnable individually):
|
||||
preflight validate metadata (blocks on findings)
|
||||
wire create repo-local symlinks the template expects
|
||||
secrets ensure domain secrets exist in the shared .env
|
||||
scaffold new-domain.sh -> domains/<domain>/
|
||||
build build the postgres-ducklake + ariadne images
|
||||
up bring the stack up; wait for bucket + postgres health
|
||||
ducklake attach the DuckLake catalog in the analytics DB
|
||||
register register with core (Keycloak/DNS/Kuma/Prometheus) + mgmt (OpenMetadata)
|
||||
activate restart ariadne + flink to pick up OIDC + metrics
|
||||
verify health-check the running domain
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets as secretslib
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
CONFIG_PATH = REPO_ROOT / "deploy" / "domain.config.yml"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
def log(msg: str) -> None:
|
||||
print(f"\033[1;36m›\033[0m {msg}", flush=True)
|
||||
|
||||
|
||||
def ok(msg: str) -> None:
|
||||
print(f" \033[1;32m✓\033[0m {msg}", flush=True)
|
||||
|
||||
|
||||
def die(msg: str) -> None:
|
||||
print(f"\033[1;31m✗ {msg}\033[0m", file=sys.stderr, flush=True)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def run(cmd: list[str], cwd: Path | None = None, check: bool = True,
|
||||
capture: bool = False) -> subprocess.CompletedProcess:
|
||||
printable = " ".join(cmd)
|
||||
log(f"$ {printable}")
|
||||
return subprocess.run(
|
||||
cmd, cwd=str(cwd or REPO_ROOT), check=check,
|
||||
text=True, capture_output=capture,
|
||||
)
|
||||
|
||||
|
||||
class Cfg:
|
||||
def __init__(self, raw: dict):
|
||||
self.metadata_dir: str = raw["metadata_dir"]
|
||||
self.domain: str = raw["domain"]
|
||||
self.port_base: int = int(raw["port_base"])
|
||||
self.template_dir: str = raw["template_dir"]
|
||||
self.platform_repo: str = raw["platform_repo"]
|
||||
self.secrets: list[str] = list(raw["secrets"])
|
||||
|
||||
@property
|
||||
def upper(self) -> str:
|
||||
return self.domain.replace("-", "_").upper()
|
||||
|
||||
@property
|
||||
def template(self) -> Path:
|
||||
return REPO_ROOT / self.template_dir
|
||||
|
||||
@property
|
||||
def compose(self) -> Path:
|
||||
return REPO_ROOT / "domains" / self.domain / "compose.yml"
|
||||
|
||||
@property
|
||||
def env_file(self) -> Path:
|
||||
return REPO_ROOT / ".env"
|
||||
|
||||
|
||||
def load_cfg() -> Cfg:
|
||||
if not CONFIG_PATH.exists():
|
||||
die(f"config not found: {CONFIG_PATH}")
|
||||
return Cfg(yaml.safe_load(CONFIG_PATH.read_text()))
|
||||
|
||||
|
||||
def compose_cmd(cfg: Cfg, *args: str) -> list[str]:
|
||||
return ["docker", "compose", "-f", str(cfg.compose), *args]
|
||||
|
||||
|
||||
def env_value(cfg: Cfg, key: str) -> str | None:
|
||||
for line in cfg.env_file.read_text().splitlines():
|
||||
if line.startswith(f"{key}="):
|
||||
return line.split("=", 1)[1]
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# steps
|
||||
# --------------------------------------------------------------------------- #
|
||||
def step_preflight(cfg: Cfg) -> None:
|
||||
log("preflight: validating MD-DDL metadata")
|
||||
checker = REPO_ROOT / ".github" / "scripts" / "preflight.py"
|
||||
run([sys.executable, str(checker), cfg.metadata_dir])
|
||||
ok("metadata pre-flight passed")
|
||||
|
||||
|
||||
def step_wire(cfg: Cfg) -> None:
|
||||
"""Repo-local symlinks so the template treats this repo as the data root."""
|
||||
log("wire: creating repo-local symlinks")
|
||||
links = {
|
||||
# compose build contexts use ../../semprini-data-domain
|
||||
"semprini-data-domain": cfg.template_dir,
|
||||
# shared platform secrets + admin creds
|
||||
".env": f"{cfg.platform_repo}/.env",
|
||||
# domain-integration.sh sources lib/core-integration.sh
|
||||
"lib": f"{cfg.platform_repo}/lib",
|
||||
}
|
||||
for name, target in links.items():
|
||||
link = REPO_ROOT / name
|
||||
if link.is_symlink() or link.exists():
|
||||
ok(f"{name} -> {os.readlink(link) if link.is_symlink() else '(exists)'}")
|
||||
continue
|
||||
link.symlink_to(target)
|
||||
ok(f"created {name} -> {target}")
|
||||
# ensure the bind-mount dirs ariadne expects are present
|
||||
for d in ("schemas", "resolvers", "flink-jobs"):
|
||||
(REPO_ROOT / "domains" / cfg.domain / d).mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def step_secrets(cfg: Cfg) -> None:
|
||||
log("secrets: ensuring domain secrets in shared .env")
|
||||
env_file = cfg.env_file
|
||||
if not env_file.exists():
|
||||
die(f"{env_file} missing (run 'wire' first; check {cfg.platform_repo}/.env)")
|
||||
existing = {
|
||||
line.split("=", 1)[0]
|
||||
for line in env_file.read_text().splitlines()
|
||||
if "=" in line and not line.startswith("#")
|
||||
}
|
||||
to_add: list[str] = []
|
||||
for key in cfg.secrets:
|
||||
if key in existing:
|
||||
ok(f"{key} present")
|
||||
continue
|
||||
if "ACCESS_KEY" in key:
|
||||
val = "fincrime" + secretslib.token_hex(8)
|
||||
else:
|
||||
val = secretslib.token_urlsafe(24)
|
||||
to_add.append(f"{key}={val}")
|
||||
ok(f"generated {key}")
|
||||
port_key = f"{cfg.upper}_PORT_BASE"
|
||||
if port_key not in existing:
|
||||
to_add.append(f"{port_key}={cfg.port_base}")
|
||||
ok(f"set {port_key}={cfg.port_base}")
|
||||
if to_add:
|
||||
with env_file.open("a") as fh:
|
||||
fh.write("\n# --- " + cfg.domain + " domain secrets ---\n")
|
||||
fh.write("\n".join(to_add) + "\n")
|
||||
|
||||
|
||||
def step_scaffold(cfg: Cfg) -> None:
|
||||
log(f"scaffold: new-domain.sh {cfg.domain} {cfg.port_base}")
|
||||
if cfg.compose.exists():
|
||||
ok(f"already scaffolded at {cfg.compose.parent}")
|
||||
return
|
||||
run([str(cfg.template / "new-domain.sh"), cfg.domain, str(cfg.port_base)])
|
||||
ok("domain scaffolded")
|
||||
|
||||
|
||||
def step_build(cfg: Cfg) -> None:
|
||||
log("build: postgres-ducklake + ariadne images")
|
||||
run(compose_cmd(cfg, "build",
|
||||
f"postgres-{cfg.domain}", f"ariadne-{cfg.domain}"))
|
||||
ok("images built")
|
||||
|
||||
|
||||
def _wait(cfg: Cfg, desc: str, check_fn, timeout: int = 180, interval: int = 5) -> None:
|
||||
log(f"waiting for {desc} (≤{timeout}s)")
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
if check_fn():
|
||||
ok(desc)
|
||||
return
|
||||
time.sleep(interval)
|
||||
die(f"timed out waiting for {desc}")
|
||||
|
||||
|
||||
def step_up(cfg: Cfg) -> None:
|
||||
log("up: starting the domain stack")
|
||||
run(compose_cmd(cfg, "up", "-d"))
|
||||
|
||||
def bucket_ready() -> bool:
|
||||
r = subprocess.run(
|
||||
["docker", "inspect", "-f", "{{.State.Status}}:{{.State.ExitCode}}",
|
||||
f"minio-init-{cfg.domain}"],
|
||||
text=True, capture_output=True,
|
||||
)
|
||||
return r.stdout.strip() == "exited:0"
|
||||
|
||||
def pg_ready() -> bool:
|
||||
r = subprocess.run(
|
||||
["docker", "inspect", "-f", "{{.State.Health.Status}}",
|
||||
f"postgres-{cfg.domain}"],
|
||||
text=True, capture_output=True,
|
||||
)
|
||||
return r.stdout.strip() == "healthy"
|
||||
|
||||
_wait(cfg, f"{cfg.domain}-iceberg bucket (minio-init exit 0)", bucket_ready)
|
||||
_wait(cfg, f"postgres-{cfg.domain} healthy", pg_ready)
|
||||
|
||||
|
||||
def step_ducklake(cfg: Cfg) -> None:
|
||||
"""Post-startup: attach the DuckLake catalog inside the analytics DB."""
|
||||
log("ducklake: attaching catalog in analytics DB")
|
||||
access = env_value(cfg, f"MINIO_{cfg.upper}_ACCESS_KEY")
|
||||
secret = env_value(cfg, f"MINIO_{cfg.upper}_SECRET_KEY")
|
||||
pg_pw = env_value(cfg, f"POSTGRES_{cfg.upper}_PASSWORD")
|
||||
if not all([access, secret, pg_pw]):
|
||||
die("missing MinIO/postgres secrets in .env for analytics attach")
|
||||
sql = f"""
|
||||
SELECT duckdb.create_secret(
|
||||
'{cfg.domain}_minio', 'S3',
|
||||
key_id := '{access}', secret := '{secret}',
|
||||
endpoint := 'minio-{cfg.domain}:9000', url_style := 'path', use_ssl := false
|
||||
);
|
||||
SELECT duckdb.raw_query($q$
|
||||
ATTACH 'ducklake:postgres:host=postgres-{cfg.domain} port=5432 dbname={cfg.domain} user={cfg.domain} password={pg_pw}'
|
||||
AS "{cfg.domain}_lake";
|
||||
$q$);
|
||||
"""
|
||||
# analytics service runs as user 'analytics', db '<domain>'
|
||||
run([
|
||||
"docker", "exec", "-i", f"analytics-{cfg.domain}",
|
||||
"psql", "-v", "ON_ERROR_STOP=1", "-U", "analytics", "-d", cfg.domain,
|
||||
] + ["-c", sql], check=False)
|
||||
ok("analytics DuckLake attach attempted (idempotent; safe to re-run)")
|
||||
|
||||
|
||||
def step_register(cfg: Cfg) -> None:
|
||||
log("register: core + management (reversible, one-way)")
|
||||
run([str(cfg.template / "domain-integration.sh"), cfg.domain, "register", "all"])
|
||||
ok("registered with core + management")
|
||||
|
||||
|
||||
def step_activate(cfg: Cfg) -> None:
|
||||
log("activate: restart ariadne + flink to pick up OIDC + metrics")
|
||||
run(compose_cmd(cfg, "restart",
|
||||
f"ariadne-{cfg.domain}", f"flink-jm-{cfg.domain}"))
|
||||
ok("ariadne + flink restarted")
|
||||
|
||||
|
||||
def step_verify(cfg: Cfg) -> None:
|
||||
log("verify: checking domain stability")
|
||||
expected = [f"{p}-{cfg.domain}" for p in
|
||||
("postgres", "kafka", "flink-jm", "flink-tm", "minio", "analytics", "ariadne")]
|
||||
r = subprocess.run(["docker", "ps", "--format", "{{.Names}}"],
|
||||
text=True, capture_output=True)
|
||||
running = set(r.stdout.split())
|
||||
missing = [c for c in expected if c not in running]
|
||||
if missing:
|
||||
die(f"services not running: {', '.join(missing)}")
|
||||
ok(f"{len(expected)} services running: {', '.join(expected)}")
|
||||
|
||||
# MinIO bucket
|
||||
rc = subprocess.run(
|
||||
["docker", "exec", f"minio-{cfg.domain}", "sh", "-c",
|
||||
f"mc alias set local http://localhost:9000 $MINIO_ROOT_USER $MINIO_ROOT_PASSWORD >/dev/null 2>&1; "
|
||||
f"mc ls local/{cfg.domain}-iceberg >/dev/null 2>&1 && echo OK"],
|
||||
text=True, capture_output=True,
|
||||
)
|
||||
ok(f"{cfg.domain}-iceberg bucket present") if "OK" in rc.stdout else \
|
||||
print(f" ! could not confirm bucket (non-fatal): {rc.stderr.strip()}")
|
||||
|
||||
print("\n Next (VPN required) — manual spot checks:")
|
||||
print(f" GraphQL : http://api.{cfg.domain}.data.internal:{cfg.port_base}/health")
|
||||
print(f" Flink : http://flink.{cfg.domain}.data.internal:{cfg.port_base + 1}")
|
||||
print(f" Keycloak: client data-{cfg.domain} + /data-{cfg.domain}/* groups")
|
||||
print(f" Catalog : OpenMetadata domain '{cfg.domain}'")
|
||||
|
||||
|
||||
def step_teardown(cfg: Cfg) -> None:
|
||||
log("teardown: unregister + stop (reverses the stand-up)")
|
||||
run([str(cfg.template / "domain-integration.sh"), cfg.domain, "unregister", "all"],
|
||||
check=False)
|
||||
run(compose_cmd(cfg, "down", "-v"), check=False)
|
||||
ok("domain unregistered + stopped")
|
||||
|
||||
|
||||
STEPS = {
|
||||
"preflight": step_preflight,
|
||||
"wire": step_wire,
|
||||
"secrets": step_secrets,
|
||||
"scaffold": step_scaffold,
|
||||
"build": step_build,
|
||||
"up": step_up,
|
||||
"ducklake": step_ducklake,
|
||||
"register": step_register,
|
||||
"activate": step_activate,
|
||||
"verify": step_verify,
|
||||
"teardown": step_teardown,
|
||||
}
|
||||
|
||||
ALL = ["preflight", "wire", "secrets", "scaffold", "build", "up",
|
||||
"ducklake", "register", "activate", "verify"]
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = sys.argv[1:] or ["all"]
|
||||
requested: list[str] = []
|
||||
for a in args:
|
||||
if a == "all":
|
||||
requested.extend(ALL)
|
||||
elif a in STEPS:
|
||||
requested.append(a)
|
||||
else:
|
||||
die(f"unknown step '{a}'. valid: {', '.join(STEPS)}, all")
|
||||
cfg = load_cfg()
|
||||
log(f"domain={cfg.domain} metadata={cfg.metadata_dir} port_base={cfg.port_base}")
|
||||
for name in requested:
|
||||
STEPS[name](cfg)
|
||||
log("done")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user