Verwalten von Snowpark Container Services (einschließlich Dienstfunktionen) mit Python¶

Erstellen von Computepools¶

Sie können einen Computepool erstellen, indem Sie die Methode ComputePoolCollection.create aufrufen und ein ComputePool-Objekt übergeben, das den zu erstellenden Computepool repräsentiert.

Um einen Computepool zu erstellen, erstellen Sie zunächst ein ComputePool-Objekt, das Pool-Eigenschaften wie die folgenden angibt:

• Name des Computepools

• Maximale und minimale Anzahl von Knoten, die der Pool enthalten soll

• Name der Instanz-Familie, die den Maschinentyp angibt, der für die Knoten im Pool bereitgestellt werden soll

• Ob der Pool automatisch fortgesetzt werden soll, wenn ein Dienst oder Job an ihn übermittelt wird

Der Code im folgenden Beispiel erstellt ein ComputePool-Objekt, das einen Pool namens my_compute_pool repräsentiert:

from snowflake.core.compute_pool import ComputePool

compute_pool = ComputePool(name="my_compute_pool", min_nodes=1, max_nodes=2, instance_family="CPU_X64_XS", auto_resume=False)
root.compute_pools.create(compute_pool)

Der Code erstellt dann den Computepool, indem das ComputePool-Objekt an die Methode ComputePoolCollection.create übergeben wird.

Abrufen von Details zum Computepool¶

Sie können Informationen zu einem Computepool erhalten, indem Sie die Methode ComputePoolResource.fetch aufrufen, die ein ComputePool-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft Informationen über einen Pool namens my_compute_pool ab:

compute_pool = root.compute_pools["my_compute_pool"].fetch()
print(compute_pool.to_dict())

Auflisten von Computepools¶

Zum Auflisten der Computepools können Sie die Methode iter verwenden, die einen PagedIter-Iterator zurückgibt.

Der Code im folgenden Beispiel listet Computepools auf, deren Name mit my beginnt:

compute_pools = root.compute_pools.iter(like="my%")
for compute_pool in compute_pools:
  print(compute_pool.name)

Ausführen von Computepool-Operationen¶

Mit einem ComputePoolResource-Objekt, das Sie mit der ComputePool.fetch-Methode abrufen, können Sie allgemeine Computepool-Operationen wie das Anhalten, Fortsetzen und Beenden von Pools ausführen.

Der Code im folgenden Beispiel hält den Computepool my_compute_pool an, setzt ihn fort und beendet ihn:

compute_pool_res = root.compute_pools["my_compute_pool"]
compute_pool_res.suspend()
compute_pool_res.resume()
compute_pool_res.stop_all_services()

Der Code verwendet die Methode Root.compute_pools, um ein Objekt ComputePool zu erstellen, das den Computepool repräsentiert. Über das ComputePool-Objekt wird ein ComputePoolResource-Objekt abgerufen, mit dem Computepool-Operationen ausgeführt werden können.

Erstellen von Image-Repositorys¶

Um ein Image-Repository zu erstellen, erstellen Sie zunächst ein ImageRepository-Objekt, das den Repository-Namen angibt.

Der Code im folgenden Beispiel erstellt ein ImageRepository-Objekt, das ein Repository namens my_repo repräsentiert:

from snowflake.core.image_repository import ImageRepository

my_repo = ImageRepository("my_repo")
root.databases["my_db"].schemas["my_schema"].image_repositories.create(my_repo)

Der Code erstellt dann das Image-Repository, indem das ImageRepository-Objekt an die Methode ImageRepositoryCollection.create übergeben und so das Image-Repository in der Datenbank my_db und dem Schema my_schema erstellt wird.

Abrufen von Details zu Image-Repositorys¶

Sie können Informationen zu einem Image-Repository abrufen, indem Sie die Methode ImageRepositoryResource.fetch aufrufen, die ein ImageRepository-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft ein ImageRepository-Objekt ab, das das Image-Repository my_repo repräsentiert, und gibt dann den Namen des Eigentümers des Repositorys aus.

my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo = my_repo_res.fetch()
print(my_repo.owner)

Auflisten von Image-Repositorys¶

Sie können die Image-Repositorys in einem bestimmten Schema mit der Methode iter auflisten, die einen Iterator PagedIter für ImageRepository-Objekte zurückgibt.

Der Code im folgenden Beispiel listet Repository-Namen in der Datenbank my_db und dem Schema my_schema auf:

repo_list = root.databases["my_db"].schemas["my_schema"].image_repositories.iter()
for repo_obj in repo_list:
  print(repo_obj.name)

Löschen eines Image-Repositorys¶

Sie können ein Image-Repository mit der Methode ImageRepositoryResource.drop löschen.

Der Code im folgenden Beispiel löscht das Repository my_repo:

my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo_res.drop()

Erstellen von Diensten¶

Um einen Dienst zu erstellen, führen Sie die Methode services.create aus und übergeben ein Service-Objekt, das den zu erstellenden Dienst repräsentiert.

Sie erstellen einen Dienst aus einer Dienstspezifikationsdatei .yaml, die in einen Stagingbereich hochgeladen wurde. Weitere Informationen zum Erstellen einer Dienstspezifikation finden Sie unter Referenz der Dienstspezifikation.

session.file.put("/local_location/my_service_spec.yaml", "@my_stage")

service_spec_string = """
// Specification as a string.
"""
session.file.put_stream(StringIO(sepc_in_string), "@my_stage/my_service_spec.yaml")

from snowflake.core.service import Service, ServiceSpec

my_service = Service(name="my_service", min_instances=1, max_instances=2, compute_pool="my_compute_pool", spec=ServiceSpec("@my_stage/my_service_spec.yaml"))
root.databases["my_db"].schemas["my_schema"].services.create(my_service)

from textwrap import dedent
from snowflake.core.service import Service, ServiceSpec

spec_text = dedent(f"""\
    spec:
      containers:
      - name: hello-world
        image: repo/hello-world:latest
      endpoints:
      - name: hello-world-endpoint
        port: 8080
        public: true
    """)

my_service = Service(name="my_service", min_instances=1, max_instances=2, compute_pool="my_compute_pool", spec=ServiceSpec(spec_text))
root.databases["my_db"].schemas["my_schema"].services.create(my_service)

from snowflake.core import CreateMode
from snowflake.core.function import FunctionArgument, ServiceFunction

root.databases["my_db"].schemas["my_schema"].functions.create(
  ServiceFunction(
    name="my-udf",
    arguments=[
        FunctionArgument(name="input", datatype="TEXT")
    ],
    returns="TEXT",
    service="hello-world",
    endpoint="'hello-world-endpoint'",
    path="/hello-world-path",
    max_batch_rows=5,
  ),
  mode = CreateMode.or_replace
)

result = root.databases["my_db"].schemas["my_schema"].functions["my-udf(TEXT)"].execute_function(["test"])
print(result)

Abrufen von Details zum Dienst¶

Sie können Informationen zu einem Snowflake-Dienst erhalten, indem Sie die Methode ServiceResource.fetch aufrufen, die ein Service-Objekt zurückgibt.

Der Code im folgenden Beispiel ruft Informationen über einen Dienst namens my_service ab:

my_service = root.databases["my_db"].schemas["my_schema"].services["my_service"].fetch()

Auflisten von Diensten¶

Sie können die Dienste in einem bestimmten Schema mit der Methode iter auflisten, die einen Iterator PagedIter für Service-Objekte zurückgibt.

Der Code im folgenden Beispiel listet Dienste auf, deren Name mit my beginnt:

services = root.databases["my_db"].schemas["my_schema"].services.iter(like="my%")
for service_obj in services:
  print(service_obj.name)

Ausführen von Dienstoperationen¶

Mit einem ServiceResource-Objekt können Sie allgemeine Dienstoperationen wie Anhalten, Fortsetzen und Abrufen des Dienststatus ausführen.

Der Code im folgenden Beispiel hält den Dienst my_service an und setzt ihn dann fort und ruft außerdem den Status des Dienstes ab:

my_service_res = root.databases["my_db"].schemas["my_schema"].services["my_service"]

my_service_res.suspend()
my_service_res.resume()
status = my_service_res.get_service_status(10)