ManagedWorkflow

class qiskit_ibm_catalog.ManagedWorkflow(title, provider=None, entrypoint=None, working_dir='./', env_vars=None, dependencies=None, description=None, version=None, tags=None, raw_data=None, image=None, runner='fleets', arguments_schema=None, type='GENERIC', sizes_map=None, default_size=None)

Defines a workflow to upload with ManagedWorkflowClient.upload(). QiskitFunction is an alias.

upload() and load() return an uploaded workflow that you can run.

from qiskit_ibm_catalog import ManagedWorkflow

workflow = ManagedWorkflow(
    title="my-workflow",
    entrypoint="main.py",
    working_dir="./src",
    sizes_map={"S": "8x32", "M": "16x128", "L": "128x1500"},
    default_size="M",
)

Parameters

title (str) – Workflow name.
entrypoint (str | None) – Script that runs as the job, relative to working_dir. For example, "main.py".
working_dir (str) – Directory that contains the entrypoint. Maximum size is 50 MB.
sizes_map (dict[str, str | ComputeProfile] | None) – Maps size labels to compute profile IDs or ComputeProfile objects. Choose a size per run with run(function_size=...). Labels are case-insensitive. See Choose a compute profile.
default_size (str | None) – Size label to use when a run doesn't pass function_size. Must be a label in sizes_map.
arguments_schema (dict | bool | None) – JSON Schema for the workflow's arguments. {} removes a schema stored by an earlier upload. Qiskit objects are checked in their encoded form, so constrain only their __type__ tag. See Define inputs and outputs.
env_vars (dict[str, str] | None) – Environment variables to set in the job.
description (str | None) – Workflow description.
version (str | None) – Workflow version.
tags (list[str] | None) – Tags for the workflow.
image (str | None) – Container image to run the workflow in.
runner (str) – Job runner. Defaults to "fleets".
type (str) – One of "GENERIC", "APPLICATION", or "CIRCUIT".
dependencies (list[str] | None) – Deprecated and ignored. Use dependencies_versions() and add_wheels() instead.

Uploaded workflow

Returned by ManagedWorkflowClient.upload() and ManagedWorkflowClient.load(). Its class is RunnableQiskitFunction.

get_default_size

get_default_size()

Returns the size label used when a run doesn't pass function_size.

print(my_workflow.get_default_size())

Output:

Return type

str | None – None if the workflow doesn't declare a default.

jobs

jobs(**kwargs)

Lists the jobs created with this workflow. Replaces the deprecated get_jobs().

for job in my_workflow.jobs():
    print(job.job_id, job.status())

Output:

e6b1ffa1-cd26-437d-bd53-b265dd4ba952 DONE
11308f2a-8d7a-40a0-a970-7fdeca2b171b ERROR

Return type

list[Job]

run

run(**kwargs)

Starts a job and returns immediately.

job = my_workflow.run(circuits=[qc], function_size="L")
print(job.job_id)

Output:

e6b1ffa1-cd26-437d-bd53-b265dd4ba952

Parameters

**kwargs – The workflow's inputs. Your entrypoint reads them with get_arguments().
function_size (str | None) – Size label from the workflow's sizes_map. Reserved keyword.
config (Configuration | None) – Job configuration. Reserved keyword.

Return type

Job

Raises

QiskitServerlessException – If the arguments are invalid. Passing compute_profile also raises this error; use function_size instead.

sizes

sizes()

Returns the size labels the workflow declares and the compute profile each one maps to.

print(my_workflow.sizes())

Output:

{'S': '8x32', 'M': '16x128', 'L': '128x1500'}

Return type

dict[str, str] – Empty if the workflow declares no sizes.

validate_arguments

validate_arguments(arguments)

Checks arguments against the workflow's arguments_schema without creating a job.

my_workflow.validate_arguments({"circuits": [circuit], "shots": 1024})

Output:

{'valid': True}

Parameters

arguments (dict) – The arguments to check.

Return type

dict – {"valid": True} if the arguments are valid.

Raises

QiskitServerlessException – If the arguments are invalid.

Was this page helpful?

Report a bug, typo, or request content on GitHub.