Usługi HTTP

Użyj @app.inference(), aby uruchomić usługę przyjmującą żądania HTTP pod adresem udostępnionym przez CGC. SDK tworzy zasób, w razie potrzeby czeka na jego uruchomienie, konfiguruje porty i zwraca obiekt DeployedEndpoint.

Poniższy przykład uruchamia prostą usługę do analizy tekstu w kontenerze z Pythonem.

Zdefiniuj usługę

from cgc.sdk import App, Image, ShellCommand


RESOURCE_NAME = "text-analysis-service"
PORT = 8000
PORT_NAME = "api"

app = App(
    name="http-service-workflow",
    image=Image.from_registry("python:3.11-slim"),
)


@app.inference(
    entity="custom",
    name=RESOURCE_NAME,
    port=PORT,
    port_name=PORT_NAME,
    cpu=1,
    memory=2,
)
def text_analysis_service():
    code = r"""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json


class Handler(BaseHTTPRequestHandler):
    def _send_json(self, status_code, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status_code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        if self.path in {"/health", "/api/health"}:
            self._send_json(200, {"status": "ok"})
            return
        self._send_json(404, {"error": "not found", "path": self.path})

    def do_POST(self):
        if self.path not in {"/analyze", "/api/analyze"}:
            self._send_json(404, {"error": "not found", "path": self.path})
            return

        length = int(self.headers.get("Content-Length", "0"))
        payload = json.loads(self.rfile.read(length) or b"{}")
        text = payload.get("text", "")
        words = text.split()
        self._send_json(
            200,
            {
                "input": text,
                "characters": len(text),
                "words": len(words),
                "uppercase": text.upper(),
                "first_five_words": words[:5],
            },
        )

    def log_message(self, format, *args):
        print(format % args, flush=True)


ThreadingHTTPServer(("0.0.0.0", 8000), Handler).serve_forever()
"""
    return ShellCommand(executable=("python", "-c", code))

entity="custom" oznacza, że CGC uruchamia obraz wybrany w App. Dekorowana funkcja wykonuje się lokalnie i zwraca polecenie, które CGC uruchamia w kontenerze. ShellCommand przechowuje program i jego argumenty oraz zabezpiecza je przed niezamierzoną interpretacją przez powłokę.

Serwer musi nasłuchiwać na 0.0.0.0 oraz na tym samym porcie, który podano w port. W tym przykładzie jest to port 8000.

Zainstaluj zależności przy uruchamianiu

Jeśli potrzebujesz dodatkowych pakietów, zainstaluj je w ShellCommand.prelude przed uruchomieniem głównego procesu:

return ShellCommand(
    executable=("python", "-m", "my_service"),
    arguments=("--host 0.0.0.0", "--port 8000"),
    prelude="python -m pip install --quiet fastapi 'uvicorn[standard]'",
)

Polecenia z prelude wykonują się przy każdym uruchomieniu kontenera. Zainstalowane pakiety nie są zapisywane w obrazie. Stałe zależności najlepiej umieścić we własnym obrazie, a prelude wykorzystywać do instalacji dodatkowych pakietów podczas pracy nad aplikacją.

Uruchom usługę i wyślij żądanie

endpoint = text_analysis_service.deploy(
    wait=True,
    reuse_existing=False,
    resource_timeout_seconds=600,
)

print("Endpoint:", endpoint.url)
endpoint.wait_http("/health", timeout_seconds=300)

result = endpoint.post_json(
    "/analyze",
    {"text": "CGC SDK workflows can run HTTP services."},
)
print(result)

deploy() najpierw tworzy zasób, a następnie konfiguruje zadeklarowany port. wait=True czeka na gotowość zasobu i odświeża jego adres publiczny. Przy reuse_existing=False istniejący zasób o tej samej nazwie powoduje błąd. Wartość True pozwala go ponownie użyć, ale nie aktualizuje jego obrazu, polecenia ani środowiska.

wait_http() czeka, aż sama aplikacja zacznie przyjmować żądania. Jest to coś innego niż wait=True: zasób może mieć stan Running, zanim serwer HTTP będzie gotowy.

URL endpointu może już zawierać prefiks ścieżki ingress wynikający z port_name. Metody wait_http() i post_json() dodają podaną ścieżkę do tego URL-a. Aplikacja musi udostępniać trasy zgodne z wynikową ścieżką. Pełny przykład przyjmuje trasy bez prefiksu, takie jak /health, oraz trasy z prefiksem, takie jak /api/health.

Argumenty entrypointu własnego obrazu

Zamiast polecenia powłoki lub obiektu ShellCommand funkcja z entity="custom" może zwracać listę argumentów:

from cgc.sdk import App, Image

args_app = App("argument-demo", image=Image.from_registry("python:3.12-slim"))

@args_app.inference(name="argument-demo", port=8000, port_name="http")
def serve_with_arguments():
    return ["python", "-m", "http.server", "8000"]

plan = serve_with_arguments.deploy(dry_run=True)

Listę przetwarza backend. Ograniczenia dotyczące spacji i powtarzających się kluczy opisano w opisie API. Dla złożonych poleceń użyj ShellCommand.

Dodaj więcej portów

Użyj ServicePort, kiedy jeden zasób nasłuchuje na więcej niż jednym porcie:

from cgc.sdk import App, Image, ServicePort, ShellCommand

multi_app = App("multi-port-demo", image=Image.from_registry("python:3.12-slim"))

@multi_app.inference(
    name="multi-port-service", port=8000, port_name="http",
    additional_ports=[
        ServicePort("metrics", 9090, ingress=False),
        ServicePort("internal", 5557, ingress=False),
    ],
)
def service():
    code = r"""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = json.dumps({"port": self.server.server_port, "status": "ok"}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

for port in (8000, 9090, 5557):
    server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
threading.Event().wait()
"""
    return ShellCommand(executable=("python", "-u", "-c", code))

plan = service.deploy(dry_run=True)

SDK najpierw tworzy zasób, czeka na stan Running, a następnie dodaje port główny i porty dodatkowe. Nazwy i numery portów muszą być unikalne. Każdy port musi mieścić się w zakresie od 1 do 65535. Ustawienie ingress=False pozostawia port bez publicznego adresu. Samo dodanie portu nie wystarcza; aplikacja działająca w kontenerze musi na nim nasłuchiwać.

Jeśli konfiguracja portów zakończy się błędem, SDK zgłasza wyjątek i pozostawia zasób, aby można było go sprawdzić lub samodzielnie usunąć.

Przy dodatkowych portach SDK czeka na stan Running nawet wtedy, gdy ustawisz wait=False. Jeśli główny port ma ingress=False, nie używaj endpoint.url, wait_http() ani post_json(). Porty wewnętrzne są dostępne z zadań w tej samej przestrzeni nazw Kubernetes.

Użyj szablonu zasobu CGC

Dla szablonu pomiń image i startup_command. Poniższa deklaracja używa ustawień szablonu Nginx:

from cgc.sdk import App

template_app = App("template-service")
resource = template_app.resource(
    entity="nginx", name="template-nginx", cpu=1, memory=1,
)
plan = resource.create(dry_run=True)

Szablon wskazany w entity musi być dostępny w wybranym środowisku. Jego argumenty przekazuj przez App.resource(arguments=[...]) albo zwracaj jako listę w funkcji App.inference(entity=...).

Sprawdź status, odczytaj logi i usuń usługę

print(endpoint.status())
print(endpoint.logs(timestamps=True))
endpoint.delete()

Uruchom pełny przykład z usunięciem usługi po zakończeniu:

python text_inference.py --delete

Pełny skrypt znajdziesz na stronie Przykłady workflow. Aby skorzystać z gotowych API modeli, przejdź do gotowych workflow.