Managing Snowpark Container Services (including service functions) 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 APIs, 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 APIs.
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)
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 APIs.
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 APIs 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 correspondingComputePool
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 named my_compute_pool
:
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)
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 named my_compute_pool
:
compute_pool = root.compute_pools["my_compute_pool"].fetch()
print(compute_pool.to_dict())
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 my
:
compute_pools = root.compute_pools.iter(like="my%")
for compute_pool in compute_pools:
print(compute_pool.name)
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:
compute_pool_res = root.compute_pools["my_compute_pool"]
compute_pool_res.suspend()
compute_pool_res.resume()
compute_pool_res.stop_all_services()
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 APIs 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 correspondingImageRepository
object and to drop 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 named my_repo
:
from snowflake.core.image_repository import ImageRepository
my_repo = ImageRepository("my_repo")
root.databases["my_db"].schemas["my_schema"].image_repositories.create(my_repo)
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:
my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo = my_repo_res.fetch()
print(my_repo.owner)
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:
repo_list = root.databases["my_db"].schemas["my_schema"].image_repositories.iter()
for repo_obj in repo_list:
print(repo_obj.name)
Dropping an image repository¶
You can drop an image repository using the ImageRepositoryResource.drop
method.
Code in the following example drops the my_repo
repository:
my_repo_res = root.databases["my_db"].schemas["my_schema"].image_repositories["my_repo"]
my_repo_res.drop()
Managing services and service functions¶
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 APIs 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 correspondingService
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")
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")
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 named my_service
from a specification in
@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)
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.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)
Creating a service function¶
After the service is up and running, you can create a service function that communicates with the service endpoint. A service function is a user-defined function (UDF) that you create and associate with a service in Snowpark Container Services. For more information, see Service functions: Using a service from an SQL query.
Code in the following example creates a UDF named my-udf
that specifies the hello-world
service and hello-world-endpoint
endpoint that you previously defined:
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
)
Invoking a service function¶
After the service function is created, you can then invoke the function to test it.
Code in the following example invokes the my-udf
service function that you previously created:
result = root.databases["my_db"].schemas["my_schema"].functions["my-udf(TEXT)"].execute_function(["test"])
print(result)
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 named my_service
:
my_service = root.databases["my_db"].schemas["my_schema"].services["my_service"].fetch()
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 my
:
services = root.databases["my_db"].schemas["my_schema"].services.iter(like="my%")
for service_obj in services:
print(service_obj.name)
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:
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)