#!/usr/bin/env python3 """Automated deployment script for LeoCRM via Coolify API. All container management is done through the Coolify API — no manual docker commands, no SSH for container lifecycle. SSH is used *only* for post-deploy verification (Alembic version, RLS table count) because the Coolify API does not expose database internals. Usage: python scripts/deploy.py # Full deploy (API + Worker) python scripts/deploy.py --skip-build # Skip build, just restart python scripts/deploy.py --worker-only # Only deploy worker service python scripts/deploy.py --verify-only # Only run verification Environment variables: COOLIFY_API_TOKEN — Coolify API token (required) COOLIFY_APP_UUID — Application UUID (default: stvabl4vaqru7jclx4ittzr3) COOLIFY_WORKER_UUID — Worker Service UUID (default: asxqaq3566to108xordck0ff) COOLIFY_BASE_URL — Coolify base URL (default: https://server.media-on.de) SSH_KEY — SSH key path for verification (default: /a0/usr/workdir/.ssh/coolify-01-root) SERVER_IP — Server IP for SSH verification (default: 46.225.91.159) Exit codes: 0 — deployment successful 1 — deployment failed 2 — configuration error """ from __future__ import annotations import argparse import base64 import os import subprocess import sys import time from dataclasses import dataclass, field from typing import Any import httpx # ─── Configuration ──────────────────────────────────────────────────── COOLIFY_BASE_URL = os.environ.get("COOLIFY_BASE_URL", "https://server.media-on.de") COOLIFY_TOKEN = os.environ.get("COOLIFY_API_TOKEN", "") APP_UUID = os.environ.get("COOLIFY_APP_UUID", "stvabl4vaqru7jclx4ittzr3") WORKER_UUID = os.environ.get("COOLIFY_WORKER_UUID", "asxqaq3566to108xordck0ff") SSH_KEY = os.environ.get("SSH_KEY", "/a0/usr/workdir/.ssh/coolify-01-root") SERVER_IP = os.environ.get("SERVER_IP", "46.225.91.159") # Domain for HTTP health / login checks APP_DOMAIN = os.environ.get("APP_DOMAIN", "https://crm.media-on.de") # Login test credentials (read-only verification) LOGIN_EMAIL = os.environ.get("LOGIN_EMAIL", "admin@media-on.de") LOGIN_PASSWORD = os.environ.get("LOGIN_PASSWORD", "Admin123!") # Worker docker-compose definition using ${VARIABLE} syntax. # Secrets are managed via Coolify's ENV API (POST/PATCH /services/{uuid}/envs). # Coolify auto-generates the .env file from these variables during deploy. WORKER_COMPOSE_YAML = """\ services: worker: image: 'stvabl4vaqru7jclx4ittzr3:latest' restart: unless-stopped entrypoint: - /app/worker.sh environment: DATABASE_URL: 'postgresql+asyncpg://crm_worker:${DB_PASSWORD}@crm-postgres:5432/crm_db' WORKER_DATABASE_URL: 'postgresql+asyncpg://crm_worker:${DB_PASSWORD}@crm-postgres:5432/crm_db' MIGRATION_DATABASE_URL: 'postgresql+asyncpg://crm_user:${DB_PASSWORD}@crm-postgres:5432/crm_db' AUTH_DATABASE_URL: 'postgresql+asyncpg://crm_auth:${DB_PASSWORD}@crm-postgres:5432/crm_db' REDIS_URL: 'redis://default:${REDIS_PASSWORD}@crm-redis:6379/0' SECRET_KEY: ${SECRET_KEY} ENVIRONMENT: ${ENVIRONMENT} STORAGE_PATH: ${STORAGE_PATH} COOLIFY_RESOURCE_UUID: asxqaq3566to108xordck0ff COOLIFY_CONTAINER_NAME: worker-asxqaq3566to108xordck0ff SERVICE_NAME_WORKER: worker volumes: - 'asxqaq3566to108xordck0ff_leocrm-worker-storage:/data/storage' networks: - coolify - asxqaq3566to108xordck0ff container_name: worker-asxqaq3566to108xordck0ff labels: - coolify.managed=true - coolify.version=4.0.0-beta.470 - coolify.serviceId=277 - coolify.type=service - coolify.name=worker-asxqaq3566to108xordck0ff - coolify.resourceName=leocrm-worker - coolify.projectName=crm - coolify.serviceName=worker - coolify.environmentName=production - coolify.pullRequestId=0 - coolify.service.subId=500 - coolify.service.subType=application - coolify.service.subName=worker volumes: leocrm-worker-storage: name: leocrm-worker-storage asxqaq3566to108xordck0ff_leocrm-worker-storage: name: asxqaq3566to108xordck0ff_leocrm-worker-storage networks: coolify: external: true name: coolify asxqaq3566to108xordck0ff: name: asxqaq3566to108xordck0ff external: true """ # Worker ENV variables to set via Coolify API. # For different deployments (dev/staging/prod), change these values. WORKER_ENVS = [ {"key": "DB_PASSWORD", "value": "4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI"}, {"key": "REDIS_PASSWORD", "value": "lAjCaTf3XFP5XSaPJ1HElgLAJhQQswLT"}, {"key": "SECRET_KEY", "value": "vVdAnvyc-ob4myE5D1rAYn-SovzoBfQLP1z4wmWteTmFPV_lveCGIn2upNoiP590"}, {"key": "ENVIRONMENT", "value": "production"}, {"key": "STORAGE_PATH", "value": "/data/storage"}, ] # ─── Data Structures ────────────────────────────────────────────────── @dataclass class StepResult: success: bool message: str duration_s: float = 0.0 details: dict[str, Any] = field(default_factory=dict) # ─── Coolify API Client ──────────────────────────────────────────────── class CoolifyClient: """Client for Coolify API operations.""" def __init__(self, base_url: str, token: str): self.base_url = base_url.rstrip("/") self.headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", } # ── Applications ── def deploy_application(self, app_uuid: str) -> dict[str, Any]: """Trigger a build & deploy for an application via Coolify API.""" resp = httpx.post( f"{self.base_url}/api/v1/deploy", headers=self.headers, json={"uuid": app_uuid}, timeout=30, ) resp.raise_for_status() return resp.json() def get_application(self, app_uuid: str) -> dict[str, Any]: """Get application details including status.""" resp = httpx.get( f"{self.base_url}/api/v1/applications/{app_uuid}", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() # ── Services (Worker) ── def get_service(self, service_uuid: str) -> dict[str, Any]: """Get service details including status.""" resp = httpx.get( f"{self.base_url}/api/v1/services/{service_uuid}", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() def update_service(self, service_uuid: str, docker_compose_raw: str) -> dict[str, Any]: """Update a service's docker_compose_raw (base64-encoded).""" encoded = base64.b64encode(docker_compose_raw.encode()).decode() resp = httpx.patch( f"{self.base_url}/api/v1/services/{service_uuid}", headers=self.headers, json={"docker_compose_raw": encoded}, timeout=30, ) resp.raise_for_status() return resp.json() def start_service(self, service_uuid: str) -> dict[str, Any]: """Start a service.""" resp = httpx.post( f"{self.base_url}/api/v1/services/{service_uuid}/start", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() def stop_service(self, service_uuid: str) -> dict[str, Any]: """Stop a service.""" resp = httpx.post( f"{self.base_url}/api/v1/services/{service_uuid}/stop", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() def restart_service(self, service_uuid: str) -> dict[str, Any]: """Restart a service.""" resp = httpx.post( f"{self.base_url}/api/v1/services/{service_uuid}/restart", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() # ── Deployments ── def get_deployment(self, deployment_uuid: str) -> dict[str, Any]: """Get deployment status.""" resp = httpx.get( f"{self.base_url}/api/v1/deployments/{deployment_uuid}", headers=self.headers, timeout=30, ) resp.raise_for_status() return resp.json() # ── Health ── def get_health(self) -> dict[str, Any]: """Check Coolify system health.""" resp = httpx.get( f"{self.base_url}/api/v1/health", headers=self.headers, timeout=15, ) resp.raise_for_status() return resp.json() # ─── SSH Helper (verification only) ──────────────────────────────────── def ssh_run(cmd: str, timeout: int = 60) -> tuple[int, str]: """Run a command on the server via SSH — used ONLY for verification.""" full_cmd = [ "ssh", "-i", SSH_KEY, "-o", "StrictHostKeyChecking=no", "-o", "ConnectTimeout=10", f"root@{SERVER_IP}", cmd, ] result = subprocess.run(full_cmd, capture_output=True, text=True, timeout=timeout) return result.returncode, result.stdout + result.stderr # ─── Deploy Steps ────────────────────────────────────────────────────── def deploy_api(client: CoolifyClient, skip_build: bool = False) -> StepResult: """Deploy or restart the API application via Coolify API.""" if skip_build: print(" Skip-build mode: restarting application via Coolify API...") # For skip-build we still trigger a deploy — Coolify will use cached image try: result = client.deploy_application(APP_UUID) deploy_uuid = _extract_deploy_uuid(result) if deploy_uuid: print(f" Deploy queued: {deploy_uuid[:12]}") return _wait_deployment(client, deploy_uuid, timeout=300) return StepResult(True, "Deploy triggered (no UUID returned)") except Exception as e: return StepResult(False, f"Deploy trigger failed: {e}") print(" Triggering Coolify build & deploy for API...") try: result = client.deploy_application(APP_UUID) deploy_uuid = _extract_deploy_uuid(result) if not deploy_uuid: return StepResult(False, "No deployment UUID returned from Coolify") print(f" Deploy queued: {deploy_uuid[:12]}") return _wait_deployment(client, deploy_uuid, timeout=300) except Exception as e: return StepResult(False, f"Deploy trigger failed: {e}") def deploy_worker(client: CoolifyClient, skip_build: bool = False) -> StepResult: """Deploy the worker service via Coolify API. Steps: 1. Write .env file to server (secrets for ${VARIABLE} substitution) 2. Update service compose (with ${VARIABLE} syntax, not hardcoded secrets) 3. Set connect_to_docker_network=True (coolify network for Redis/Postgres) 4. Tag latest API image as :latest (Coolify uses commit-hash tags) 5. Deploy via POST /deploy (creates new container) 6. Wait for healthy """ print(" Deploying worker service via Coolify API...") try: # Step 1: Update service compose with ${VARIABLE} syntax print(" Updating worker service compose...") client.update_service(WORKER_UUID, WORKER_COMPOSE_YAML) time.sleep(2) # Step 2: Set connect_to_docker_network=True print(" Ensuring coolify network connection...") import httpx resp = httpx.patch( f"{client.base_url}/api/v1/services/{WORKER_UUID}", headers=client.headers, json={"connect_to_docker_network": True}, timeout=30, ) if resp.status_code != 200: print(f" Warning: could not set connect_to_docker_network ({resp.status_code})") # Step 3: Set ENV variables via Coolify API (Coolify auto-generates .env) print(" Setting ENV variables via Coolify API...") for env in WORKER_ENVS: # POST creates the ENV variable; if it already exists (409), # PATCH updates it resp = httpx.post( f"{client.base_url}/api/v1/services/{WORKER_UUID}/envs", headers=client.headers, json=env, timeout=30, ) if resp.status_code == 409: # Already exists — update via PATCH resp = httpx.patch( f"{client.base_url}/api/v1/services/{WORKER_UUID}/envs", headers=client.headers, json=env, timeout=30, ) if resp.status_code not in (200, 201): print(f" Warning: could not set ENV {env['key']} ({resp.status_code})") # Step 4: Tag the latest API image as :latest print(" Tagging latest API image as :latest...") tag_code, tag_output = ssh_run( 'docker images --format "{{.Repository}}:{{.Tag}}" | ' 'grep "^stvabl4vaqru7jclx4ittzr3:" | grep -v latest | head -1 | ' 'xargs -I{} docker tag {} stvabl4vaqru7jclx4ittzr3:latest' ) if tag_code != 0: print(f" Warning: could not tag :latest ({tag_output.strip()})") # Step 5: Deploy via POST /deploy print(" Deploying worker service...") result = client.deploy_application(WORKER_UUID) deploy_uuid = _extract_deploy_uuid(result) if deploy_uuid: print(f" Worker deploy queued: {deploy_uuid[:12]}") dep_result = _wait_deployment(client, deploy_uuid, timeout=120) if not dep_result.success: return dep_result else: print(" No deployment UUID returned, waiting for healthy...") # Step 6: Wait for healthy return _wait_service_healthy(client, WORKER_UUID, timeout=120) except Exception as e: return StepResult(False, f"Worker deploy failed: {e}") def _extract_deploy_uuid(result: dict[str, Any]) -> str | None: """Extract deployment UUID from Coolify deploy response.""" # Coolify returns {"deployments": [{"deployment_uuid": "..."}]} deployments = result.get("deployments", []) if deployments and isinstance(deployments, list): return deployments[0].get("deployment_uuid") # Some versions return {"deployment_uuid": "..."} directly return result.get("deployment_uuid") def _wait_deployment(client: CoolifyClient, deploy_uuid: str, timeout: int = 300) -> StepResult: """Wait for a Coolify deployment to reach success/failed status.""" print(f" Waiting for deployment {deploy_uuid[:12]}...") start = time.time() while time.time() - start < timeout: try: dep = client.get_deployment(deploy_uuid) status = dep.get("status", "unknown") elapsed = int(time.time() - start) print(f" [{elapsed}s] Deployment status: {status}") if status in ("success", "finished"): return StepResult(True, "Deployment successful", time.time() - start, dep) if status == "failed": return StepResult(False, f"Deployment failed: {dep.get('message', 'unknown')}", time.time() - start, dep) except Exception as e: print(f" Warning: API error: {e}") time.sleep(10) return StepResult(False, f"Deployment timed out after {timeout}s", time.time() - start) def _wait_service_healthy(client: CoolifyClient, service_uuid: str, timeout: int = 120) -> StepResult: """Wait for a Coolify service to reach running:healthy status.""" print(f" Waiting for service {service_uuid[:12]} to become healthy...") start = time.time() while time.time() - start < timeout: try: svc = client.get_service(service_uuid) status = svc.get("status", "unknown") elapsed = int(time.time() - start) print(f" [{elapsed}s] Service status: {status}") # Only accept exact 'running:healthy' or 'healthy' if status == "running:healthy" or status == "healthy": return StepResult(True, f"Service healthy: {status}", time.time() - start, svc) if "failed" in status.lower() or "error" in status.lower(): return StepResult(False, f"Service failed: {status}", time.time() - start, svc) except Exception as e: print(f" Warning: API error: {e}") time.sleep(5) return StepResult(False, f"Service did not become healthy in {timeout}s", time.time() - start) # ─── Verification ────────────────────────────────────────────────────── def verify_http_health() -> StepResult: """Verify the API is healthy via HTTP endpoint.""" print(" Verifying API health via HTTP...") url = f"{APP_DOMAIN}/api/v1/health" try: resp = httpx.get(url, timeout=30, follow_redirects=True) if resp.status_code == 200: body = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else resp.text if isinstance(body, dict) and body.get("status", "").lower() in ("healthy", "ok"): return StepResult(True, f"Health check passed: {body}") return StepResult(True, f"Health check HTTP 200: {body}") return StepResult(False, f"Health check failed: HTTP {resp.status_code}") except Exception as e: return StepResult(False, f"Health check error: {e}") def verify_login() -> StepResult: """Verify login works by sending a test login request.""" print(" Verifying login...") url = f"{APP_DOMAIN}/api/v1/auth/login" try: resp = httpx.post( url, json={"email": LOGIN_EMAIL, "password": LOGIN_PASSWORD}, headers={"Origin": APP_DOMAIN}, timeout=30, follow_redirects=True, ) if resp.status_code == 200: body = resp.json() token = body.get("access_token") or body.get("token") if token: return StepResult(True, "Login successful — token received") return StepResult(True, f"Login HTTP 200: {list(body.keys())}") if resp.status_code == 422: return StepResult(False, f"Login validation error (422): {resp.text[:200]}") return StepResult(False, f"Login failed: HTTP {resp.status_code}") except Exception as e: return StepResult(False, f"Login error: {e}") def verify_alembic() -> StepResult: """Verify Alembic migration head via SSH (Coolify API doesn't expose DB internals).""" print(" Verifying Alembic migration head...") code, output = ssh_run( 'docker exec crm-postgres psql -U crm_user -d crm_db -t -c ' '"SELECT version_num FROM alembic_version" 2>/dev/null' ) version = output.strip() if not version: return StepResult(False, "Could not read Alembic version") # Accept versions >= 0085 (migrations 0085-0090 handle RLS) version_ok = version >= "0085" return StepResult( version_ok, f"Alembic version: {version} ({'OK' if version_ok else 'BEHIND — expected >= 0085'})", details={"alembic_version": version}, ) def verify_rls() -> StepResult: """Verify RLS is active on tenant tables via SSH (read-only check).""" print(" Verifying RLS tables...") code, output = ssh_run( 'docker exec crm-postgres psql -U crm_user -d crm_db -t -c ' '"SELECT count(*) FROM pg_class WHERE relrowsecurity=true AND relforcerowsecurity=true" 2>/dev/null' ) rls_count = output.strip() if not rls_count.isdigit(): return StepResult(False, f"Could not read RLS table count: {output}") count = int(rls_count) # RLS should be active on all tenant tables (typically 90+) rls_ok = count >= 90 return StepResult( rls_ok, f"RLS tables: {count} ({'OK' if rls_ok else 'LOW — expected >= 90'})", details={"rls_tables": count}, ) def verify_worker_service(client: CoolifyClient) -> StepResult: """Verify worker service is running via Coolify API.""" print(" Verifying worker service via Coolify API...") try: svc = client.get_service(WORKER_UUID) status = svc.get("status", "unknown") status_lower = status.lower() # Accept 'running:healthy', 'running', 'healthy', or 'up' if "running" in status_lower or "healthy" in status_lower or status_lower == "up": return StepResult(True, f"Worker service: {status}", details={"status": status}) return StepResult(False, f"Worker service not running: {status}", details={"status": status}) except Exception as e: return StepResult(False, f"Worker service check failed: {e}") def run_verification(client: CoolifyClient, check_worker: bool = True) -> list[tuple[str, StepResult]]: """Run all verification checks and return results.""" results: list[tuple[str, StepResult]] = [] # 1. HTTP health print("\n[Verify] HTTP health check...") r = verify_http_health() results.append(("HTTP health", r)) _print_result(r) # 2. Login test print("\n[Verify] Login test...") r = verify_login() results.append(("Login test", r)) _print_result(r) # 3. Alembic version print("\n[Verify] Alembic migration head...") r = verify_alembic() results.append(("Alembic version", r)) _print_result(r) # 4. RLS tables print("\n[Verify] RLS tables...") r = verify_rls() results.append(("RLS tables", r)) _print_result(r) # 5. Worker service (optional) if check_worker: print("\n[Verify] Worker service status...") r = verify_worker_service(client) results.append(("Worker service", r)) _print_result(r) return results def _print_result(r: StepResult) -> None: if r.success: print(f" ✅ {r.message}") else: print(f" ❌ {r.message}") # ─── Summary ─────────────────────────────────────────────────────────── def print_summary(steps: list[tuple[str, StepResult]]) -> bool: """Print deployment summary and return overall success.""" print(f"\n{'='*60}") print(" Deploy Summary") print(f"{'='*60}") for name, result in steps: status = "✅" if result.success else "❌" print(f" {status} {name}: {result.message}") all_ok = all(r.success for _, r in steps) print(f"\n Overall: {'✅ SUCCESS' if all_ok else '❌ FAILED'}\n") return all_ok # ─── Main Deploy Pipeline ────────────────────────────────────────────── def deploy_full(skip_build: bool = False) -> int: """Full deploy: API + Worker + Verification.""" print(f"\n{'='*60}") print(" LeoCRM Full Deploy") print(f"{'='*60}\n") if not COOLIFY_TOKEN: print("ERROR: COOLIFY_API_TOKEN not set") return 2 client = CoolifyClient(COOLIFY_BASE_URL, COOLIFY_TOKEN) steps: list[tuple[str, StepResult]] = [] # Step 1: Deploy API print("\n[1/3] Deploying API via Coolify API...") r = deploy_api(client, skip_build=skip_build) steps.append(("API deploy", r)) _print_result(r) if not r.success: print_summary(steps) return 1 # Step 2: Deploy Worker print("\n[2/3] Deploying Worker via Coolify API...") r = deploy_worker(client, skip_build=skip_build) steps.append(("Worker deploy", r)) _print_result(r) # Worker failure is non-fatal but reported # Step 3: Verification print("\n[3/3] Running verification...") verify_results = run_verification(client, check_worker=True) steps.extend(verify_results) # Summary all_ok = print_summary(steps) return 0 if all_ok else 1 def deploy_worker_only(skip_build: bool = False) -> int: """Deploy only the worker service + verification.""" print(f"\n{'='*60}") print(" LeoCRM Worker-Only Deploy") print(f"{'='*60}\n") if not COOLIFY_TOKEN: print("ERROR: COOLIFY_API_TOKEN not set") return 2 client = CoolifyClient(COOLIFY_BASE_URL, COOLIFY_TOKEN) steps: list[tuple[str, StepResult]] = [] # Step 1: Deploy Worker print("\n[1/2] Deploying Worker via Coolify API...") r = deploy_worker(client, skip_build=skip_build) steps.append(("Worker deploy", r)) _print_result(r) if not r.success: print_summary(steps) return 1 # Step 2: Verification print("\n[2/2] Running verification...") verify_results = run_verification(client, check_worker=True) steps.extend(verify_results) all_ok = print_summary(steps) return 0 if all_ok else 1 def verify_only() -> int: """Run only verification checks.""" print(f"\n{'='*60}") print(" LeoCRM Verification Only") print(f"{'='*60}\n") if not COOLIFY_TOKEN: print("ERROR: COOLIFY_API_TOKEN not set") return 2 client = CoolifyClient(COOLIFY_BASE_URL, COOLIFY_TOKEN) results = run_verification(client, check_worker=True) all_ok = print_summary(results) return 0 if all_ok else 1 # ─── CLI ─────────────────────────────────────────────────────────────── # ─── Initial deployment: create all Coolify resources from scratch ────────── # PostgreSQL Compose (pgvector for embeddings) POSTGRES_COMPOSE = """\ services: postgres: image: pgvector/pgvector:pg16 restart: unless-stopped environment: POSTGRES_USER: ${DB_USER} POSTGRES_PASSWORD: ${DB_PASSWORD} POSTGRES_DB: ${DB_NAME} PGDATA: /var/lib/postgresql/data/pgdata volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: ['CMD-SHELL', 'pg_isready -U ${DB_USER} -d ${DB_NAME}'] interval: 10s timeout: 5s retries: 5 start_period: 10s volumes: pgdata: """ # Redis Compose REDIS_COMPOSE = """\ services: redis: image: redis:7-alpine restart: unless-stopped command: redis-server --requirepass ${REDIS_PASSWORD} volumes: - redisdata:/data healthcheck: test: ['CMD-SHELL', 'redis-cli ping || exit 1'] interval: 10s timeout: 5s retries: 5 volumes: redisdata: """ # ENV variables for PostgreSQL service POSTGRES_ENVS = [ {"key": "DB_USER", "value": "crm_user"}, {"key": "DB_PASSWORD", "value": "4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI"}, {"key": "DB_NAME", "value": "crm_db"}, ] # ENV variables for Redis service REDIS_ENVS = [ {"key": "REDIS_PASSWORD", "value": "lAjCaTf3XFP5XSaPJ1HElgLAJhQQswLT"}, ] # API Application configuration API_GIT_REPO = os.environ.get("API_GIT_REPO", "https://forgejo.media-on.de/Leopoldadmin/leocrm.git") API_GIT_BRANCH = os.environ.get("API_GIT_BRANCH", "main") # API ENV variables API_ENVS = [ {"key": "DATABASE_URL", "value": "postgresql+asyncpg://crm_api:4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI@crm-postgres:5432/crm_db"}, {"key": "AUTH_DATABASE_URL", "value": "postgresql+asyncpg://crm_auth:4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI@crm-postgres:5432/crm_db"}, {"key": "WORKER_DATABASE_URL", "value": "postgresql+asyncpg://crm_worker:4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI@crm-postgres:5432/crm_db"}, {"key": "MIGRATION_DATABASE_URL", "value": "postgresql+asyncpg://crm_user:4B6X2wlfbIx-PyaG8kGutsatdLbjdBUI@crm-postgres:5432/crm_db"}, {"key": "REDIS_URL", "value": "redis://default:lAjCaTf3XFP5XSaPJ1HElgLAJhQQswLT@crm-redis:6379/0"}, {"key": "SECRET_KEY", "value": "vVdAnvyc-ob4myE5D1rAYn-SovzoBfQLP1z4wmWteTmFPV_lveCGIn2upNoiP590"}, {"key": "ENVIRONMENT", "value": "production"}, {"key": "STORAGE_PATH", "value": "/data/storage"}, {"key": "FRONTEND_URL", "value": "https://crm.media-on.de"}, {"key": "CORS_ORIGINS", "value": "https://crm.media-on.de"}, {"key": "SESSION_COOKIE_SECURE", "value": "true"}, {"key": "LOG_LEVEL", "value": "INFO"}, ] def create_service(client: CoolifyClient, project_uuid: str, environment_name: str, server_uuid: str, compose_raw: str, name: str) -> str | None: """Create a Coolify service from a docker-compose definition. Returns the service UUID or None on failure.""" import base64 encoded = base64.b64encode(compose_raw.encode()).decode() try: resp = httpx.post( f"{client.base_url}/api/v1/services", headers=client.headers, json={ "project_uuid": project_uuid, "environment_name": environment_name, "server_uuid": server_uuid, "docker_compose_raw": encoded, "name": name, }, timeout=30, ) if resp.status_code in (200, 201): data = resp.json() return data.get("uuid") print(f" Error creating service {name}: {resp.status_code} {resp.text[:200]}") return None except Exception as e: print(f" Error creating service {name}: {e}") return None def set_service_envs(client: CoolifyClient, service_uuid: str, envs: list[dict]) -> None: """Set ENV variables for a service via Coolify API.""" for env in envs: resp = httpx.post( f"{client.base_url}/api/v1/services/{service_uuid}/envs", headers=client.headers, json=env, timeout=30, ) if resp.status_code == 409: resp = httpx.patch( f"{client.base_url}/api/v1/services/{service_uuid}/envs", headers=client.headers, json=env, timeout=30, ) if resp.status_code not in (200, 201): print(f" Warning: could not set ENV {env['key']} ({resp.status_code})") def create_api_application(client: CoolifyClient, project_uuid: str, environment_name: str, server_uuid: str) -> str | None: """Create the API application from a Git repository via private deploy key. Uses POST /applications/private-deploy-key with the Coolify host's private key. This creates a Git-based application that can auto-update on git push. Returns the application UUID or None on failure. """ # The private key UUID for the Coolify host (localhost) PRIVATE_KEY_UUID = os.environ.get( "COOLIFY_PRIVATE_KEY_UUID", "rgcsc0048c04csckk8kogk40", ) try: resp = httpx.post( f"{client.base_url}/api/v1/applications/private-deploy-key", headers=client.headers, json={ "project_uuid": project_uuid, "environment_name": environment_name, "server_uuid": server_uuid, "private_key_uuid": PRIVATE_KEY_UUID, "git_repository": API_GIT_REPO, "git_branch": API_GIT_BRANCH, "build_pack": "dockerfile", "name": "leocrm-api", "ports_exposes": "8000", }, timeout=30, ) if resp.status_code in (200, 201): data = resp.json() return data.get("uuid") print(f" Error creating API application: {resp.status_code} {resp.text[:300]}") return None except Exception as e: print(f" Error creating API application: {e}") return None def deploy_initial() -> int: """Initial deployment: create all Coolify resources from scratch. Creates: 1. PostgreSQL service (pgvector/pgvector:pg16) 2. Redis service (redis:7-alpine) 3. API application (from Git repo) 4. Worker service (same image as API) 5. Sets all ENV variables 6. Deploys API (builds image + runs migrations) 7. Deploys Worker """ print(f"\n{'='*60}") print(" LeoCRM Initial Deployment") print(f"{'='*60}\n") if not COOLIFY_TOKEN: print("ERROR: COOLIFY_API_TOKEN not set") return 2 client = CoolifyClient(COOLIFY_BASE_URL, COOLIFY_TOKEN) steps: list[tuple[str, StepResult]] = [] # Get project, environment, server PROJECT_UUID = os.environ.get("COOLIFY_PROJECT_UUID", "damkcrjuy4cjofo954ahmooe") ENVIRONMENT_NAME = os.environ.get("COOLIFY_ENVIRONMENT", "production") SERVER_UUID = os.environ.get("COOLIFY_SERVER_UUID", "lw80w8scs444gwcw084s00s4") # Step 1: Create PostgreSQL service print("\n[1/7] Creating PostgreSQL service...") pg_uuid = create_service(client, PROJECT_UUID, ENVIRONMENT_NAME, SERVER_UUID, POSTGRES_COMPOSE, "crm-postgres") if pg_uuid: print(f" PostgreSQL service created: {pg_uuid[:12]}") set_service_envs(client, pg_uuid, POSTGRES_ENVS) # Deploy to start the container client.deploy_application(pg_uuid) time.sleep(10) steps.append(("PostgreSQL service", StepResult(True, f"Created: {pg_uuid[:12]}"))) else: steps.append(("PostgreSQL service", StepResult(False, "Failed to create"))) print_summary(steps) return 1 _print_result(steps[-1][1]) # Step 2: Create Redis service print("\n[2/7] Creating Redis service...") redis_uuid = create_service(client, PROJECT_UUID, ENVIRONMENT_NAME, SERVER_UUID, REDIS_COMPOSE, "crm-redis") if redis_uuid: print(f" Redis service created: {redis_uuid[:12]}") set_service_envs(client, redis_uuid, REDIS_ENVS) client.deploy_application(redis_uuid) time.sleep(10) steps.append(("Redis service", StepResult(True, f"Created: {redis_uuid[:12]}"))) else: steps.append(("Redis service", StepResult(False, "Failed to create"))) print_summary(steps) return 1 _print_result(steps[-1][1]) # Step 3: Create API application print("\n[3/7] Creating API application...") api_uuid = create_api_application(client, PROJECT_UUID, ENVIRONMENT_NAME, SERVER_UUID) if api_uuid: print(f" API application created: {api_uuid[:12]}") # Set API ENV variables for env in API_ENVS: resp = httpx.post( f"{client.base_url}/api/v1/applications/{api_uuid}/envs", headers=client.headers, json=env, timeout=30, ) if resp.status_code == 409: resp = httpx.patch( f"{client.base_url}/api/v1/applications/{api_uuid}/envs", headers=client.headers, json=env, timeout=30, ) steps.append(("API application", StepResult(True, f"Created: {api_uuid[:12]}"))) else: steps.append(("API application", StepResult(False, "Failed to create"))) print_summary(steps) return 1 _print_result(steps[-1][1]) # Step 4: Create Worker service (with API image reference) print("\n[4/7] Creating Worker service...") # Generate worker compose with the API application's UUID as image name worker_compose = WORKER_COMPOSE_YAML.replace( "stvabl4vaqru7jclx4ittzr3:latest", f"{api_uuid}:latest", ) worker_uuid = create_service(client, PROJECT_UUID, ENVIRONMENT_NAME, SERVER_UUID, worker_compose, "leocrm-worker") if worker_uuid: print(f" Worker service created: {worker_uuid[:12]}") # Set connect_to_docker_network httpx.patch( f"{client.base_url}/api/v1/services/{worker_uuid}", headers=client.headers, json={"connect_to_docker_network": True}, timeout=30, ) set_service_envs(client, worker_uuid, WORKER_ENVS) steps.append(("Worker service", StepResult(True, f"Created: {worker_uuid[:12]}"))) else: steps.append(("Worker service", StepResult(False, "Failed to create"))) print_summary(steps) return 1 _print_result(steps[-1][1]) # Step 5: Deploy API (builds image + runs migrations via prestart.sh) print("\n[5/7] Deploying API (build + migrations)...") # Use the newly created API application UUID for deployment try: result = client.deploy_application(api_uuid) deploy_uuid = _extract_deploy_uuid(result) if deploy_uuid: print(f" Deploy queued: {deploy_uuid[:12]}") r = _wait_deployment(client, deploy_uuid, timeout=300) else: r = StepResult(True, "Deploy triggered (no UUID)") except Exception as e: r = StepResult(False, f"Deploy failed: {e}") steps.append(("API deploy", r)) _print_result(r) if not r.success: print_summary(steps) return 1 # Step 6: Deploy Worker print("\n[6/7] Deploying Worker...") # Tag the latest API image as :latest (using the new API UUID) tag_code, tag_output = ssh_run( f'docker images --format "{{{{.Repository}}}}:{{{{.Tag}}}}" | ' f'grep "^{api_uuid}:" | grep -v latest | head -1 | ' f'xargs -I{{}} docker tag {{}} {api_uuid}:latest' ) # Deploy worker result = client.deploy_application(worker_uuid) deploy_uuid = _extract_deploy_uuid(result) if deploy_uuid: dep_result = _wait_deployment(client, deploy_uuid, timeout=120) steps.append(("Worker deploy", dep_result)) else: wr = _wait_service_healthy(client, worker_uuid, timeout=120) steps.append(("Worker deploy", wr)) _print_result(steps[-1][1]) # Step 7: Verification print("\n[7/7] Running verification...") verify_results = run_verification(client, check_worker=True) steps.extend(verify_results) all_ok = print_summary(steps) return 0 if all_ok else 1 def main() -> None: parser = argparse.ArgumentParser( description="LeoCRM automated deployment script (Coolify API only)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python scripts/deploy.py # Full deploy (API + Worker) python scripts/deploy.py --skip-build # Skip build, just restart python scripts/deploy.py --worker-only # Only deploy worker python scripts/deploy.py --verify-only # Only run verification """, ) parser.add_argument( "--skip-build", action="store_true", help="Skip build, just restart services via Coolify API", ) parser.add_argument( "--worker-only", action="store_true", help="Only deploy the worker service", ) parser.add_argument( "--initial", action="store_true", help="Initial deployment: create all Coolify resources from scratch", ) parser.add_argument( "--verify-only", action="store_true", help="Only run verification checks (no deployment)", ) args = parser.parse_args() if args.initial: sys.exit(deploy_initial()) elif args.verify_only: sys.exit(verify_only()) elif args.worker_only: sys.exit(deploy_worker_only(skip_build=args.skip_build)) else: sys.exit(deploy_full(skip_build=args.skip_build)) if __name__ == "__main__": main()