Automatisierung von Aufgaben mit der API

Während die Codesphere CLI ideal für Terminal-Workflows und die Nutzung in CI/CD-Pipelines mit vordefinierten Anwendungsfällen ist, ermöglicht die Public API das Schreiben eigener Skripte, um Pipelines zu automatisieren, Infrastruktur-Lebenszyklen zu verwalten und Codesphere in interne Tools zu integrieren.

Dieser Leitfaden behandelt die Grundlagen der Skripterstellung und liefert praktische Automatisierungsbeispiele.

Erste Schritte​

Um programmatisch mit der Codesphere API zu interagieren, benötigst du zwei Dinge: einen HTTP-Client und eine sichere Methode, um dein Authentifizierungs-Token zu übergeben.

Authentifizierungs-Header​

Jede API-Anfrage benötigt einen Authorization-Header, der dein API-Token im Bearer-Token-Format enthält.

Grundlegendes Setup eines HTTP-Clients​

Hier ist ein Beispiel, wie du eine grundlegende authentifizierte Anfrage einrichtest, um die Workspaces deines Teams in gängigen Programmiersprachen abzurufen. Idealerweise testest du einzelne Anfragen zunächst mit Tools wie Postman, bevor du diese Skripte schreibst.

import os
import requests

# Load credentials from environment variables
API_TOKEN = os.environ.get("CODESPHERE_TOKEN")
TEAM_ID = os.environ.get("CODESPHERE_TEAM_ID")
BASE_URL = "https://cloud.codesphere.com/api"

headers = {
    "Authorization": f"Bearer {API_TOKEN}",
    "Content-Type": "application/json"
}

# Example: Fetch all workspaces
response = requests.get(f"{BASE_URL}/workspaces?teamId={TEAM_ID}", headers=headers)

if response.status_code == 200:
    print("Workspaces fetched successfully!")
    print(response.json())
else:
    print(f"Error: {response.status_code} - {response.text}")

// Load credentials from environment variables
const API_TOKEN = process.env.CODESPHERE_TOKEN;
const TEAM_ID = process.env.CODESPHERE_TEAM_ID;
const BASE_URL = "https://cloud.codesphere.com/api";

const headers = {
    "Authorization": `Bearer ${API_TOKEN}`,
    "Content-Type": "application/json"
};

// Example: Fetch all workspaces
async function getWorkspaces() {
    const response = await fetch(`${BASE_URL}/workspaces?teamId=${TEAM_ID}`, {
        method: 'GET',
        headers: headers
    });

    if (response.ok) {
        const data = await response.json();
        console.log("Workspaces fetched successfully!", data);
    } else {
        console.error(`Error: ${response.status} - ${await response.text()}`);
    }
}

getWorkspaces();

Automatisierungsbeispiele​

Im Folgenden findest du praxisnahe Beispiele, wie du die API nutzen kannst, um mühsame Aufgaben im Lebenszyklus-Management zu automatisieren.

Integration des Deployment-Status in interne Tools​

Wenn dein Unternehmen ein eigenes internes Developer-Portal (wie Backstage) oder einen eigenen Slack-Bot nutzt, möchtest du die API vielleicht regelmäßig abfragen, um den Status und Pipeline-Zustand eines bestimmten Deployments zu erfassen.

Dieses Skript zeigt, wie du den Status der Pipeline eines bestimmten Workspaces abrufst.

import os
import requests

API_TOKEN = os.environ.get("CODESPHERE_TOKEN")
WORKSPACE_ID = "YOUR_WORKSPACE_ID"
BASE_URL = "https://cloud.codesphere.com/api"
HEADERS = {"Authorization": f"Bearer {API_TOKEN}"}

def check_deployment_status():
    # Fetch the status of the run pipeline
    url = f"{BASE_URL}/workspaces/{WORKSPACE_ID}/pipeline/run"
    response = requests.get(url, headers=HEADERS)
    
    if response.status_code == 200:
        status_data = response.json()
        
        # Extract relevant status information
        state = status_data.get("status", "UNKNOWN")
        is_running = state == "RUNNING"
        
        print(f"Deployment Status: {state}")
        
        # Example logic for internal tool integration
        if not is_running:
            print("Alert: The application is currently down or stopped!")
            # trigger_slack_alert(f"Workspace {WORKSPACE_ID} is down!")
        return state
    else:
        print(f"Failed to fetch status: {response.text}")
        return None

if __name__ == "__main__":
    check_deployment_status()

const API_TOKEN = process.env.CODESPHERE_TOKEN;
const WORKSPACE_ID = "YOUR_WORKSPACE_ID";
const BASE_URL = "https://cloud.codesphere.com/api";
const HEADERS = { "Authorization": `Bearer ${API_TOKEN}` };

async function checkDeploymentStatus() {
    // Fetch the status of the run pipeline
    const url = `${BASE_URL}/workspaces/${WORKSPACE_ID}/pipeline/run`;
    const response = await fetch(url, { headers: HEADERS });
    
    if (response.ok) {
        const statusData = await response.json();
        
        // Extract relevant status information
        const state = statusData.status || "UNKNOWN";
        const isRunning = state === "RUNNING";
        
        console.log(`Deployment Status: ${state}`);
        
        // Example logic for internal tool integration
        if (!isRunning) {
            console.warn("Alert: The application is currently down or stopped!");
            // triggerSlackAlert(`Workspace ${WORKSPACE_ID} is down!`);
        }
        return state;
    } else {
        console.error(`Failed to fetch status: ${await response.text()}`);
        return null;
    }
}

checkDeploymentStatus();

Deployment- und Release-Workflows​

Während einzelne Endpunkte für sich genommen schon nützlich sind, entfaltet sich die eigentliche Stärke durch die Kombination mehrerer API-Aufrufe zu einem vollständigen Release-Flow (z. B. das Auslösen eines Releases bei einem Merge in einer GitHub Action).

Einfacher Release-Fall​

Der einfache Fall eignet sich hervorragend für statische Websites und Anwendungen, die nahezu sofort neu starten können.

1. Änderungen abrufen (Pull): POST /workspaces/{workspaceId}/git/pull/{remote}
2. Anwendung neu bauen (optional): POST /workspaces/{workspaceId}/pipeline/prepare/start
3. Auf den Build warten: GET /workspaces/{workspaceId}/pipeline/prepare abfragen, bis der Status 200 zurückgegeben wird.
4. Anwendung stoppen: POST /workspaces/{workspaceId}/pipeline/run/stop
5. Anwendung neu starten: POST /workspaces/{workspaceId}/pipeline/run/start

Automatisierung von Zero-Downtime-Releases​

Für unternehmenskritische Anwendungen, bei denen Ausfallzeiten nicht akzeptabel sind, kannst du ein Blue/Green-Zero-Downtime-Release vollständig über die API automatisieren:

1. Neuen Workspace erstellen: POST /workspaces (Definiere den neuen Git-Commit, die Team-ID, die Plan-ID, den Branch und die Anzahl der Replicas).
2. Abhängigkeiten bauen: POST /workspaces/{workspaceId}/pipeline/prepare/start
3. Auf den Build warten: GET /workspaces/{workspaceId}/pipeline/prepare abfragen, bis er erfolgreich ist.
4. Tests ausführen (optional): POST /workspaces/{workspaceId}/pipeline/test/start und abfragen, bis der Vorgang erfolgreich ist. Stoppe den Ablauf, falls Tests fehlschlagen!
5. Anwendung starten: POST /workspaces/{workspaceId}/pipeline/run/start
6. Domain-Routing umschalten: Sobald die neue Anwendung fehlerfrei läuft, leite den Traffic sofort dorthin um, indem du die Workspace-Verbindung aktualisierst: PUT /domains/team/{teamId}/domain/{domainName}/workspace-connections

Programmatische Skalierung von Replicas​

Codesphere Workspaces können mehrere Services gleichzeitig als Teil einer Landscape ausführen (z. B. ein Frontend, ein Backend und eine Datenbank).

Du kannst die Anzahl der horizontalen Replicas für bestimmte Services innerhalb deines Workspace dynamisch anpassen.

Über den Endpunkt PATCH /workspaces/{workspaceId}/landscape/scale kannst du ein JSON-Objekt übergeben, bei dem die Keys die exakten Namen deiner Services sind und die Werte die gewünschte Anzahl an Replicas angeben.

import os
import json
import urllib.request

API_TOKEN = os.environ.get("CODESPHERE_TOKEN")
WORKSPACE_ID = "YOUR_WORKSPACE_ID" 
BASE_URL = "YOUR_CODESPHERE_INSTANCE_URL/api"

url = f"{BASE_URL}/workspaces/{WORKSPACE_ID}/landscape/scale"

payload = json.dumps({
"YOUR-SERVICE-NAME": 1  # <--- CHANGE THIS TO YOUR SERVICE NAME & desired number of replicas (i.e. "backend")
}).encode('utf-8')

headers = {
"Authorization": f"Bearer {API_TOKEN}",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
}

req = urllib.request.Request(url, data=payload, headers=headers, method='PATCH')

try:
with urllib.request.urlopen(req) as response:
    if response.status in [200, 204]:
        print(f"Success! Services in Workspace {WORKSPACE_ID} have been scaled.")
except urllib.error.HTTPError as e:
print(f"Failed to scale. Error code: {e.code}")
print(e.read().decode())
except Exception as e:
print(f"An error occurred: {e}")

const API_TOKEN = process.env.CODESPHERE_TOKEN;
const WORKSPACE_ID = "YOUR_WORKSPACE_ID";
const BASE_URL = "https://cloud.codesphere.com/api";

const headers = {
"Authorization": `Bearer ${API_TOKEN}`,
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
};

async function scaleServices(servicesMap) {
// The landscape scaling endpoint
const url = `${BASE_URL}/workspaces/${WORKSPACE_ID}/landscape/scale`;

const response = await fetch(url, {
    method: 'PATCH',
    headers: headers,
    body: JSON.stringify(servicesMap)
});

if (response.ok) {
    console.log(`Successfully scaled services in workspace ${WORKSPACE_ID}!`);
} else {
    const errorText = await response.text();
    console.error(`Failed to scale. Status code: ${response.status}`);
    console.error(errorText);
}
}

// Scale frontend to 2 replicas
scaleServices({
"YOUR-SERVICE-NAME": 2 // <--- CHANGE THIS TO YOUR ACTUAL SERVICE NAME (e.g., "backend")
});