# Managing temporary resources

Sometimes you have workloads that depend on resources that need to be setup, used, then cleaned up. Examples might include:

  • Compute resources like Dask or Spark clusters
  • Cloud resources like Virtual machines or Kubernetes deployments
  • Docker containers
  • ...

To simplify this pattern, Prefect provides a ResourceManager object for encapsulating the setup and cleanup tasks for resources. The most common way to define a ResourceManager is using the resource_manager decorator. This decorator wraps a class with the following methods:

  • __init__: Initializes the resource manager with whatever arguments are needed.
  • setup: Creates the resource. Takes no arguments, and may optionally return a value that can be used by downstream tasks.
  • cleanup: Cleans up the resource. Takes the result of setup as an argument.
from prefect import resource_manager

@resource_manager
class MyResource:
    def __init__(self, ...):
        """Initialize the resource manager.

        This should store any values required by the setup and cleanup steps.
        """
        ...

    def setup(self):
        """Setup the resource.

        The result of this method can be used in downstream tasks.
        """
        ...

    def cleanup(self, resource):
        """Cleanup the resource.

        This receives the result of `setup`, and is always called if `setup`
        succeeds, even if other upstream tasks failed.
        """
        ...

The resulting ResourceManager can then be used when building a Flow as a context-manager around tasks that rely on that resource. The resource will be created upon entering the context block, and will be cleaned up upon exiting the block, even if tasks contained inside the context fail.

with Flow("example") as flow:
    with MyResource(...) as resource:
        some_task(resource)
        other_task(resource)

ResourceManager objects are intended for defining resources where the cleanup of the resource should also be monitored and managed by Prefect as a Task in your Flow. This is good for things that can be expensive, like cloud resources. For things where a failure to cleanup an object isn't detrimental (like e.g. a boto client) you may be better off relying on other patterns.

# Example: Creating a temporary Dask Cluster

Here we provide a full example for using a ResourceManager to setup and cleanup a temporary Dask cluster.