INNER CODE UNIT · Python

main

theonedev/onedev · server-product/helm/files/stop-servers.py:87

def main():
    credentials = Path("/var/run/secrets/kubernetes.io/serviceaccount")
    context = ssl.create_default_context(cafile=str(credentials / "ca.crt"))
    host = os.environ["KUBERNETES_SERVICE_HOST"]
    if ":" in host:
        host = f"[{host}]"
    endpoint = f"https://{host}:{os.environ['KUBERNETES_SERVICE_PORT']}"

    def api(method, path, body=None):
        request = urllib.request.Request(endpoint + path, method=method,
            data=json.dumps(body).encode() if body is not None else None,
            headers={"Authorization": "Bearer " + (credentials / "token").read_text().strip(),
                     "Content-Type": "application/merge-patch+json" if method == "PATCH" else "application/json"})
        with urllib.request.urlopen(request, context=context, timeout=15) as response:
            return json.load(response)

    stop_servers(api, os.environ["POD_NAMESPACE"], os.environ["STATEFULSET_NAME"],
                 os.environ["TARGET_IMAGE"], int(os.environ["STOP_TIMEOUT"]))

View source record →