File size: 3,160 Bytes
1e23d14 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | import logging
from typing import Any, Optional
logger = logging.getLogger(__name__)
try:
from googleapiclient import discovery
from google.oauth2 import service_account
except ImportError:
discovery = None # type: ignore
service_account = None # type: ignore
class CloudRunClient:
SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
def __init__(self, project_id: str, region: str = "us-central1",
credentials_file: Optional[str] = None) -> None:
self.project_id = project_id
self.region = region
if discovery is None:
raise ImportError("google-api-python-client is not installed")
if credentials_file and service_account:
creds = service_account.Credentials.from_service_account_file(
credentials_file, scopes=self.SCOPES
)
else:
import google.auth
creds, _ = google.auth.default(scopes=self.SCOPES)
self._svc = discovery.build("run", "v1", credentials=creds)
self._parent = f"namespaces/{project_id}"
logger.debug("Cloud Run client for %s/%s", project_id, region)
def list_services(self) -> list[dict[str, Any]]:
response = (
self._svc.namespaces()
.services()
.list(parent=self._parent)
.execute()
)
return [
{
"name": svc["metadata"]["name"],
"url": svc.get("status", {}).get("url", ""),
"ready": any(
c.get("type") == "Ready" and c.get("status") == "True"
for c in svc.get("status", {}).get("conditions", [])
),
}
for svc in response.get("items", [])
]
def get_service(self, service_name: str) -> dict[str, Any]:
name = f"{self._parent}/services/{service_name}"
svc = self._svc.namespaces().services().get(name=name).execute()
return {
"name": svc["metadata"]["name"],
"url": svc.get("status", {}).get("url"),
"image": (svc.get("spec", {})
.get("template", {})
.get("spec", {})
.get("containers", [{}])[0]
.get("image")),
"annotations": svc.get("metadata", {}).get("annotations", {}),
}
def update_traffic(self, service_name: str,
traffic: list[dict[str, Any]]) -> None:
name = f"{self._parent}/services/{service_name}"
svc = self._svc.namespaces().services().get(name=name).execute()
svc["spec"]["traffic"] = traffic
self._svc.namespaces().services().replaceService(
name=name, body=svc
).execute()
logger.info("Updated traffic for service %s", service_name)
def delete_service(self, service_name: str) -> None:
name = f"{self._parent}/services/{service_name}"
self._svc.namespaces().services().delete(name=name).execute()
logger.info("Deleted Cloud Run service %s", service_name)
|