# 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.function`)
=================================================
.. currentmodule:: qiskit_serverless.core.function
Qiskit Serverless function
==========================
.. autosummary::
:toctree: ../stubs/
QiskitFunction
"""
from abc import ABC, abstractmethod
import dataclasses
import warnings
from dataclasses import dataclass
from typing import Optional, Dict, List, Any, Tuple, Union
from qiskit_serverless.core.job import (
Job,
Configuration,
)
[docs]@dataclass
class QiskitFunction: # pylint: disable=too-many-instance-attributes
"""Serverless QiskitPattern.
Args:
title: program name
provider: Qiskit Function provider reference
entrypoint: is a script that will be executed as a job
ex: job.py
env_vars: env vars
dependencies: list of python dependencies to execute a program
working_dir: directory where entrypoint file is located (max size 50MB)
description: description of a program
version: version of a program
"""
title: str
provider: Optional[str] = None
entrypoint: Optional[str] = None
working_dir: Optional[str] = "./"
env_vars: Optional[Dict[str, str]] = None
dependencies: Optional[List[str]] = None
description: Optional[str] = None
version: Optional[str] = None
tags: Optional[List[str]] = None
raw_data: Optional[Dict[str, Any]] = None
image: Optional[str] = None
validate: bool = True
schema: Optional[str] = None
def __post_init__(self):
title_has_provider = "/" in self.title
if title_has_provider:
title_split = self.title.split("/")
if len(title_split) > 2:
raise ValueError("Invalid title: it can only contain one slash.")
if self.provider != title_split[0] and self.provider is not None:
raise ValueError(
"Invalid provider: you provided two different "
+ f"providers [{self.provider}] and [{title_split[0]}]."
)
self.provider = title_split[0]
self.title = title_split[1]
@classmethod
def from_json(cls, data: Dict[str, Any]):
"""Reconstructs QiskitPattern from dictionary."""
field_names = set(f.name for f in dataclasses.fields(QiskitFunction))
return QiskitFunction(**{k: v for k, v in data.items() if k in field_names})
def __str__(self):
if self.provider is not None:
return f"QiskitFunction({self.provider}/{self.title})"
return f"QiskitFunction({self.title})"
def __repr__(self):
return self.__str__()
def _validate_function(self) -> Tuple[bool, List[str]]:
"""Validate function arguments using schema provided.
Returns:
Tuple[bool, List[str]]:
boolean specifiying if function arguments are valid
list of validation errors, if any
"""
return True, []
class RunService(ABC):
"""Provide access to run a function and retrieve the jobs associated to that function"""
@abstractmethod
def jobs(self, **kwargs) -> List[Job]:
"""Return list of jobs.
Returns:
list of jobs.
"""
@abstractmethod
def run(
self,
program: Union[QiskitFunction, str],
arguments: Optional[Dict[str, Any]] = None,
config: Optional[Configuration] = None,
) -> Job:
"""Run a function and return its job."""
class RunnableQiskitFunction(QiskitFunction):
"""Serverless QiskitPattern.
Args:
title: program name
provider: Qiskit Function provider reference
entrypoint: is a script that will be executed as a job
ex: job.py
env_vars: env vars
dependencies: list of python dependencies to execute a program
working_dir: directory where entrypoint file is located (max size 50MB)
description: description of a program
version: version of a program
"""
_run_service: RunService = None
def __init__( # pylint: disable=too-many-positional-arguments
self, client: RunService, **kwargs
):
self._run_service = client
super().__init__(**kwargs)
@classmethod
def from_json(cls, data: Dict[str, Any]):
"""Reconstructs QiskitPattern from dictionary."""
field_names = set(f.name for f in dataclasses.fields(QiskitFunction))
client = data["client"]
return RunnableQiskitFunction(
client, **{k: v for k, v in data.items() if k in field_names}
)
def run(self, **kwargs):
"""Run function
Raises:
QiskitServerlessException: validation exception
Returns:
Job: job handler for function execution
"""
if self._run_service is None:
raise ValueError("No clients specified for a function.")
if self.validate:
is_valid, validation_errors = self._validate_function()
if not is_valid:
error_string = "\n".join(validation_errors)
raise ValueError(
f"Function validation failed. Validation errors:\n {error_string}",
)
config = kwargs.pop("config", None)
return self._run_service.run(
program=self,
arguments=kwargs,
config=config,
)
def get_jobs(self):
# pylint: disable=duplicate-code
"""List of jobs created in this function.
Raises:
QiskitServerlessException: validation exception
Returns:
[Job] : list of jobs
"""
warnings.warn(
"`get_jobs` method has been deprecated. "
"And will be removed in future releases. "
"Please, use `jobs` instead.",
DeprecationWarning,
)
return self.jobs()
def jobs(self):
"""List of jobs created in this function.
Raises:
QiskitServerlessException: validation exception
Returns:
[Job] : list of jobs
"""
if self._run_service is None:
raise ValueError("No clients specified for a function.")
if self.validate:
is_valid, validation_errors = self._validate_function()
if not is_valid:
error_string = "\n".join(validation_errors)
raise ValueError(
f"Function validation failed. Validation errors:\n {error_string}",
)
jobs = self._run_service.jobs(
title=self.title,
provider=self.provider,
)
return jobs
# pylint: disable=abstract-method
# pylint: disable=too-few-public-methods
[docs]class QiskitPattern(QiskitFunction):
"""
[Deprecated since version 0.10.0] Use :class:`.QiskitFunction` instead.
A provider for connecting to a ray head node. This class has been
renamed to :class:`.QiskitFunction`.
"""