Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

We need `ResourceManager` `map`ped to list of days.
We have a task, archiving data for one day. The task is using temporary dir (mapped to shared memory) and we use ResourceManager to cleanup the tmp dir when a day is processed (there is day subdir per processed day).
Now we want to run the same for list of days and we struggle with mapping days to resource manager.

To me it seems like the Resource Manager cannot be simply mapped this way.

Hey <@U01HLS4KD4N>, I think I understand what you are saying. Do you think it can be achieved if you did:
```@task
def abc(x):
    with TempDir() as tmpdir:
         # do stuff
    return```

But yeah the map would create multiple instances.

<@U01QEJ9PP53> what you propose, looks exactly as the solution we need.
It is inverted to what is in doc (manager here is not managing the task from outside but serving inside).
It shall definitely simplify our code.

Oh ok. Do you know how to do that? with the `___enter___` and `___exit___` methods?

I think the resource manager in the docs is more for temporary compute resources.

I sometime forget about standard tools we have at hand.

My temporary directory context manager:
```import shutil
from contextlib import contextmanager
from pathlib import Path

import prefect


@contextmanager
def managed_tmp_dir(relative_workdir):
    base_dir = prefect.context.base_dir.path
    base_path = Path(base_dir) / relative_workdir
    base_path.mkdir(exist_ok=True, parents=True)
    try:
        yield base_path
    finally:
        shutil.rmtree(base_path, ignore_errors=True)```


Used in task as follows:
```class CreateDailyArchive(ShellTask):
    def run(self, f1_sample_bucket_key, day_str, max_versions):
        """Fetch XMLs for one day and create an archive.

        It attemtps to clean up all used resources as soon as possible, but it is advisable
        to have follow up task wchich unconditionally cleans up what could be left.
        """
        logger = prefect.context.get("logger")

        day = datetime.strptime(day_str, "%Y-%m-%d")

        key_namer = KeyNamer(prefect.context, f1_sample_bucket_key)
        source_bucket_key = key_namer.f1_sample()
        target_bucket_key = key_namer.daily_archive(day)

        relative_workdir = key_namer.f2_day_workdir(day)

        with managed_tmp_dir(relative_workdir) as workdir:

            <http://logger.info|logger.info>("Getting f1_dvi.")
            f1_dvi = self.get_f1_dvi(f1_sample_bucket_key, day)

            <http://logger.info|logger.info>("Fetching versions.")
            xmlgz_dir = self.fetch_files(
                source_bucket_key, f1_dvi, workdir, max_versions
            )```