Managing Snowpark Container Services with Python

You can use Python to manage Snowpark Container Services, a fully managed container service through which you can deploy, manage, and scale containerized applications. For an overview of Snowpark Container Services, see About Snowpark Container Services.

With the Snowflake Python API, you can manage compute pools, image repositories, and services.

Prerequisites

The examples in this topic assume that you’ve added code to connect with Snowflake and to create a Root object from which to use the Snowflake Python API.

For example, the following code uses connection parameters defined in a configuration file to create a connection to Snowflake:

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

Using the resulting Session object, the code creates a Root object to use the API’s types and methods. For more information, see Connect to Snowflake with the Snowflake Python API.

Managing compute pools

You can manage compute pools, which are collections of virtual machine (VM) nodes on which Snowflake runs your Snowpark Container Services jobs and services.

The Snowflake Python API represents compute pools with two separate types:

  • ComputePool: Exposes a compute pool’s properties, such as its warehouse, maximum and minimum nodes, and auto resume and auto suspend settings.

  • ComputePoolResource: Exposes methods for performing actions on compute pools, such as fetching a corresponding ComputePool object and suspending, resuming, and stopping pools.

For more information about compute pools, see Snowpark Container Services: Working with compute pools.

Creating a compute pool

You can create a compute pool by calling the ComputePoolCollection.create method, passing a ComputePool object that represents the compute pool you want to create.

To create a compute pool, first create a ComputePool object that specifies pool properties such as the following:

  • Compute pool name

  • Maximum and minimum number of nodes that the pool will contain

  • Name of the instance family that identifies the type of machine to provision for nodes in the pool

  • Whether the pool should automatically resume when a service or job is submitted to it

Code in the following example creates a ComputePool object that represents a pool called my_compute_pool:

from snowflake.core import Root
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)
Copy

The code then creates the compute pool by passing the ComputePool object to the ComputePoolCollection.create method.

Getting compute pool details

You can get information about a compute pool by calling the ComputePoolResource.fetch method, which returns a ComputePool object.

Code in the following example gets information about a pool called my_compute_pool:

from snowflake.core import Root
from snowflake.core.compute_pool import ComputePool

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

Creating or updating a compute pool

You can update characteristics of an existing compute pool by setting properties of a ComputePool object that represents an existing pool, and then passing the updated object to Snowflake with the create_or_update method, which returns a ComputePoolResource object.

You can also pass a ComputePool object describing a new pool when you want to create the pool.

Code in the following example sets the my_compute_pool compute pool’s maximum allowed nodes and then updates the pool on Snowflake:

from snowflake.core import Root
from snowflake.core.compute_pool import ComputePool

compute_pool = root.compute_pools["my_compute_pool"].fetch()
compute_pool.max_node = 3
compute_pool_res = root.compute_pools.create_or_update(compute_pool)
Copy

Listing compute pools

You can list compute pools using the iter method, which returns a PagedIter iterator.

Code in the following example lists compute pools whose name begins with abc:

from snowflake.core import Root

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

Performing compute pool operations

You can perform common compute pool operations—such as suspending, resuming, and stopping pools—with a ComputePoolResource object, which you can get by using the ComputePool.fetch method.

Code in the following example suspends, resumes, and stops the my_compute_pool compute pool:

from snowflake.core import Root
from snowflake.core.compute_pool import ComputePoolResource

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

The code uses the Root.compute_pools method to create a ComputePool object representing the compute pool. From the ComputePool object, it fetches a ComputePoolResource object with which to perform compute pool operations.

Managing image repositories

You can manage image repositories, which store images for applications you run on container services.

An image repository is a schema-level object. When you create or reference a repository, you do so in the context of its schema.

The Snowflake Python API represents image repositories with two separate types:

  • ImageRepository: Exposes an image repository’s properties, such as its database and schema names, repository URL, and owner.

  • ImageRepositoryResource: Exposes methods you can use to fetch a corresponding ImageRepository object and to delete the image repository resource.

For more information about image repositories, see Snowpark Container Services: Working with an image registry and repository.

Creating an image repository

To create an image repository, first create an ImageRepository object that specifies the repository name.

Code in the following example creates an ImageRepository object that represents a repository called my_repo:

from snowflake.core import Root
from snowflake.core.image_repository import ImageRepository

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

The code then creates the image repository by passing the ImageRepository object to the ImageRepositoryCollection.create method, creating the image repository in the my_db database and my_schema schema.

Getting image repository details

You can get information about an image repository by calling the ImageRepositoryResource.fetch method, which returns an ImageRepository object.

Code in the following example gets an ImageRepository object representing the my_repo image repository and then prints the name of the repository’s owner:

from snowflake.core import Root
from snowflake.core.image_repository import ImageRepository

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

Listing image repositories

You can list the image repositories in a specified schema using the iter method, which returns a PagedIter iterator of ImageRepository objects.

Code in the following example lists repository names in the my_db database and my_schema schema:

from snowflake.core import Root

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

Deleting an image repository

You can delete an image repository using the ImageRepositoryResource.delete method.

Code in the following example deletes the my_repo repository:

from snowflake.core import Root
from snowflake.core.image_repository import ImageRepositoryResource

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

Managing services

You can manage services, which run application containers until you stop them. Snowflake restarts a service automatically if the service container stops. In this way, the service effectively runs uninterrupted.

A service is a schema-level object. When you create or reference a service, you do so in the context of its schema.

The Snowflake Python API represents services with two separate types:

  • Service: Exposes a service’s properties such as its specification, minimum and maximum instances, and database and schema name.

  • ServiceResource: Exposes methods you can use to fetch a corresponding Service object, suspend and resume the service, and get its status.

For more information about services, see Snowpark Container Services: Working with services.

Creating a service

To create a service, you run the services.create method, passing a Service object representing the service you want to create.

You create a service from a service specification .yaml file that has been uploaded to a stage. For more information about creating a service specification, see Service specification reference.

Uploading the specification

If you’re creating a service from a specification that hasn’t yet been uploaded to a stage, you can upload the specification using a Snowpark FileOperation object.

Code in the following example uses the FileOperation.put method to upload a specification as a file:

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

Code in the following example uses the FileOperation.put_stream method to upload a specification as a string:

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

Creating the service

To create a service from a staged specification, first create a Service object that specifies service properties such as the following:

  • Service name

  • Maximum and minimum number of service instances that Snowflake can create

  • Compute pool to which the service should be added

  • Stage location and name of the specification

Code in the following example creates a Service object representing a service called my_service from a specification in @my_stage/my_service_spec.yaml:

from snowflake.core import Root
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)
Copy

The code then creates the service by passing the Service object to the ServiceCollection.create method, creating the service in the my_db database and my_schema schema.

You can also create a service from a specification that you provide as inline text, as shown in the following example. The ServiceSpec function takes a single string argument spec. If the string starts with @, the function interprets and validates it as a stage file path. Otherwise the string is passed through as inline text.

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

spec_text = dedent(f"""\
    spec:
      containers:
      - name: hello-world
        image: repo/hello-world:latest
    """)

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)
Copy

Getting service details

You can get information about a Snowflake service by calling the ServiceResource.fetch method, which returns a Service object.

Code in the following example gets information about a service called my_service:

from snowflake.core import Root
from snowflake.core.service import Service

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

Listing services

You can list the services in a specified schema using the iter method, which returns a PagedIter iterator of Service objects.

Code in the following example lists services whose name begins with abc:

from snowflake.core import Root

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

Performing service operations

You can perform common service operations—such as suspending, resuming, and getting service status—with a ServiceResource object.

Code in the following example suspends and resumes the my_service service and also gets the service’s status:

from snowflake.core import Root
from snowflake.core.service import ServiceResource

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)
Copy
Language: English