geronimo.deploy_cloud

 15class GeronimoCloudClient:
 16    """Client for interacting with Geronimo Cloud API."""
 17
 18    DEFAULT_API_URL = "https://api.geronimo.dev/v1"
 19    """Default base URL for the API."""
 20
 21    api_url: str
 22    """The API URL for Geronimo Cloud."""
 23
 24    token: Optional[str]
 25    """The authentication token."""
 26
 27    headers: Dict[str, str]
 28    """HTTP headers for API requests."""
 29
 30    def __init__(self, api_url: Optional[str] = None, token: Optional[str] = None):
 31        """Initialize the cloud client.
 32        
 33        Args:
 34            api_url: Optional API URL override.
 35            token: Optional auth token. If not provided, tries to load from credentials file.
 36        """
 37        self.api_url = api_url or os.getenv("GERONIMO_API_URL", self.DEFAULT_API_URL)
 38        self.token = token or self._load_token()
 39        
 40        self.headers = {
 41            "Authorization": f"Bearer {self.token}" if self.token else "",
 42            "User-Agent": f"geronimo-cli/{__version__}",
 43            "Content-Type": "application/json",
 44        }
 45        
 46    def _load_token(self) -> Optional[str]:
 47        """Load token from credentials file."""
 48        creds_path = Path.home() / ".geronimo" / "credentials"
 49        if creds_path.exists():
 50            try:
 51                data = json.loads(creds_path.read_text())
 52                return data.get("token")
 53            except Exception:
 54                return None
 55        return None
 56
 57    def login(self, token: str) -> Dict[str, Any]:
 58        """Verify token and save credentials."""
 59        # Validate token with API
 60        with api_client(self.api_url, {"Authorization": f"Bearer {token}"}, operation="login") as client:
 61            response = client.get("/auth/verify")
 62            response.raise_for_status()
 63            user_data = response.json()
 64            
 65        # Save to file
 66        creds_dir = Path.home() / ".geronimo"
 67        creds_dir.mkdir(parents=True, exist_ok=True)
 68        (creds_dir / "credentials").write_text(json.dumps({"token": token}))
 69        
 70        # Update current instance
 71        self.token = token
 72        self.headers["Authorization"] = f"Bearer {token}"
 73        
 74        return user_data
 75
 76    def deploy_project(self, project_name: str, config: Dict[str, Any], zip_path: Path) -> Dict[str, Any]:
 77        """Deploy a project to the cloud.
 78        
 79        Args:
 80            project_name: Name of the project.
 81            config: Full deployment configuration.
 82            zip_path: Path to the zipped project artifacts.
 83        """
 84        if not self.token:
 85            raise RuntimeError("Not authenticated. Run 'geronimo auth login' first.")
 86
 87        # 1. Create deployment record
 88        with api_client(self.api_url, self.headers, operation="create_deployment") as client:
 89            resp = client.post("/deployments", json={
 90                "project": project_name,
 91                "config": config
 92            })
 93            resp.raise_for_status()
 94            deployment = resp.json()
 95            upload_url = deployment["upload_url"]
 96            deployment_id = deployment["id"]
 97
 98        # 2. Upload artifacts
 99        with open(zip_path, "rb") as f:
100            with transfer_client(operation="upload_deployment") as client:
101                client.put(upload_url, content=f)
102            
103        # 3. Trigger build/deploy
104        with api_client(self.api_url, self.headers, operation="start_deployment") as client:
105            resp = client.post(f"/deployments/{deployment_id}/start")
106            resp.raise_for_status()
107            return resp.json()
108
109    def get_status(self, deployment_id: str) -> Dict[str, Any]:
110        """Get deployment status."""
111        with api_client(self.api_url, self.headers, operation="get_status") as client:
112            resp = client.get(f"/deployments/{deployment_id}")
113            resp.raise_for_status()
114            return resp.json()
115
116    def get_logs(self, deployment_id: str) -> str:
117        """Get build/runtime logs."""
118        with api_client(self.api_url, self.headers, operation="get_logs") as client:
119            resp = client.get(f"/deployments/{deployment_id}/logs")
120            resp.raise_for_status()
121            return resp.text
122
123    def teardown(self, deployment_id: str) -> Dict[str, Any]:
124        """Teardown a deployment."""
125        with api_client(self.api_url, self.headers, operation="teardown") as client:
126            resp = client.delete(f"/deployments/{deployment_id}")
127            resp.raise_for_status()
128            return resp.json()
129
130    def sync_keys(self, keys: list[Dict[str, Any]]) -> Dict[str, Any]:
131        """Sync local API keys to Geronimo Cloud.
132        
133        Args:
134            keys: List of key dictionaries from APIKey.to_dict().
135            
136        Returns:
137            Response with synced/skipped counts.
138            
139        Raises:
140            RuntimeError: If not authenticated.
141            httpx.HTTPStatusError: If API request fails.
142        """
143        if not self.token:
144            raise RuntimeError("Not authenticated. Run 'geronimo auth login' first.")
145        
146        with api_client(self.api_url, self.headers, operation="sync_keys") as client:
147            resp = client.post("/inference-keys/sync", json={"keys": keys})
148            resp.raise_for_status()
149            return resp.json()

14class GeronimoCloudTarget:
15    """Deployment target for Geronimo Cloud."""
16
17    config: DeploymentConfig
18    """The deployment configuration."""
19
20    client: GeronimoCloudClient
21    """Client for communicating with the cloud API."""
22
23    def __init__(self, config: DeploymentConfig):
24        self.config = config
25        self.client = GeronimoCloudClient()
26
27    def deploy(self, component: Optional[str] = None) -> Dict[str, Any]:
28        """Deploy to Geronimo Cloud.
29        
30        Args:
31            component: Optional component filter (ignored for now, full project deployed).
32        
33        Returns:
34            Deployment result with status and URLs.
35        """
36        print(f"Deploying project '{self.config.project}' to Geronimo Cloud...")
37
38        # 1. Package Project
39        with tempfile.TemporaryDirectory() as temp_dir:
40            archive_path = shutil.make_archive(
41                base_name=f"{temp_dir}/project",
42                format="zip",
43                root_dir=".",  # Assuming running from project root
44                base_dir="."   
45            )
46            
47            # 2. Upload and Deploy
48            try:
49                result = self.client.deploy_project(
50                    project_name=self.config.project,
51                    config=self.config.model_dump(),  # Serialize config
52                    zip_path=Path(archive_path)
53                )
54                
55                deployment_id = result["id"]
56                print(f"Deployment started: {deployment_id}")
57                
58                # 3. Poll for completion (simple implementation)
59                return self._wait_for_deployment(deployment_id)
60                
61            except Exception as e:
62                raise RuntimeError(f"Cloud deployment failed: {e}")
63
64    def _wait_for_deployment(self, deployment_id: str) -> Dict[str, Any]:
65        """Wait for deployment to complete."""
66        print("Waiting for deployment to complete...")
67        while True:
68            status = self.client.get_status(deployment_id)
69            state = status.get("status")
70            
71            if state == "active":
72                return status
73            elif state in ("failed", "error"):
74                raise RuntimeError(f"Deployment failed: {status.get('error')}")
75            
76            time.sleep(5)
77
78    def destroy(self) -> Dict[str, Any]:
79        """Destroy cloud resources."""
80        # In a real scenario, we'd need to know which deployment ID to destroy.
81        # For now, we might need to look it up or assume the user provides it.
82        # This is a simplification.
83        print(f"Teardown requested for project '{self.config.project}'")
84        # TODO: Implement lookup of active deployment for this project/stack
85        raise NotImplementedError("Teardown not yet fully implemented for cloud target without ID lookup.")
86
87    def logs(self, deployment_id: str) -> str:
88        """Get logs."""
89        return self.client.get_logs(deployment_id)