# This code is a Qiskit project.
#
# (C) Copyright IBM 2022.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""
================================================
Provider (:mod:`qiskit_serverless.core.client`)
================================================
.. currentmodule:: qiskit_serverless.core.client
Qiskit Serverless provider
===========================
.. autosummary::
:toctree: ../stubs/
ServerlessClient
"""
# pylint: disable=duplicate-code,too-many-lines
import glob
import json
import os.path
import os
import re
import tarfile
import time
import warnings
from pathlib import Path
from urllib.parse import urlparse
from dataclasses import asdict, dataclass
from typing import Optional, List, Dict, Any, Union
from collections.abc import Callable
import requests
from opentelemetry import trace
from qiskit.providers.backend import BackendV2 as Backend
from qiskit.providers.exceptions import QiskitBackendNotFoundError
from qiskit_ibm_runtime import QiskitRuntimeService, IBMBackend
from qiskit_ibm_runtime.accounts.exceptions import InvalidAccountError
from qiskit_serverless.core.constants import (
REQUESTS_TIMEOUT,
ENV_GATEWAY_PROVIDER_HOST,
ENV_GATEWAY_PROVIDER_VERSION,
ENV_GATEWAY_PROVIDER_TOKEN,
GATEWAY_PROVIDER_VERSION_DEFAULT,
IBM_SERVERLESS_HOST_URL,
MAX_ARTIFACT_FILE_SIZE_MB,
USAGE_LOW_THRESHOLD_SECONDS,
USAGE_ZERO_EPSILON_SECONDS,
)
from qiskit_serverless.core.client import BaseClient
from qiskit_serverless.core.decorators import trace_decorator_factory
from qiskit_serverless.core.enums import Channel
from qiskit_serverless.core.files import GatewayFilesClient
from qiskit_serverless.core.wheels import WHEELS_SUBDIR
from qiskit_serverless.core.job import (
Job,
Configuration,
_map_status_to_serverless,
)
from qiskit_serverless.core.job_event import JobEvent
from qiskit_serverless.core.compute_profile import (
ComputeProfile,
ComputeProfileList,
format_profiles,
)
from qiskit_serverless.core.function import (
QiskitFunction,
RunService,
RunnableQiskitFunction,
RAY_RUNNER_DEPRECATION,
)
from qiskit_serverless.exception import QiskitServerlessException
from qiskit_serverless.utils.http import get_headers
from qiskit_serverless.utils.json import (
safe_json_request_as_dict,
safe_json_request_as_list,
safe_json_request,
raise_for_non_ok_response,
)
from qiskit_serverless.utils.formatting import format_provider_name_and_title
from qiskit_serverless.serializers.program_serializers import (
QiskitObjectsEncoder,
QiskitObjectsDecoder,
)
_trace_job = trace_decorator_factory("job")
_trace_functions = trace_decorator_factory("function")
_STATUS_TOKEN_HEADER = "X-Job-Status-Token"
# The gateway signs the token for 5 minutes, so the client stops using it 30 s before
_STATUS_TOKEN_REUSE_SECONDS = 270
_TOKEN_REFUSED_CODES = (401, 403, 404)
# A requirement looks like "name[extras]<specifier>", e.g. "numpy>=2",
# "qiskit-addon-aqc-tensor[quimb-jax]==0.3.1", or a bare "mypkg".
_REQUIREMENT_RE = re.compile(r"^\s*([A-Za-z0-9._-]+)\s*(?:\[[^\]]*\])?\s*(.*?)\s*$")
def _split_requirement(requirement: str) -> tuple:
"""Split a requirement string into (name, specifier-or-version).
For a baked entry like ``"numpy==2.1.0"`` the second item is the pinned version
``"2.1.0"``; for a user requirement like ``"jax<0.5"`` it is the specifier ``"<0.5"``.
Extras are dropped. A bare name yields an empty specifier.
"""
match = _REQUIREMENT_RE.match(requirement)
if not match:
return requirement.strip(), ""
name, spec = match.group(1), match.group(2).strip()
# Normalize a bare "==x" baked pin down to just the version for display/comparison.
if spec.startswith("=="):
spec = spec[2:].strip()
return name, spec
def _normalize_dep_name(name: str) -> str:
"""PEP 503 name normalization: lowercase, runs of -_. collapse to a single '-'."""
return re.sub(r"[-_.]+", "-", name).lower()
def _spec_allows(spec: str, baked_version: str) -> bool:
"""Best-effort check that a baked version satisfies a user specifier.
Deliberately simple and dependency-free: an exact ``==`` or bare version must match the
baked version; anything else (ranges like ``>=2``, ``<0.5``) is treated as compatible so the
report does not raise false clashes. The report is advisory, not an installer.
"""
spec = spec.strip()
if not spec:
return True
if spec.startswith("=="):
return spec[2:].strip() == baked_version
# A bare version with no operator (e.g. report entry already reduced to "2.1.0").
if re.match(r"^[0-9]", spec):
return spec == baked_version
return True
@dataclass
class DependencyReportEntry:
"""One row of a :class:`DependencyReport`.
``status`` is one of ``"baked"`` (already in the image, don't bring), ``"bring"`` (upload a
wheel), or ``"clash"`` (the image ships a different version).
"""
requirement: str
status: str
note: str
class DependencyReport(List[DependencyReportEntry]):
"""A list of :class:`DependencyReportEntry` that prints as an aligned table.
Behaves like an ordinary list; only ``str``/``repr`` differ so
``print(client.dependency_report([...]))`` shows the table without the caller iterating.
"""
def __str__(self) -> str:
return _format_dependency_report(self)
def __repr__(self) -> str:
return _format_dependency_report(self)
def _format_dependency_report(entries: List[DependencyReportEntry]) -> str:
header = ("requirement", "status", "note")
rows = [header] + [(e.requirement, e.status, e.note) for e in entries]
widths = [max(len(row[col]) for row in rows) for col in range(3)]
return "\n".join(" ".join(cell.ljust(widths[col]) for col, cell in enumerate(row)) for row in rows)
[docs]
class ServerlessClient(BaseClient): # pylint: disable=too-many-public-methods
"""
A client for connecting to a specified host.
Example:
>>> client = ServerlessClient(
>>> host="<HOST>",
>>> token="<TOKEN>",
>>> instance="<CRN>",
>>> )
"""
[docs]
def __init__( # pylint: disable=too-many-positional-arguments
self,
host: Optional[str] = None,
version: Optional[str] = None,
token: Optional[str] = None,
instance: Optional[str] = None,
channel: Optional[str] = None,
):
"""
Initializes the ServerlessClient instance.
Args:
host: host of gateway. If None, it uses the ENV_GATEWAY_PROVIDER_HOST env var
version: version of gateway
token: authorization token
instance: IBM Cloud CRN
channel: identifies the method to use to authenticate the user
"""
host = host or os.environ.get(ENV_GATEWAY_PROVIDER_HOST)
if host is None:
raise QiskitServerlessException("Please provide `host` of gateway.")
host = host.rstrip("/")
version = version or os.environ.get(ENV_GATEWAY_PROVIDER_VERSION)
if version is None:
version = GATEWAY_PROVIDER_VERSION_DEFAULT
token = token or os.environ.get(ENV_GATEWAY_PROVIDER_TOKEN)
if token is None:
raise QiskitServerlessException("Authentication credentials must be provided in form of `token`.")
channel = channel or Channel.IBM_QUANTUM_PLATFORM.value
try:
channel_enum = Channel(channel)
except ValueError as error:
raise ValueError(
"Your channel value is not correct. Use one of the available channels: "
f"{Channel.LOCAL.value}, "
f"{Channel.IBM_CLOUD.value}, {Channel.IBM_QUANTUM_PLATFORM.value}"
) from error
if channel_enum is Channel.IBM_CLOUD and instance is None:
raise QiskitServerlessException("Authentication with IBM Cloud requires to pass the CRN as an instance.")
if channel_enum is Channel.IBM_QUANTUM_PLATFORM and instance is None:
raise QiskitServerlessException(
"Authentication with IBM Quantum Platform requires to pass the CRN as an instance."
)
super().__init__(host, token, instance, channel)
self.version = version
# job id -> (job status token, monotonic time it was received)
self._status_tokens: Dict[str, tuple[str, float]] = {}
self._verify_credentials()
self._files_client = GatewayFilesClient(self.host, self.token, self.version, self.instance, self.channel)
@classmethod
def from_dict(cls, dictionary: dict):
# Remove 'name' if present for backward compatibility with serialized clients
data = {k: v for k, v in dictionary.items() if k != "name"}
return ServerlessClient(**data)
def _verify_credentials(self):
"""Verify against the API that the credentials are correct."""
try:
safe_json_request(
request=lambda: requests.get(
url=f"{self.host}/api/v1/programs/",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
except QiskitServerlessException as reason:
raise QiskitServerlessException(f"Credentials couldn't be verified: {reason}") from reason
def dependencies_versions(self):
"""Discover the dependencies (and versions) baked into the runtime image.
Read-only discovery: this reports what the runtime image already ships, so you know
what you do *not* need to bring yourself.
"""
return safe_json_request_as_list(
request=lambda: requests.get(
url=f"{self.host}/api/{self.version}/dependencies-versions/",
headers=get_headers(token=self.token, instance=self.instance),
timeout=REQUESTS_TIMEOUT,
)
)
def compute_profiles(self) -> ComputeProfileList:
"""Get the compute profiles registered on this deployment.
These are the profiles a Fleets function's ``sizes_map`` can name. The catalog is
deployment-specific, so discover it here rather than hard-coding identifiers. The
returned list prints as an aligned table, so ``print(client.compute_profiles())`` shows
the catalog without iterating, while still behaving as an ordinary list of
:class:`ComputeProfile`.
"""
profiles = safe_json_request_as_list(
request=lambda: requests.get(
url=f"{self.host}/api/{self.version}/compute-profiles/",
headers=get_headers(token=self.token, instance=self.instance),
timeout=REQUESTS_TIMEOUT,
)
)
return ComputeProfileList(ComputeProfile.from_json(profile) for profile in profiles)
def compute_profile(
self,
compute_profile_id: Optional[str] = None,
*,
cpu: Optional[Union[int, str]] = None,
memory: Optional[Union[int, str]] = None,
gpu: Optional[str] = None,
) -> ComputeProfile:
"""Get a single registered compute profile, by id or by exact resources.
Validates against the deployment's catalog and returns the matching
:class:`ComputeProfile`, so the result is always a profile the deployment has
registered. Call it one of two ways:
- by id: ``client.compute_profile("24x120")``
- by resources: ``client.compute_profile(cpu=24, memory=120[, gpu="1l40s"])`` — ``cpu``
and ``memory`` are required (``int`` or ``str``), ``gpu`` optional (omitted means
CPU-only). Every value given must match a single registered profile exactly.
Raises:
QiskitServerlessException: if no profile matches (the message lists the available
profiles), or if the arguments mix the two forms / omit required resources.
"""
by_id = compute_profile_id is not None
by_resources = cpu is not None or memory is not None or gpu is not None
if by_id and by_resources:
raise QiskitServerlessException("Pass either a compute_profile_id or cpu/memory/gpu, not both.")
if not by_id and not by_resources:
raise QiskitServerlessException("Provide a compute_profile_id, or cpu and memory.")
profiles = self.compute_profiles()
if by_id:
match = next((p for p in profiles if p.compute_profile_id == compute_profile_id), None)
criteria = f"compute_profile_id={compute_profile_id!r}"
else:
if cpu is None or memory is None:
raise QiskitServerlessException("Both cpu and memory are required to match by resources.")
match = next(
(p for p in profiles if p.cpu == str(cpu) and p.memory == str(memory) and p.gpu == gpu),
None,
)
criteria = f"cpu={cpu}, memory={memory}" + (f", gpu={gpu!r}" if gpu is not None else "")
if match is None:
available = format_profiles(profiles) or "none are registered on this deployment"
raise QiskitServerlessException(f"No compute profile matches {criteria}. Available profiles:\n{available}")
return match
####################
####### JOBS #######
####################
@_trace_job("list")
def jobs(self, function: Optional[QiskitFunction] = None, **kwargs) -> List[Job]:
"""Retrieve a list of jobs with optional filtering.
Args:
function (QiskitFunction): The function that created the jobs we want to retrieve.
limit (int, optional): Maximum number of jobs to return. Defaults to 10.
offset (int, optional): Number of jobs to skip. Defaults to 0.
status (str, optional): Filter by job status.
created_after (str, optional): Filter jobs created after this timestamp.
**kwargs: Additional query parameters.
Returns:
List[Job]: List of Job objects matching the criteria.
"""
limit = kwargs.get("limit", 10)
kwargs["limit"] = limit
offset = kwargs.get("offset", 0)
kwargs["offset"] = offset
status = kwargs.get("status", None)
if status:
status, _ = _map_status_to_serverless(status)
kwargs["status"] = status
created_after = kwargs.get("created_after", None)
kwargs["created_after"] = created_after
if function:
kwargs["function"] = function.title
kwargs["provider"] = function.provider
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/",
params=kwargs,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
return [
Job(
job.get("id"),
job_service=self,
raw_data=job,
)
for job in response_data.get("results", [])
]
@_trace_job("provider_list")
def provider_jobs(self, function: Optional[QiskitFunction], **kwargs) -> List[Job]:
"""Retrieve jobs for a specific provider and function.
Args:
function (QiskitFunction): Function object.
limit (int, optional): Maximum number of jobs to return. Defaults to 10.
offset (int, optional): Number of jobs to skip. Defaults to 0.
status (str, optional): Filter by job status.
created_after (str, optional): Filter jobs created after this timestamp.
**kwargs: Additional query parameters.
Returns:
List[Job]: List of Job objects for the specified provider and function.
Raises:
QiskitServerlessException: If the function doesn't have an associated provider.
"""
if not function.provider:
raise QiskitServerlessException("`function` doesn't have a provider.")
limit = kwargs.get("limit", 10)
kwargs["limit"] = limit
offset = kwargs.get("offset", 0)
kwargs["offset"] = offset
status = kwargs.get("status", None)
if status:
status, _ = _map_status_to_serverless(status)
kwargs["status"] = status
created_after = kwargs.get("created_after", None)
kwargs["created_after"] = created_after
if function:
kwargs["function"] = function.title
kwargs["provider"] = function.provider
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/provider/",
params=kwargs,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
return [
Job(
job.get("id"),
job_service=self,
raw_data=job,
)
for job in response_data.get("results", [])
]
@_trace_job("get")
def job(self, job_id: str) -> Optional[Job]:
url = f"{self.host}/api/{self.version}/jobs/{job_id}/"
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
url,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
params={"with_result": "false"},
timeout=REQUESTS_TIMEOUT,
)
)
job = None
job_id = response_data.get("id")
if job_id is not None:
job = Job(
job_id=job_id,
job_service=self,
raw_data=response_data,
)
return job
def run(
self,
program: Union[QiskitFunction, str],
arguments: Optional[Dict[str, Any]] = None,
config: Optional[Configuration] = None,
provider: Optional[str] = None,
*,
function_size: Optional[str] = None,
) -> Job:
if isinstance(program, QiskitFunction):
title = program.title
provider = program.provider
ran_on_ray = program.runner == "ray"
else:
title = str(program)
# A bare title gives us no runner to inspect, so we cannot tell locally whether this
# ran on Ray. The gateway still returns its deprecation text in the warning field
# below; we just surface it as a plain warning rather than a DeprecationWarning.
ran_on_ray = False
tracer = trace.get_tracer("client.tracer")
with tracer.start_as_current_span("job.run") as span:
span.set_attribute("function", title)
span.set_attribute("provider", provider)
span.set_attribute("arguments", str(arguments))
url = f"{self.host}/api/{self.version}/programs/run/"
data = {
"title": title,
"provider": provider,
"function_size": function_size,
"arguments": json.dumps(arguments or {}, cls=QiskitObjectsEncoder),
} # type: Dict[str, Any]
if config:
data["config"] = asdict(config)
else:
data["config"] = asdict(Configuration())
response_data = safe_json_request_as_dict(
request=lambda: requests.post(
url=url,
json=data,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
job_id = response_data.get("id")
span.set_attribute("job.id", job_id)
if response_data.get("warning"):
# The gateway returns the warning text (Ray deprecation, a blocked resource, or both)
# so every client, old or new, is covered. When we can see locally that the function
# ran on the deprecated Ray runner, raise the category to DeprecationWarning so it is
# filterable as one; otherwise keep the default warning category.
category = DeprecationWarning if ran_on_ray else UserWarning
warnings.warn(response_data["warning"], category=category, stacklevel=2)
return Job(
job_id,
job_service=self,
raw_data=response_data,
)
def get_job_data(self, job_id: str) -> Optional[dict]:
"""The job, read with its status token while the token is young, and with the API key otherwise."""
cached = self._status_tokens.get(job_id)
if cached is not None and time.monotonic() - cached[1] < _STATUS_TOKEN_REUSE_SECONDS:
try:
return safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/status/",
headers={_STATUS_TOKEN_HEADER: cached[0]},
timeout=REQUESTS_TIMEOUT,
)
)
except QiskitServerlessException as error:
if getattr(error, "_code", None) not in _TOKEN_REFUSED_CODES:
raise
self._status_tokens.pop(job_id, None)
responses = []
def request():
responses.append(
requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/",
params={"with_result": "false"},
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
return responses[0]
data = safe_json_request_as_dict(request=request) or None
token = (getattr(responses[0], "headers", None) or {}).get(_STATUS_TOKEN_HEADER)
if token:
now = time.monotonic()
for old in [
key for key, (_, at) in list(self._status_tokens.items()) if now - at >= _STATUS_TOKEN_REUSE_SECONDS
]:
self._status_tokens.pop(old, None)
self._status_tokens[job_id] = (token, now)
else:
self._status_tokens.pop(job_id, None)
return data
@_trace_job
def status(self, job_id: str):
default_status = "Unknown"
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/",
params={"with_result": "false"},
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
status = response_data.get("status", default_status)
sub_status = response_data.get("sub_status")
if status == Job.RUNNING and sub_status is not None:
return sub_status
return status
@_trace_job
def stop(self, job_id: str, service: Optional[QiskitRuntimeService] = None):
if not service:
try:
service = QiskitRuntimeService(channel=self.channel, instance=self.instance, token=self.token)
except InvalidAccountError:
warnings.warn(
"No QiskitRuntimeService can be associated to the given token and instance. "
"Continuing without a QiskitRuntimeService."
)
service = None
data: dict[str, Any] = {
"service": json.dumps(service, cls=QiskitObjectsEncoder),
}
response_data = safe_json_request_as_dict(
request=lambda: requests.post(
f"{self.host}/api/{self.version}/jobs/{job_id}/stop/",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
json=data,
)
)
return response_data.get("message")
@_trace_job
def result(self, job_id: str) -> Dict[str, Any]:
gateway_url = f"{self.host}/api/{self.version}/jobs/{job_id}/result/"
response = requests.get(
gateway_url,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
if response.status_code == 204:
return {}
# Any non-OK response (from the gateway, COS, or an intermediary such as
# Cloudflare returning a block page) is surfaced with its status and body
# instead of being fed to the JSON parser as a cryptic decoding error.
raise_for_non_ok_response(response)
# Not all redirects go to COS — HTTP→HTTPS redirects stay on the same host.
# Checking the hostname detects only redirects to an external host (COS/MinIO).
redirected_to_cos = urlparse(response.url).hostname != urlparse(gateway_url).hostname
if redirected_to_cos:
return json.loads(response.text, cls=QiskitObjectsDecoder)
return json.loads(response.json().get("result", "{}") or "{}", cls=QiskitObjectsDecoder)
@_trace_job
def logs(self, job_id: str):
gateway_url = f"{self.host}/api/{self.version}/jobs/{job_id}/logs/"
response = requests.get(
gateway_url,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
if response.status_code == 204:
return "No logs yet."
# Any non-OK response (from the gateway, COS, or an intermediary such as
# Cloudflare returning a block page) is surfaced with its status and body
# instead of being fed to the JSON parser as a cryptic decoding error.
raise_for_non_ok_response(response)
# Not all redirects go to COS — HTTP→HTTPS redirects stay on the same host.
# Checking the hostname detects only redirects to an external host (COS/MinIO).
redirected_to_cos = urlparse(response.url).hostname != urlparse(gateway_url).hostname
if redirected_to_cos:
return response.text
return safe_json_request_as_dict(request=lambda: response).get("logs")
@_trace_job
def provider_logs(self, job_id: str):
gateway_url = f"{self.host}/api/{self.version}/jobs/{job_id}/provider-logs/"
response = requests.get(
gateway_url,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
if response.status_code == 204:
return "No logs yet."
# Any non-OK response (from the gateway, COS, or an intermediary such as
# Cloudflare returning a block page) is surfaced with its status and body
# instead of being fed to the JSON parser as a cryptic decoding error.
raise_for_non_ok_response(response)
# Not all redirects go to COS — HTTP→HTTPS redirects stay on the same host.
# Checking the hostname detects only redirects to an external host (COS/MinIO).
redirected_to_cos = urlparse(response.url).hostname != urlparse(gateway_url).hostname
if redirected_to_cos:
return response.text
return safe_json_request_as_dict(request=lambda: response).get("logs")
@_trace_job
def runtime_jobs(self, job_id: str, runtime_session: Optional[str] = None) -> list[str]:
"""Retrieve Qiskit IBM Runtime job ids that correspond to a
given serverless job_id execution and, optionally, filtered by session id."""
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/runtime_jobs/",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
if runtime_session:
return [
job.get("runtime_job")
for job in response_data.get("runtime_jobs", [])
if job.get("runtime_session") == runtime_session
]
return [job.get("runtime_job") for job in response_data.get("runtime_jobs", [])]
@_trace_job
def runtime_sessions(self, job_id: str):
"""Retrieve Qiskit IBM Runtime session ids that correspond to a
given serverless job_id execution."""
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/runtime_jobs/",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
runtime_jobs = response_data.get("runtime_jobs", [])
out_sessions = sorted({job["runtime_session"] for job in runtime_jobs if job.get("runtime_session")})
return out_sessions
def filtered_logs(self, job_id: str, **kwargs):
all_logs = self.logs(job_id=job_id)
included = ""
include = kwargs.get("include")
if include is not None:
for line in all_logs.split("\n"):
if re.search(include, line) is not None:
included = included + line + "\n"
else:
included = all_logs
excluded = ""
exclude = kwargs.get("exclude")
if exclude is not None:
for line in included.split("\n"):
if line != "" and re.search(exclude, line) is None:
excluded = excluded + line + "\n"
else:
excluded = included
return excluded
def events(self, job_id: str, **kwargs) -> list[JobEvent]:
"""Returns events of the job.
Args:
job_id: The job id
"""
response_data = safe_json_request_as_list(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/jobs/{job_id}/events/",
params=kwargs,
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
return [JobEvent.from_json(event) for event in response_data]
#########################
####### Functions #######
#########################
def upload(self, program: QiskitFunction) -> Optional[RunnableQiskitFunction]:
if program.runner == "ray":
warnings.warn(RAY_RUNNER_DEPRECATION, DeprecationWarning, stacklevel=2)
if program.dependencies:
warnings.warn(
"`dependencies=` is deprecated and no longer installs anything; it is accepted but "
"ignored and will be removed in a future release. Use "
"`client.dependencies_versions()` to see what the runtime image already ships.",
DeprecationWarning,
stacklevel=2,
)
tracer = trace.get_tracer("client.tracer")
with tracer.start_as_current_span("function.upload") as span:
span.set_attribute("function", program.title)
url = f"{self.host}/api/{self.version}/programs/upload/"
if program.image:
# upload function with custom image
function_uploaded = _upload_with_docker_image(
program=program,
url=url,
token=self.token,
span=span,
client=self,
instance=self.instance,
channel=self.channel,
)
elif program.entrypoint:
# upload function with artifact
function_uploaded = _upload_with_artifact(
program=program,
url=url,
token=self.token,
span=span,
client=self,
instance=self.instance,
channel=self.channel,
)
else:
raise QiskitServerlessException("Function must either have `entrypoint` or `image` specified.")
return function_uploaded
@_trace_functions("list")
def functions(self, provider: Optional[str] = None, **kwargs) -> List[RunnableQiskitFunction]:
"""Returns list of available functions.
Args:
provider: if given, only functions belonging to this provider are returned,
e.g. ``functions(provider="q-ctrl")``.
"""
params = dict(kwargs)
if provider:
params["provider"] = provider
response_data = safe_json_request_as_list(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/programs",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
params=params,
timeout=REQUESTS_TIMEOUT,
)
)
for program_data in response_data:
program_data["client"] = self
return [RunnableQiskitFunction.from_json(program_data) for program_data in response_data]
@_trace_functions("get_by_title")
def function(self, title: str, provider: Optional[str] = None) -> Optional[RunnableQiskitFunction]:
"""Returns program based on parameters."""
provider, title = format_provider_name_and_title(request_provider=provider, title=title)
response_data = safe_json_request_as_dict(
request=lambda: requests.get(
f"{self.host}/api/{self.version}/programs/get_by_title/{title}",
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
params={"provider": provider},
timeout=REQUESTS_TIMEOUT,
)
)
response_data["client"] = self
the_function = RunnableQiskitFunction.from_json(response_data)
return the_function
def validate_arguments(
self,
title: str,
arguments: Optional[Dict[str, Any]] = None,
provider: Optional[str] = None,
) -> dict:
"""Validate arguments against a function's schema without creating a job.
Args:
title: function title, optionally in "provider/title" format
arguments: arguments dict to validate
provider: optional provider name
Returns:
dict: response from the gateway, e.g. {"valid": True}
Raises:
QiskitServerlessException: if arguments are invalid or function not found.
"""
provider_name, function_title = format_provider_name_and_title(request_provider=provider, title=title)
response = safe_json_request_as_dict(
request=lambda: requests.post(
f"{self.host}/api/{self.version}/programs/validate_arguments/",
json={
"title": function_title,
"arguments": json.dumps(arguments or {}, cls=QiskitObjectsEncoder),
"provider": provider_name,
},
headers=get_headers(token=self.token, instance=self.instance, channel=self.channel),
timeout=REQUESTS_TIMEOUT,
)
)
return response
#####################
####### FILES #######
#####################
def files(self, function: QiskitFunction, sub_path: Optional[str] = None) -> List[str]:
"""Returns the list of files available for the user in the Qiskit Function folder.
``sub_path`` optionally lists a sub-directory of the function's folder (e.g. ``"wheels"``).
"""
return self._files_client.list(function, sub_path)
def provider_files(self, function: QiskitFunction) -> List[str]:
"""Returns the list of files available for the provider in the Qiskit Function folder."""
return self._files_client.provider_list(function)
def file_download(
self,
file: str,
function: QiskitFunction,
target_name: Optional[str] = None,
download_location: str = "./",
):
"""Download a file available to the user for the specific Qiskit Function."""
return self._files_client.download(file, download_location, function, target_name)
def provider_file_download(
self,
file: str,
function: QiskitFunction,
target_name: Optional[str] = None,
download_location: str = "./",
):
"""Download a file available to the provider for the specific Qiskit Function."""
return self._files_client.provider_download(file, download_location, function, target_name)
def file_delete(self, file: str, function: QiskitFunction, sub_path: Optional[str] = None):
"""Deletes a file available to the user for the specific Qiskit Function.
``sub_path`` optionally targets a sub-directory of the function's folder (e.g. ``"wheels"``).
"""
return self._files_client.delete(file, function, sub_path)
def provider_file_delete(self, file: str, function: QiskitFunction):
"""Deletes a file available to the provider for the specific Qiskit Function."""
return self._files_client.provider_delete(file, function)
def file_upload(self, file: str, function: QiskitFunction, sub_path: Optional[str] = None):
"""Uploads a file in the specific user's Qiskit Function folder.
``sub_path`` optionally stores the file in a sub-directory of the function's folder
(e.g. ``"wheels"``).
"""
return self._files_client.upload(file, function, sub_path)
def provider_file_upload(self, file: str, function: QiskitFunction):
"""Uploads a file in the specific provider's Qiskit Function folder."""
return self._files_client.provider_upload(file, function)
def add_wheels(self, function: QiskitFunction, patterns: List[str]) -> List[str]:
"""Upload locally-built wheels to a function's file area.
Each pattern is a local path or glob (e.g. ``"dist/mypkg-1.0-*.whl"``); every match
must be a ``.whl`` file. This is a thin wrapper over :meth:`file_upload` that expands
the patterns and uploads each wheel into the function's ``wheels`` sub-directory, so
they are available to the function at runtime. Install them from your entrypoint with
``install_uploaded_wheels()``, which reads that same directory. Use :meth:`list_wheels`
and :meth:`remove_wheels` to see and remove what has been uploaded.
Args:
function: the function whose file area the wheels are uploaded to.
patterns: local paths or globs matching the wheel files to upload.
Returns:
The uploaded wheel file names, in upload order.
Raises:
QiskitServerlessException: if a pattern matches no files, or a match is not a
``.whl`` file.
"""
paths: List[str] = []
for pattern in patterns:
matches = sorted(glob.glob(pattern))
if not matches:
raise QiskitServerlessException(f"No files match wheel pattern: {pattern}")
paths.extend(matches)
non_wheels = [p for p in paths if not p.endswith(".whl")]
if non_wheels:
raise QiskitServerlessException(f"add_wheels only accepts .whl files; got: {', '.join(non_wheels)}")
uploaded: List[str] = []
for path in paths:
self.file_upload(path, function, sub_path=WHEELS_SUBDIR)
uploaded.append(os.path.basename(path))
return uploaded
def list_wheels(self, function: QiskitFunction) -> List[str]:
"""List the wheels currently uploaded to a function (via :meth:`add_wheels`).
Returns the wheel file names in the function's ``wheels`` sub-directory.
"""
return self.files(function, sub_path=WHEELS_SUBDIR)
def remove_wheels(self, function: QiskitFunction, names: List[str]) -> None:
"""Remove previously uploaded wheels from a function.
To *update* a wheel, remove the old one and :meth:`add_wheels` the new build (a rebuild
with a different version has a different file name, so it would otherwise accumulate
alongside the old one and both would be installed).
Args:
function: the function to remove wheels from.
names: wheel file names to remove, as returned by :meth:`list_wheels`.
"""
for name in names:
self.file_delete(name, function, sub_path=WHEELS_SUBDIR)
def dependency_report(self, requirements: List[str]) -> "DependencyReport":
"""Diff your requirements against what the runtime image already ships.
For each requirement this reports whether it is already ``baked`` into the image (so you
should not bring it), needs to be brought as a wheel (``bring``), or ``clashes`` with a
different baked version. It is a client-side diff over :meth:`dependencies_versions`; it
uploads nothing. The result prints as an aligned table.
Args:
requirements: requirement strings, e.g. ``["numpy>=2", "mypkg", "jax<0.5"]``.
Returns:
A :class:`DependencyReport` (a list of :class:`DependencyReportEntry` that prints as
a table).
"""
baked = {}
for dep in self.dependencies_versions():
name, version = _split_requirement(dep)
baked[_normalize_dep_name(name)] = version
entries: List[DependencyReportEntry] = []
for requirement in requirements:
name, spec = _split_requirement(requirement)
baked_version = baked.get(_normalize_dep_name(name))
if baked_version is None:
status, note = "bring", "not in image — upload a wheel"
elif spec and not _spec_allows(spec, baked_version):
status, note = "clash", f"image ships {name}=={baked_version}"
else:
status, note = "baked", f"already in image ({name}=={baked_version})"
entries.append(DependencyReportEntry(requirement=requirement, status=status, note=note))
return DependencyReport(entries)
[docs]
class IBMServerlessClient(ServerlessClient):
"""
A client for connecting to the IBM serverless host.
Credentials can be saved to disk by calling the `save_account()` method::
from qiskit_serverless import IBMServerlessClient
IBMServerlessClient.save_account(token=<INSERT_IBM_QUANTUM_TOKEN>, instance=<INSERT_CRN>)
Once the credentials are saved, you can simply instantiate the client with no
constructor args, as shown below.
from qiskit_serverless import IBMServerlessClient
client = IBMServerlessClient()
Instead of saving credentials to disk, you can also set the environment variable
ENV_GATEWAY_PROVIDER_TOKEN and then instantiate the client as below::
from qiskit_serverless import IBMServerlessClient
client = IBMServerlessClient()
You can also enable an account just for the current session by instantiating the
provider with the API token::
from qiskit_serverless import IBMServerlessClient
client = IBMServerlessClient(token=<INSERT_IBM_QUANTUM_TOKEN>, instance=<INSERT_CRN>)
"""
[docs]
def __init__(
self,
token: Optional[str] = None,
name: Optional[str] = None,
instance: Optional[str] = None,
channel: Optional[str] = None,
*,
host: Optional[str] = None,
):
"""
Initialize a client with access to an IBMQ-provided remote cluster.
If a ``token`` is used to initialize an instance, the ``name`` argument
will be ignored.
If only a ``name`` is provided, the token for the named account will
be retrieved from the user's local IBM Quantum account config file.
If neither argument is provided, the token will be searched for in the
environment variables and also in the local IBM Quantum account config
file using the default account name.
Args:
host: host of gateway. Optional. It uses IBM_SERVERLESS_HOST_URL env var or IBM host
token: IBM quantum token
name: Name of the account to load
instance: IBM Cloud CRN
channel: identifies the method to use to authenticate the user
"""
channel = channel or Channel.IBM_QUANTUM_PLATFORM.value # For backwards compatibility
# Initialize QiskitRuntimeService
self._service = QiskitRuntimeService(channel=channel, token=token, name=name, instance=instance)
self.account = self._service._account
# Per-instance cache keyed by backend name; populated lazily by backends() or _get_backend().
# Instance-level (not class-level) to avoid cross-client leakage.
self._backends_cache: Dict[str, Any] = {}
super().__init__(
channel=self.account.channel,
token=self.account.token,
instance=self.account.instance,
host=host if host else IBM_SERVERLESS_HOST_URL,
)
@staticmethod
def save_account(
token: Optional[str] = None,
name: Optional[str] = None,
overwrite: Optional[bool] = False,
instance: Optional[str] = None,
channel: Optional[str] = None,
) -> None:
"""
Save the account to disk for future use.
Args:
token: IBM Quantum API token
name: Name of the account to save
overwrite: ``True`` if the existing account is to be overwritten
instance: IBM Cloud CRN
channel: identifies the method to use to authenticate the user
"""
try:
QiskitRuntimeService.save_account(
token=token,
name=name,
overwrite=overwrite,
instance=instance,
channel=channel,
)
except InvalidAccountError as ex:
raise QiskitServerlessException(f"Invalid format in account inputs - {ex}") from ex
def usage(self) -> dict[str, Any]:
"""Return runtime usage information for the active instance.
Exposes the underlying :meth:`QiskitRuntimeService.usage` method to retrieve
the instance's runtime quota information, including ``usage_remaining_seconds``
and ``usage_limit_reached``.
Returns:
A dictionary containing usage information as reported by the runtime service.
Raises:
QiskitServerlessException: If the usage information cannot be retrieved.
"""
try:
return self._service.usage()
except Exception as exc: # pylint: disable=broad-except
raise QiskitServerlessException(
f"Failed to retrieve usage information for instance '{self.instance}': {exc}"
) from exc
def backends( # pylint: disable=too-many-positional-arguments
self,
refresh_cache: bool = False,
name: str | None = None,
min_num_qubits: int | None = None,
filters: Callable[[IBMBackend], bool] | None = None,
**kwargs: Any,
) -> list[IBMBackend]:
"""Return backends accessible through this instance.
Exposes the underlying :meth:`QiskitRuntimeService.backends` method with
per-instance caching. Results are cached after the first call; use
``refresh_cache=True`` to force a refresh.
Args:
refresh_cache: If ``True``, refresh the cache by fetching backends from the service.
name: Backend name to filter by.
min_num_qubits: Minimum number of qubits the backend must have.
filters: More complex filters, such as lambda functions.
For example::
client.backends(filters=lambda b: b.max_shots > 50000)
client.backends(filters=lambda x: "rz" in x.basis_gates)
**kwargs: Simple filters that require a specific value for an attribute in
backend configuration or status.
Examples::
# Get backends with at least 127 qubits
client.backends(min_num_qubits=127)
For the full list of backend attributes, see the `IBMBackend class documentation
<https://quantum.cloud.ibm.com/docs/api/qiskit-ibm-runtime/qiskit-runtime-service#backends>`_
Returns:
List of available backends that match the filter criteria.
Raises:
QiskitServerlessException: If the backend listing call fails.
"""
if not refresh_cache and self._backends_cache:
# cache is available and refresh flag is false
return list(self._backends_cache.values())
try:
backend_list = self._service.backends(
name=name,
min_num_qubits=min_num_qubits,
filters=filters,
**kwargs,
)
except Exception as exc:
raise QiskitServerlessException(
f"Failed to retrieve backends for instance '{self.instance}': {exc}"
) from exc
# deleting cached backends as they could have unavailable backends and update with new accessible backends
self._backends_cache = {}
for backend in backend_list:
self._backends_cache[backend.name] = backend
return backend_list
def backend(self, name: str, **kwargs: Any) -> Backend:
"""Fetch a single backend by name and update the cache.
Exposes the underlying :meth:`QiskitRuntimeService.backend` method to perform
a targeted backend lookup. This is more efficient than listing all backends
and validates that the caller has access to the requested backend. The cache
is updated on every call to reflect current access permissions.
Args:
name: Name of the backend (e.g., ``"ibm_torino"``).
**kwargs: Simple filters that require a specific value for an attribute in
backend configuration or status.
For the full list of backend attributes, see the `IBMBackend class documentation
<https://quantum.cloud.ibm.com/docs/api/qiskit-ibm-runtime/qiskit-runtime-service#backend>`_
Returns:
Backend matching the specified name.
Raises:
QiskitServerlessException: If the backend is not found or is inaccessible.
"""
try:
backend = self._service.backend(name=name, **kwargs)
except QiskitBackendNotFoundError as exc:
raise QiskitServerlessException(
f"Backend '{name}' is not available or you do not have access to it "
f"with instance '{self.instance}'. "
f"Call client.backends() to list accessible backends."
) from exc
except Exception as exc:
raise QiskitServerlessException(f"Failed to retrieve backend '{name}': {exc}") from exc
self._backends_cache[name] = backend
return backend
def least_busy(
self,
min_num_qubits: int | None = None,
filters: Callable[[IBMBackend], bool] | None = None,
**kwargs: Any,
) -> IBMBackend:
"""Return the least busy backend matching the specified criteria.
Exposes the underlying :meth:`QiskitRuntimeService.least_busy` method to find
the backend with the fewest pending jobs. The result is cached in the instance.
Args:
min_num_qubits: Minimum number of qubits the backend must have.
filters: Filters can be defined as for the :meth:`backends` method.
Example::
client.least_busy(min_num_qubits=5, operational=True)
**kwargs: Additional filters for backend configuration or status attributes.
Returns:
The backend with the fewest number of pending jobs that matches the criteria.
Raises:
QiskitServerlessException: If no backend matches the criteria or the call fails.
"""
try:
backend = self._service.least_busy(
min_num_qubits=min_num_qubits,
filters=filters,
**kwargs,
)
except QiskitBackendNotFoundError as exc:
raise QiskitServerlessException(
f"No available backend matches the criteria with instance '{self.instance}'. "
f"Call client.backends() to list accessible backends."
) from exc
except Exception as exc:
raise QiskitServerlessException(f"Failed to retrieve backend: {exc}") from exc
self._backends_cache[backend.name] = backend
return backend
def _check_usage(self, suppress_low_usage_warning: bool = False) -> None:
"""Check instance runtime quota and warn or raise if insufficient.
Retrieves ``usage_remaining_seconds`` from :meth:`usage`. If the key is absent,
the plan is unlimited and no action is taken. Transient errors from the usage
endpoint are downgraded to a warning to avoid blocking job submission.
Thresholds are configurable via environment variables (see constants.py):
- ``USAGE_ZERO_EPSILON_SECONDS`` (default 1 s): Raises an exception when
remaining time is at or below this threshold.
- ``USAGE_LOW_THRESHOLD_SECONDS`` (default 600 s): Emits a warning when
remaining time is below this threshold (unless suppressed).
Args:
suppress_low_usage_warning: If ``True``, suppress the warning when remaining
quota is below ``USAGE_LOW_THRESHOLD_SECONDS``. The exception for
exhausted quota (at or below ``USAGE_ZERO_EPSILON_SECONDS``) is still raised.
Raises:
QiskitServerlessException: When remaining quota is at or below
``USAGE_ZERO_EPSILON_SECONDS``, or when ``usage_limit_reached`` is
``True`` with no remaining time recorded.
"""
try:
usage = self.usage()
except Exception as exc: # pylint: disable=broad-except
warnings.warn(
f"Could not retrieve usage information for instance '{self.instance}': {exc}. "
"Proceeding with job submission; verify your instance quota manually."
)
return
remaining = usage.get("usage_remaining_seconds")
limit_reached = usage.get("usage_limit_reached", False)
if remaining is None and not limit_reached:
return
# usage_limit_reached without a recorded remaining value → treat as zero.
if limit_reached and remaining is None:
remaining = 0.0
if remaining is not None and remaining <= USAGE_ZERO_EPSILON_SECONDS:
raise QiskitServerlessException(
f"Instance '{self.instance}' has no remaining runtime quota "
f"(remaining: {remaining:.2f}s). "
"The job would wait indefinitely for runtime resources. "
"Check your instance quota at https://quantum.cloud.ibm.com/instances."
)
if not suppress_low_usage_warning and remaining is not None and remaining <= USAGE_LOW_THRESHOLD_SECONDS:
warnings.warn(
f"Instance '{self.instance}' has low remaining runtime quota "
f"({remaining:.0f}s remaining, threshold: {USAGE_LOW_THRESHOLD_SECONDS:.0f}s). "
"Consider checking your quota before submitting long-running jobs. "
"Check https://quantum.cloud.ibm.com/instances for details."
)
def run( # pylint: disable=too-many-positional-arguments
self,
program: Union[QiskitFunction, str],
arguments: Optional[Dict[str, Any]] = None,
config: Optional[Configuration] = None,
provider: Optional[str] = None,
*,
function_size: Optional[str] = None,
suppress_usage_warning: bool = False,
bypass_usage_check: bool = False,
) -> "Job":
"""Run a Qiskit Function with pre-flight validation before submitting to the gateway.
Checks are performed before delegating to ``ServerlessClient.run``. Checks include instance has no
remaining runtime; warns when low and check accessibility to the backed
Args:
program: ``QiskitFunction`` object or function title string.
arguments: Runtime arguments. The optional ``backend_name`` key is inspected for
the backend access check but forwarded unchanged.
config: Optional execution configuration.
provider: Optional provider name override.
function_size: Declared size label (e.g. ``"m"``) to run at; resolves through the
function's size catalog on the gateway.
suppress_usage_warning: If ``True``, suppress the warning when remaining runtime
quota is below ``USAGE_LOW_THRESHOLD_SECONDS`` and the warning emitted when
``bypass_usage_check`` is ``True``. The warning emitted when usage information
cannot be retrieved is still shown, and the exception for exhausted quota
(at or below ``USAGE_ZERO_EPSILON_SECONDS``) is still raised.
bypass_usage_check: If ``True``, skip the client-side runtime quota check entirely, including
the exhausted-quota exception and the low-quota warning. A warning is emitted instead,
unless ``suppress_usage_warning`` is ``True``. The gateway still enforces its own quota checks.
Returns:
:class:`~qiskit_serverless.core.job.Job` handle for the submitted run.
Raises:
QiskitServerlessException: If quota is exhausted, the backend is inaccessible, or
the instance has no available backends.
"""
backend_name = (arguments or {}).get("backend_name")
if not bypass_usage_check:
self._check_usage(suppress_low_usage_warning=suppress_usage_warning)
elif not suppress_usage_warning:
warnings.warn(
f"bypass_usage_check=True: skipping the runtime quota check for instance '{self.instance}'. "
"The job will be submitted even if the remaining quota is insufficient to complete it, "
"and usage may still be billed. "
"Pass suppress_usage_warning=True to silence this warning."
)
if backend_name:
# Single-backend lookup — fast and confirms access without a full instance listing.
self.backend(backend_name)
else:
# No specific backend requested; verify the instance exposes at least one.
if not self.backends():
raise QiskitServerlessException(
f"Instance '{self.instance}' has no backends available. "
"Cannot run a Qiskit Function without an accessible backend. "
"Check your instance configuration at https://quantum.cloud.ibm.com/instances."
)
return super().run(
program=program,
arguments=arguments,
config=config,
provider=provider,
function_size=function_size,
)
def _sizes_payload(program: QiskitFunction) -> Dict[str, Any]:
"""Build the size-catalog fields of an upload request.
``sizes_map`` carries the ``{size_label: compute_profile}`` catalog on the
function; it is sent back to the gateway under the key it expects, ``sizes``,
as a JSON object, and ``default_size`` as a plain string. Each is omitted when
unset so an upload that does not touch sizes leaves any stored catalog
untouched (the gateway treats an absent field as "not sent") rather than
clearing it.
A ``sizes_map`` value may be a ``compute_profile_id`` string or a ``ComputeProfile``
object; ``str()`` serializes either to the same id, so the wire format is identical.
"""
payload: Dict[str, Any] = {}
if program.sizes_map is not None:
payload["sizes"] = json.dumps({label: str(profile) for label, profile in program.sizes_map.items()})
if program.default_size is not None:
payload["default_size"] = program.default_size
return payload
def _upload_with_docker_image( # pylint: disable=too-many-positional-arguments
program: QiskitFunction,
url: str,
token: str,
span: Any,
client: RunService,
instance: Optional[str],
channel: Optional[str],
) -> RunnableQiskitFunction:
"""Uploads function with custom docker image.
Args:
program (QiskitFunction): function instance
url (str): upload gateway url
token (str): auth token
span (Any): tracing span
instance (Optional[str]): IBM Cloud crn
Returns:
str: uploaded function name
"""
response_data = safe_json_request_as_dict(
request=lambda: requests.post(
url=url,
data={
"title": program.title,
"provider": program.provider,
"image": program.image,
"runner": program.runner,
"arguments": json.dumps({}),
"dependencies": json.dumps(program.dependencies or []),
"env_vars": json.dumps(program.env_vars or {}),
"description": program.description,
"version": program.version,
**(
{"arguments_schema": json.dumps(program.arguments_schema)}
if program.arguments_schema is not None
else {}
),
**_sizes_payload(program),
},
headers=get_headers(token=token, instance=instance, channel=channel),
timeout=REQUESTS_TIMEOUT,
)
)
program_title = response_data.get("title", "na")
program_provider = response_data.get("provider", "na")
span.set_attribute("function.title", program_title)
span.set_attribute("function.provider", program_provider)
response_data["client"] = client
return RunnableQiskitFunction.from_json(response_data)
def _upload_with_artifact( # pylint: disable=too-many-positional-arguments, too-many-locals
program: QiskitFunction,
url: str,
token: str,
span: Any,
client: RunService,
instance: Optional[str],
channel: Optional[str],
) -> RunnableQiskitFunction:
"""Uploads function with artifact.
Args:
program (QiskitFunction): function instance
url (str): endpoint for gateway upload
token (str): auth token
span (Any): tracing span
instance (Optional[str]): IBM Cloud crn
Raises:
QiskitServerlessException: if no entrypoint or size of artifact is too large.
Returns:
str: uploaded function name
"""
artifact_file_path = os.path.join(program.working_dir, "artifact.tar")
# check if entrypoint exists
if not os.path.exists(os.path.join(program.working_dir, program.entrypoint)) or program.entrypoint[0] == "/":
raise QiskitServerlessException(
f"Entrypoint file [{program.entrypoint}] does not exist " f"in [{program.working_dir}] working directory."
)
try:
with tarfile.open(artifact_file_path, "w", dereference=True) as tar:
for filename in os.listdir(program.working_dir):
fpath = os.path.join(program.working_dir, filename)
tar.add(fpath, arcname=filename)
# check file size
size_in_mb = Path(artifact_file_path).stat().st_size / 1024**2
if size_in_mb > MAX_ARTIFACT_FILE_SIZE_MB:
raise QiskitServerlessException(
f"{artifact_file_path} is {int(size_in_mb)} Mb, "
f"which is greater than {MAX_ARTIFACT_FILE_SIZE_MB} allowed. "
f"Try to reduce size of `working_dir`."
)
with open(artifact_file_path, "rb") as file:
response_data = safe_json_request_as_dict(
request=lambda: requests.post(
url=url,
data={
"title": program.title,
"provider": program.provider,
"entrypoint": program.entrypoint,
"runner": program.runner,
"arguments": json.dumps({}),
"dependencies": json.dumps(program.dependencies or []),
"env_vars": json.dumps(program.env_vars or {}),
"description": program.description,
"version": program.version,
**(
{"arguments_schema": json.dumps(program.arguments_schema)}
if program.arguments_schema is not None
else {}
),
**_sizes_payload(program),
},
files={"artifact": file},
headers=get_headers(token=token, instance=instance, channel=channel),
timeout=REQUESTS_TIMEOUT,
)
)
span.set_attribute("function.title", response_data.get("title", "na"))
span.set_attribute("function.provider", response_data.get("provider", "na"))
response_data["client"] = client
response_function = RunnableQiskitFunction.from_json(response_data)
except QiskitServerlessException as error:
raise error
except Exception as error: # pylint: disable=broad-exception-caught
raise QiskitServerlessException from error
finally:
if os.path.exists(artifact_file_path):
os.remove(artifact_file_path)
return response_function