---
title: Manage files across runs
description: Upload input files for a managed workflow, read and write them in a job, and download the outputs.
source: https://quantum.cloud.ibm.com/docs/en/managed-workflows/manage-files-across-managed-workflows
---

# Manage files across runs

A managed workflow job works like a batch job on an HPC compute node. The job's local disk is scratch space. The job can write to it, but the workflow deletes it when the job ends. To keep files after the job ends, save them to file storage, which works like your own storage bucket. Files that you upload, and files that your workflow saves, remain in file storage, and later runs of the workflow can read them.

Each workflow has its own file storage for each user. Other users who run the same workflow do not see your files.

![File storage for the workflow sits at the center. Your local notebook reaches it with the client file methods file\_upload(), files(), file\_download(), and file\_delete(). File storage is yours only, because other users of the workflow do not see your files. It is kept after each job ends and is shared by all of your runs, and a file with the same name replaces the earlier file. Each job sees file storage as the folder /function\_user\_data. Run 1 also has a local disk at /tmp, which is deleted when the job ends, and copies finished files from /tmp to /function\_user\_data. Run 2 reads the files that earlier runs saved.](https://quantum.cloud.ibm.com/docs/images/managed-workflows/manage-files.svg)

## Upload an input file

Upload a file from your local notebook with `file_upload()`, and then list the files in storage with `files()`:

```python
from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

client.file_upload("hamiltonian.npz", my_workflow)
print(client.files(my_workflow))
```

```text
['hamiltonian.npz']
```

### Supported file types

A managed workflow checks the contents of each uploaded file, not its extension, and rejects files of other types:

| Accepted                                                                                                                                  | Rejected                             |
| ----------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------ |
| JSON, plain text, YAML, and Python files; binary files such as NumPy (`.npy`, `.npz`), QPY, and pickle files; tar, gzip, and zip archives | CSV, PNG, PDF, HDF5, and empty files |

### File names

- **Use only letters, digits, underscores (`_`), hyphens (`-`), and periods (`.`).** A managed workflow removes other characters, such as spaces, from the name when you download or delete a file. For example, you can upload `my results.json`, but you cannot download it.
- **File storage has no folders.** `file_upload()` keeps only the file name, so `inputs/params.json` is stored as `params.json`.

### Read files in a workflow

Pass the name of the file when you run the workflow:

```python
job = my_workflow.run(input_file="hamiltonian.npz")
```

To enable this, edit your workflow code to read the file from `/function_user_data`, which is where every job sees file storage:

```python
from pathlib import Path

import numpy as np
from qiskit_serverless import get_arguments

FILE_STORAGE = Path("/function_user_data")

arguments = get_arguments()
hamiltonian = np.load(FILE_STORAGE / arguments["input_file"])
```

Inside a job, read and write files through this folder. The client file methods, such as `file_upload()` and `file_download()`, are for your local notebook.

To read an archive, extract it to the local disk:

```python
import tarfile

with tarfile.open(FILE_STORAGE / "inputs.tar.gz") as tar:
    tar.extractall("/tmp/inputs", filter="data")
```

Write temporary files to `/tmp`. File storage is object storage that the job sees as a folder, so file storage is slower than the local disk. Do heavy reads and writes on the local disk, and copy finished files to file storage.

## Save output files

Your local notebook reads the names of the output files from the result:

```python
result = job.result()
print(result["output_files"])  # ['energies.npy', 'summary.json']
```

To enable this, save each output file at the top level of `/function_user_data`, and return the file names with `save_result()`:

```python
import json

from qiskit_serverless import save_result

np.save(FILE_STORAGE / "energies.npy", energies)
(FILE_STORAGE / "summary.json").write_text(json.dumps({"min_energy": float(energies.min())}))

save_result({"output_files": ["energies.npy", "summary.json"]})
```

Files that the job writes outside `/function_user_data`, such as in `/tmp` or the current folder, are deleted when the job ends.

### Download output files

After the job ends, download each file with `file_download()`. Set `target_name` to choose the local file name, and `download_location` to choose an existing local folder:

```python
from pathlib import Path

Path("outputs").mkdir(exist_ok=True)

for name in result["output_files"]:
    client.file_download(name, my_workflow, target_name=name, download_location="outputs")
```

Without `target_name`, `file_download()` adds a random prefix to the local file name, such as `downloaded_3f2a9c1b_energies.npy`, and returns the name that it used.

### Download many outputs as one archive

If a job saves many output files, or files in subfolders, bundle them into one archive in the workflow, and download only the archive:

```python
import tarfile

client.file_download(result["archive"], my_workflow, target_name=result["archive"], download_location="outputs")

with tarfile.open(Path("outputs") / result["archive"]) as tar:
    tar.extractall("outputs", filter="data")
```

To enable this, write the outputs to a folder on the local disk, add the folder to an archive in `/tmp`, and copy the archive to file storage:

```python
import shutil
import tarfile

output_dir = Path("/tmp/outputs")
output_dir.mkdir()
np.save(output_dir / "energies.npy", energies)
# Save other outputs, including subfolders, to output_dir

archive_name = f"{run_name}_outputs.tar.gz"
with tarfile.open(Path("/tmp") / archive_name, "w:gz") as tar:
    tar.add(output_dir, arcname=run_name)
shutil.copy(Path("/tmp") / archive_name, FILE_STORAGE / archive_name)

save_result({"archive": archive_name})
```

Because the archive is built on the local disk and copied in one step, file storage never holds an incomplete archive. The archive keeps the folder structure, and it can contain files of any type. In this example, the files are extracted to `outputs/<run_name>/`.

## Delete files

A managed workflow does not delete files from file storage, so files remain until you delete them with `file_delete()`:

```python
client.file_delete("hamiltonian.npz", my_workflow)
```

`file_delete()` reports success even if the file does not exist, so check the name with `files()` first. To delete all the files of one run, filter `files()` by the run's name prefix, as described in the next section:

```python
for name in client.files(my_workflow):
    if name.startswith("theta-0.1_"):
        client.file_delete(name, my_workflow)
```

## Best practice: give each run's files a unique name

All of your runs of a workflow share the same file storage, and a file that a job saves replaces any stored file with the same name. If you start several runs at the same time and each run saves `energies.npy`, only the last one is kept.

As a best practice, pass a run name as an argument, and start the name of every file that the run saves with it:

```python
jobs = {}
for theta in [0.1, 0.2, 0.3]:
    run_name = f"theta-{theta}"
    jobs[run_name] = my_workflow.run(run_name=run_name, theta=theta, input_file="hamiltonian.npz")
```

```python
run_name = arguments["run_name"]
np.save(FILE_STORAGE / f"{run_name}_energies.npy", energies)
```

All runs can read the same input file, `hamiltonian.npz`, because they only read it. Upload a shared input one time, before you start the runs.

Use run names that you can recognize later, and that contain only letters, digits, underscores, hyphens, and periods. With one prefix for each run, you can find, download, and [delete](#delete-files) the files of a run together.

## Best practice: checkpoint long jobs

A job that is canceled, fails, or reaches the 24-hour limit loses everything that is not in file storage. For long runs, save a checkpoint to file storage at regular intervals, and resume from it when you run the workflow again with the same run name:

```python
import json
import shutil

checkpoint = FILE_STORAGE / f"{run_name}_checkpoint.json"

state = {"step": 0, "params": initial_params}
if checkpoint.exists():
    state = json.loads(checkpoint.read_text())

for step in range(state["step"], num_steps):
    state["params"] = optimize_step(state["params"])
    state["step"] = step + 1

    if state["step"] % 10 == 0:
        local_copy = Path("/tmp") / checkpoint.name
        local_copy.write_text(json.dumps(state))
        shutil.copy(local_copy, checkpoint)
```

To continue a run that stopped, call `run()` again with the same arguments.

Keep the following points in mind:

- **Write the checkpoint to the local disk first, and then copy it.** If the job stops during a write, the stored checkpoint is incomplete, and the next run cannot read it.
- **Choose an interval that matches the cost of a step.** Save often enough that you do not lose much work, but not so often that writing checkpoints slows the job.
- **Delete checkpoints after the run succeeds.** A later run with the same name then starts from the beginning.

For a job that is canceled, you can also save partial results in a [`SIGTERM` handler](/docs/managed-workflows/managed-workflow-logs-and-errors#clean-up-when-a-job-is-canceled).

## Next steps

> **Recommendations**
>
> - [Run a parameter sweep](/docs/managed-workflows/run-parameter-sweep) to start one job for each input.
> - [Write logs and handle errors](/docs/managed-workflows/managed-workflow-logs-and-errors) to save partial results when a job is canceled.