from __future__ import annotations from uuid import uuid4 from sqlalchemy.orm import Session from app.core.time import format_now from app.models.app_metadata import AppMetadata from app.repositories.metadata_repository import AppMetadataRepository class MetadataService: DEMO_ITEMS = [ { "app_code": "order-service", "env": "test", "process_name": "java", "command_contains": "order-service", "health_check_url": "http://127.0.0.1:18080/actuator/health", "log_path": "sample-apps/order-service/runtime/logs/order-service.log", "listen_port": 18080, "startup_keyword": "Started order-service", }, { "app_code": "order-service", "env": "prod", "process_name": "java", "command_contains": "order-service", "health_check_url": "http://order-service.prod.demo/actuator/health", "log_path": "logs/order-service.log", "listen_port": 8080, "startup_keyword": "Started order-service", }, { "app_code": "payment-service", "env": "test", "process_name": "java", "command_contains": "payment-service", "health_check_url": "http://payment-service.test.demo/actuator/health", "log_path": "logs/payment-service.log", "listen_port": 8081, "startup_keyword": "Started payment-service", }, ] def __init__(self, db: Session, timezone_name: str) -> None: self.db = db self.timezone_name = timezone_name self.repository = AppMetadataRepository(db) def ensure_demo_metadata(self) -> None: for item in self.DEMO_ITEMS: existing = self.repository.get_by_app_env(item["app_code"], item["env"]) if existing: continue current_time = format_now(self.timezone_name) self.repository.add( AppMetadata( app_metadata_id=f"app-meta-{uuid4().hex[:12]}", app_code=item["app_code"], env=item["env"], process_name=item["process_name"], command_contains=item["command_contains"], health_check_url=item["health_check_url"], log_path=item["log_path"], listen_port=item["listen_port"], startup_keyword=item["startup_keyword"], created_at=current_time, updated_at=current_time, ) ) def get_app_metadata(self, app_code: str | None, env: str | None) -> AppMetadata | None: if not app_code or not env: return None return self.repository.get_by_app_env(app_code, env)