API documentation of storage providers

The module that contains all the necessary logic for communication with the external storage for the jobs. It creates an abstract API layer for the storage providers.

StorageProvider

Bases: ABC

The template for accessing any storage providers like dropbox, mongodb, amazon S3 etc.

__init__(name, is_active=True)

Any storage provide must have a name that is not empty.

Parameters:

Name	Type	Description	Default
name	str	The name of the storage provider	required
is_active	bool	Is the storage provider active.	True

backend_dict_to_qiskit(backend_config_info)

This function transforms the dictionary that is safed in the storage provider into a qiskit backend dictionnary.

Parameters:

Name	Type	Description	Default
backend_config_info	BackendConfigSchemaIn	The dictionary that contains the configuration of the backend	required

Returns:

Type	Description
BackendConfigSchemaOut	The qiskit backend dictionary

backend_dict_to_qiskit_status(backend_dict)

This function transforms the dictionary that is safed in the storage provider into a qiskit backend status dictionnary.

Parameters:

Name	Type	Description	Default
backend_dict	BackendConfigSchemaIn	The dictionary that contains the configuration of the backend	required

Returns:

Type	Description
BackendStatusSchemaOut	The qiskit backend dictionary

delete_file(storage_path, job_id) abstractmethod

Delete the file from the storage

get_backend_dict(display_name)

The configuration dictionary of the backend such that it can be sent out to the API to the common user. We make sure that it is compatible with QISKIT within this function.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The identifier of the backend	required

Returns:

Type	Description
BackendConfigSchemaOut	The full schema of the backend.

Raises:

Type	Description
FileNotFoundError	If the backend does not exist

get_backend_status(display_name) abstractmethod

Get the status of the backend. This follows the qiskit logic.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend	required

Returns:

Type	Description
BackendStatusSchemaOut	The status dict of the backend

get_backends() abstractmethod

Get a list of all the backends that the provider offers.

get_config(display_name) abstractmethod

The function that downloads the spooler configuration to the storage.

Parameters:

Name	Type	Description	Default
display_name		The name of the backend	required

Returns:

Type	Description
BackendConfigSchemaIn	The configuration of the backend in complete form.

get_file_content(storage_path, job_id) abstractmethod

Get the file content from the storage.

Parameters:

Name	Type	Description	Default
storage_path	str	The path to the file	required
job_id	str	The id of the job	required

Returns:

Type	Description
dict	The content of the file

Raises:

Type	Description
FileNotFoundError	If the file is not found

get_file_queue(storage_path) abstractmethod

Get a list of files

Parameters:

Name	Type	Description	Default
storage_path	str	Where are we looking for the files.	required

Returns:

Type	Description
list[str]	A list of files that was found.

get_job_content(storage_path, job_id) abstractmethod

Get the content of the job from the storage. This is a wrapper around get_file_content and and handles the different ways of identifiying the job.

storage_path: the path towards the file, excluding the filename / id job_id: the id of the file we are about to look up

Returns:

Type	Description
dict	The content of the job

get_next_job_in_queue(display_name) abstractmethod

A function that obtains the next job in the queue. If there is no job, it returns an empty dict. If there is a job, it moves the job from the queue to the running folder. It also update the time stamp for when the system last looked into the file queue.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend	required

Returns:

Type	Description
NextJobSchema	the job dict

get_result(display_name, username, job_id) abstractmethod

This function gets the result file from the backend and returns the result dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
ResultDict	The result dict of the job. If the information is not available, the result dict
ResultDict	has a status of "ERROR".

get_status(display_name, username, job_id) abstractmethod

This function gets the status file from the backend and returns the status dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
StatusMsgDict	The status dict of the job

long_backend_name(display_name, simulator)

This function returns the long name of the backend.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend	required
simulator	bool	True if the backend is a simulator	required

Returns:

Type	Description
BackendNameStr	The long name of the backend

move_file(start_path, final_path, job_id) abstractmethod

Move the file from start_path to final_path

timestamp_queue(display_name)

Updates the time stamp for when the system last looked into the file queue. This allows us to track if the system is actually online or not.

Parameters:

Name	Type	Description	Default
display_name		The name of the backend	required

Returns:

Type	Description
None	None

update_file(content_dict, storage_path, job_id) abstractmethod

Update the file content. It replaces the old content with the new content.

Parameters:

Name	Type	Description	Default
content_dict	dict	The dictionary containing the new content of the file	required
storage_path	str	The path to the file	required
job_id	str	The id of the job	required

Returns:

Type	Description
None	None

Raises:

Type	Description
FileNotFoundError	If the file is not found

update_in_database(result_dict, status_msg_dict, job_id, display_name) abstractmethod

Upload the status and result to the StorageProvider.

Parameters:

Name	Type	Description	Default
result_dict	ResultDict	the dictionary containing the result of the job	required
status_msg_dict	StatusMsgDict	the dictionary containing the status message of the job	required
job_id	str	the name of the job	required
display_name	DisplayNameStr	the name of the backend	required

Returns:

Type	Description
None	None

upload(content_dict, storage_path, job_id) abstractmethod

Upload the file to the storage.

Parameters:

Name	Type	Description	Default
content_dict	Mapping	The dictionary containing the content of the file	required
storage_path	str	The path to the file	required
job_id	str	The id of the job	required

upload_config(config_dict, display_name) abstractmethod

The function that uploads the spooler configuration to the storage.

Parameters:

Name	Type	Description	Default
config_dict	BackendConfigSchemaIn	The model containing the configuration	required
display_name		The name of the backend	required

Returns:

Type	Description
None	None

upload_job(job_dict, display_name, username) abstractmethod

Upload the job to the storage provider.

Parameters:

Name	Type	Description	Default
job_dict	dict	the full job dict	required
display_name	DisplayNameStr	the name of the backend	required
username	str	the name of the user that submitted the job	required

Returns:

Type	Description
str	The job id of the uploaded job.

upload_status(display_name, username, job_id) abstractmethod

This function uploads a status file to the backend and creates the status dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
StatusMsgDict	The status dict of the job

validate_active(func)

Decorator to check if the storage provider is active.

The module that contains all the necessary logic for communication with the MongoDb storage providers.

MongodbProvider

Bases: MongodbProviderExtended

The access to the mongodb. This is the simplified version for people that are running devices.

Source code in src/sqooler/storage_providers/mongodb.py

class MongodbProvider(MongodbProviderExtended):
    """
    The access to the mongodb. This is the simplified version for people that are running devices.
    """

    def __init__(self, login_dict: MongodbLoginInformation) -> None:
        """
        Set up the neccessary keys and create the client through which all the connections will run.
        """
        super().__init__(login_dict, name="default", is_active=True)

__init__(login_dict)

Set up the neccessary keys and create the client through which all the connections will run.

Source code in src/sqooler/storage_providers/mongodb.py

def __init__(self, login_dict: MongodbLoginInformation) -> None:
    """
    Set up the neccessary keys and create the client through which all the connections will run.
    """
    super().__init__(login_dict, name="default", is_active=True)

MongodbProviderExtended

Bases: StorageProvider

The access to the mongodb

Source code in src/sqooler/storage_providers/mongodb.py

class MongodbProviderExtended(StorageProvider):
    """
    The access to the mongodb
    """

    def __init__(
        self, login_dict: MongodbLoginInformation, name: str, is_active: bool = True
    ) -> None:
        """
        Set up the neccessary keys and create the client through which all the connections will run.

        Args:
            login_dict: The login dict that contains the neccessary
                        information to connect to the mongodb
            name: The name of the storage provider
            is_active: Is the storage provider active.


        Raises:
            ValidationError: If the login_dict is not valid
        """
        super().__init__(name, is_active)
        mongodb_username = login_dict.mongodb_username
        mongodb_password = login_dict.mongodb_password
        mongodb_database_url = login_dict.mongodb_database_url

        uri = f"mongodb+srv://{mongodb_username}:{mongodb_password}@{mongodb_database_url}"
        uri = uri + "/?retryWrites=true&w=majority"
        # Create a new client and connect to the server
        self.client: MongoClient = MongoClient(uri)

        # Send a ping to confirm a successful connection
        self.client.admin.command("ping")

    @validate_active
    def upload(self, content_dict: dict, storage_path: str, job_id: str) -> None:
        """
        Upload the file to the storage

        content_dict: the content that should be uploaded onto the mongodb base
        storage_path: the access path towards the mongodb collection
        job_id: the id of the file we are about to create
        """
        storage_splitted = storage_path.split("/")

        # get the database on which we work
        database = self.client[storage_splitted[0]]

        # get the collection on which we work
        collection_name = ".".join(storage_splitted[1:])
        collection = database[collection_name]

        content_dict["_id"] = ObjectId(job_id)
        collection.insert_one(content_dict)

        # remove the id from the content dict for further use
        content_dict.pop("_id", None)

    @validate_active
    def get_file_content(self, storage_path: str, job_id: str) -> dict:
        """
        Get the file content from the storage

        Args:
            storage_path: the path towards the file, excluding the filename / id
            job_id: the id of the file we are about to look up

        Returns:
            The content of the file
        """
        try:
            document_to_find = {"_id": ObjectId(job_id)}
        except InvalidId as err:
            raise FileNotFoundError(
                f"The job_id {job_id} is not valid. Please check the job_id."
            ) from err

        document_to_find = {"_id": ObjectId(job_id)}

        # get the database on which we work
        database = self.client[storage_path.split("/")[0]]

        # get the collection on which we work
        collection_name = ".".join(storage_path.split("/")[1:])
        collection = database[collection_name]

        result_found = collection.find_one(document_to_find)

        if not result_found:
            raise FileNotFoundError(
                f"Could not find a file under {storage_path} with the id {job_id}."
            )

        # remove the id from the result dict for further use
        result_found.pop("_id", None)
        return result_found

    def get_job_content(self, storage_path: str, job_id: str) -> dict:
        """
        Get the content of the job from the storage. This is a wrapper around get_file_content
        and and handles the different ways of identifiying the job.

        storage_path: the path towards the file, excluding the filename / id
        job_id: the id of the file we are about to look up

        Returns:

        """
        job_dict = self.get_file_content(storage_path=storage_path, job_id=job_id)
        job_dict.pop("_id", None)
        return job_dict

    def update_file(self, content_dict: dict, storage_path: str, job_id: str) -> None:
        """
        Update the file content. It replaces the old content with the new content.


        Args:
            content_dict: The dictionary containing the new content of the file
            storage_path: The path to the file
            job_id: The id of the job

        Returns:
            None

        Raises:
            FileNotFoundError: If the file is not found
        """
        # get the database on which we work
        database = self.client[storage_path.split("/")[0]]

        # get the collection on which we work
        collection_name = ".".join(storage_path.split("/")[1:])
        collection = database[collection_name]

        filter_dict = {"_id": ObjectId(job_id)}
        result = collection.replace_one(filter_dict, content_dict)

        if result.matched_count == 0:
            raise FileNotFoundError(f"Could not update file under {storage_path}")

    @validate_active
    def move_file(self, start_path: str, final_path: str, job_id: str) -> None:
        """
        Move the file from start_path to final_path

        start_path: the path where the file is currently stored, but excluding the file name
        final_path: the path where the file should be stored, but excluding the file name
        job_id: the name of the file. Is a json file

        Returns:
            None
        """
        # get the database on which we work
        database = self.client[start_path.split("/")[0]]

        # get the collection on which we work
        collection_name = ".".join(start_path.split("/")[1:])
        collection = database[collection_name]

        document_to_find = {"_id": ObjectId(job_id)}
        result_found = collection.find_one(document_to_find)

        # delete the old file
        collection.delete_one(document_to_find)

        # add the document to the new collection
        database = self.client[final_path.split("/")[0]]
        collection_name = ".".join(final_path.split("/")[1:])
        collection = database[collection_name]
        collection.insert_one(result_found)

    @validate_active
    def delete_file(self, storage_path: str, job_id: str) -> None:
        """
        Remove the file from the mongodb database

        Args:
            storage_path: the path where the file is currently stored, but excluding the file name
            job_id: the name of the file

        Returns:
            None
        """
        # get the database on which we work
        database = self.client[storage_path.split("/")[0]]

        # get the collection on which we work
        collection_name = ".".join(storage_path.split("/")[1:])
        collection = database[collection_name]

        document_to_find = {"_id": ObjectId(job_id)}
        collection.delete_one(document_to_find)

    @validate_active
    def get_backends(self) -> list[DisplayNameStr]:
        """
        Get a list of all the backends that the provider offers.
        """

        # get the database on which we work
        database = self.client["backends"]
        config_collection = database["configs"]
        # get all the documents in the collection configs and save the disply_name in a list
        backend_names: list[DisplayNameStr] = []
        for config_dict in config_collection.find():
            backend_names.append(config_dict["display_name"])
        return backend_names

    def get_backend_status(
        self, display_name: DisplayNameStr
    ) -> BackendStatusSchemaOut:
        """
        Get the status of the backend. This follows the qiskit logic.

        Args:
            display_name: The name of the backend

        Returns:
            The status dict of the backend

        Raises:
            FileNotFoundError: If the backend does not exist
        """
        # get the database on which we work
        database = self.client["backends"]
        config_collection = database["configs"]

        # create the filter for the document with display_name that is equal to display_name
        document_to_find = {"display_name": display_name}
        backend_config_dict = config_collection.find_one(document_to_find)

        if not backend_config_dict:
            raise FileNotFoundError(
                f"The backend {display_name} does not exist for the given storageprovider."
            )

        backend_config_dict.pop("_id")
        backend_config_info = BackendConfigSchemaIn(**backend_config_dict)
        qiskit_backend_dict = self.backend_dict_to_qiskit_status(backend_config_info)
        return qiskit_backend_dict

    def upload_config(
        self, config_dict: BackendConfigSchemaIn, display_name: DisplayNameStr
    ) -> None:
        """
        The function that uploads the spooler configuration to the storage.

        Args:
            config_dict: The dictionary containing the configuration
            display_name : The name of the backend

        Returns:
            None
        """
        config_path = "backends/configs"

        # first we have to check if the device already exists in the database

        document_to_find = {"display_name": display_name}

        # get the database on which we work
        database = self.client["backends"]

        # get the collection on which we work
        collection = database["configs"]

        result_found = collection.find_one(document_to_find)
        config_dict.display_name = display_name
        if result_found:
            # update the file
            self.update_file(
                content_dict=config_dict.model_dump(),
                storage_path=config_path,
                job_id=result_found["_id"],
            )
            return

        # if the device does not exist, we have to create it

        config_id = uuid.uuid4().hex[:24]
        self.upload(config_dict.model_dump(), config_path, config_id)

    @validate_active
    def get_config(self, display_name: DisplayNameStr) -> BackendConfigSchemaIn:
        """
        The function that downloads the spooler configuration to the storage.

        Args:
            display_name : The name of the backend

        Raises:
            FileNotFoundError: If the backend does not exist

        Returns:
            The configuration of the backend in complete form.
        """
        # get the database on which we work
        database = self.client["backends"]
        config_collection = database["configs"]

        # create the filter for the document with display_name that is equal to display_name
        document_to_find = {"display_name": display_name}
        backend_config_dict = config_collection.find_one(document_to_find)

        if not backend_config_dict:
            raise FileNotFoundError("The backend does not exist for the given storage.")

        backend_config_dict.pop("_id")
        return BackendConfigSchemaIn(**backend_config_dict)

    def upload_job(
        self, job_dict: dict, display_name: DisplayNameStr, username: str
    ) -> str:
        """
        Upload the job to the storage provider.

        Args:
            job_dict: the full job dict
            display_name: the name of the backend
            username: the name of the user that submitted the job

        Returns:
            The job id of the uploaded job.
        """

        storage_path = "jobs/queued/" + display_name
        job_id = (uuid.uuid4().hex)[:24]

        self.upload(content_dict=job_dict, storage_path=storage_path, job_id=job_id)
        return job_id

    def upload_status(
        self, display_name: DisplayNameStr, username: str, job_id: str
    ) -> StatusMsgDict:
        """
        This function uploads a status file to the backend and creates the status dict.

        Args:
            display_name: The name of the backend to which we want to upload the job
            username: The username of the user that is uploading the job
            job_id: The job_id of the job that we want to upload the status for

        Returns:
            The status dict of the job
        """
        storage_path = "status/" + display_name
        status_draft = {
            "job_id": job_id,
            "status": "INITIALIZING",
            "detail": "Got your json.",
            "error_message": "None",
        }

        # should we also upload the username into the dict ?
        status_dict = StatusMsgDict(**status_draft)
        # now upload the status dict
        self.upload(
            content_dict=status_dict.model_dump(),
            storage_path=storage_path,
            job_id=job_id,
        )
        return status_dict

    def get_status(
        self, display_name: DisplayNameStr, username: str, job_id: str
    ) -> StatusMsgDict:
        """
        This function gets the status file from the backend and returns the status dict.

        Args:
            display_name: The name of the backend to which we want to upload the job
            username: The username of the user that is uploading the job
            job_id: The job_id of the job that we want to upload the status for

        Returns:
            The status dict of the job
        """
        status_json_dir = "status/" + display_name

        try:
            status_dict = self.get_file_content(
                storage_path=status_json_dir, job_id=job_id
            )
            return StatusMsgDict(**status_dict)
        except FileNotFoundError as err:
            # if the job_id is not valid, we return an error
            return StatusMsgDict(
                job_id=job_id,
                status="ERROR",
                detail="The job_id is not valid.",
                error_message=str(err),
            )

    def get_result(
        self, display_name: DisplayNameStr, username: str, job_id: str
    ) -> ResultDict:
        """
        This function gets the result file from the backend and returns the result dict.

        Args:
            display_name: The name of the backend to which we want to upload the job
            username: The username of the user that is uploading the job
            job_id: The job_id of the job that we want to upload the status for

        Returns:
            The result dict of the job. If the information is not available, the result dict
            has a status of "ERROR".
        """
        result_json_dir = "results/" + display_name
        try:
            result_dict = self.get_file_content(
                storage_path=result_json_dir, job_id=job_id
            )
        except FileNotFoundError:
            # if the job_id is not valid, we return an error
            return ResultDict(
                display_name=display_name,
                backend_version="",
                job_id=job_id,
                qobj_id=None,
                success=False,
                status="ERROR",
                header={},
                results=[],
            )
        backend_config_info = self.get_backend_dict(display_name)
        result_dict["backend_name"] = backend_config_info.backend_name

        typed_result = ResultDict(**result_dict)
        return typed_result

    def update_in_database(
        self,
        result_dict: ResultDict | None,
        status_msg_dict: StatusMsgDict,
        job_id: str,
        display_name: DisplayNameStr,
    ) -> None:
        """
        Upload the status and result to the `StorageProvider`.

        The function checks if the reported status of the job has changed to DONE. If so, it will create
        a result json file and move the job json file to the finished folder. It will also update the
        status json file.

        Args:
            result_dict: the dictionary containing the result of the job
            status_msg_dict: the dictionary containing the status message of the job
            job_id: the name of the job
            display_name: the name of the backend

        Returns:
            None

        Raises:

        """

        job_json_start_dir = "jobs/running"
        # check if the job is done or had an error
        if status_msg_dict.status == "DONE":
            # test if the result dict is None
            if result_dict is None:
                raise ValueError(
                    "The 'result_dict' argument cannot be None if the job is done."
                )
            # let us create the result json file
            result_json_dir = "results/" + display_name
            self.upload(result_dict.model_dump(), result_json_dir, job_id)

            # now move the job out of the running jobs into the finished jobs
            job_finished_json_dir = "jobs/finished/" + display_name
            self.move_file(job_json_start_dir, job_finished_json_dir, job_id)

        elif status_msg_dict.status == "ERROR":
            # because there was an error, we move the job to the deleted jobs
            deleted_json_dir = "jobs/deleted"
            self.move_file(job_json_start_dir, deleted_json_dir, job_id)

        # TODO: most likely we should raise an error if the status of the job is not DONE or ERROR

        # and create the status json file
        status_json_dir = "status/" + display_name
        self.update_file(status_msg_dict.model_dump(), status_json_dir, job_id)

    def get_file_queue(self, storage_path: str) -> list[str]:
        """
        Get a list of documents in the collection of all the queued jobs.

        Args:
            storage_path: Where are we looking for the files.

        Returns:
            A list of files that was found.
        """
        # strip trailing and leading slashes from the paths
        storage_path = storage_path.strip("/")

        # get the database on which we work
        database = self.client[storage_path.split("/")[0]]

        # get the collection on which we work
        collection_name = ".".join(storage_path.split("/")[1:])
        collection = database[collection_name]

        # now get the id of all the documents in the collection
        results = collection.find({}, {"_id": 1})
        file_list = []
        for result in results:
            file_list.append(str(result["_id"]))
        return file_list

    def get_next_job_in_queue(self, display_name: str) -> NextJobSchema:
        """
        A function that obtains the next job in the queue. If there is no job, it returns an empty
        dict. If there is a job, it moves the job from the queue to the running folder.
        It also update the time stamp for when the system last looked into the file queue.

        Args:
            display_name: The name of the backend

        Returns:
            the job dict
        """

        queue_dir = "jobs/queued/" + display_name
        job_dict = {"job_id": 0, "job_json_path": "None"}
        job_list = self.get_file_queue(queue_dir)

        # update the time stamp of the last job
        self.timestamp_queue(display_name)

        # if there is a job, we should move it
        if job_list:
            job_id = job_list[0]
            job_dict["job_id"] = job_id

            # and move the file into the right directory
            self.move_file(queue_dir, "jobs/running", job_id)
            job_dict["job_json_path"] = "jobs/running"
        return NextJobSchema(**job_dict)

__init__(login_dict, name, is_active=True)

Set up the neccessary keys and create the client through which all the connections will run.

Parameters:

Name	Type	Description	Default
login_dict	MongodbLoginInformation	The login dict that contains the neccessary information to connect to the mongodb	required
name	str	The name of the storage provider	required
is_active	bool	Is the storage provider active.	True

Raises:

Type	Description
ValidationError	If the login_dict is not valid

Source code in src/sqooler/storage_providers/mongodb.py

def __init__(
    self, login_dict: MongodbLoginInformation, name: str, is_active: bool = True
) -> None:
    """
    Set up the neccessary keys and create the client through which all the connections will run.

    Args:
        login_dict: The login dict that contains the neccessary
                    information to connect to the mongodb
        name: The name of the storage provider
        is_active: Is the storage provider active.


    Raises:
        ValidationError: If the login_dict is not valid
    """
    super().__init__(name, is_active)
    mongodb_username = login_dict.mongodb_username
    mongodb_password = login_dict.mongodb_password
    mongodb_database_url = login_dict.mongodb_database_url

    uri = f"mongodb+srv://{mongodb_username}:{mongodb_password}@{mongodb_database_url}"
    uri = uri + "/?retryWrites=true&w=majority"
    # Create a new client and connect to the server
    self.client: MongoClient = MongoClient(uri)

    # Send a ping to confirm a successful connection
    self.client.admin.command("ping")

delete_file(storage_path, job_id)

Remove the file from the mongodb database

Parameters:

Name	Type	Description	Default
storage_path	str	the path where the file is currently stored, but excluding the file name	required
job_id	str	the name of the file	required

Returns:

Type	Description
None	None

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def delete_file(self, storage_path: str, job_id: str) -> None:
    """
    Remove the file from the mongodb database

    Args:
        storage_path: the path where the file is currently stored, but excluding the file name
        job_id: the name of the file

    Returns:
        None
    """
    # get the database on which we work
    database = self.client[storage_path.split("/")[0]]

    # get the collection on which we work
    collection_name = ".".join(storage_path.split("/")[1:])
    collection = database[collection_name]

    document_to_find = {"_id": ObjectId(job_id)}
    collection.delete_one(document_to_find)

get_backend_status(display_name)

Get the status of the backend. This follows the qiskit logic.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend	required

Returns:

Type	Description
BackendStatusSchemaOut	The status dict of the backend

Raises:

Type	Description
FileNotFoundError	If the backend does not exist

Source code in src/sqooler/storage_providers/mongodb.py

def get_backend_status(
    self, display_name: DisplayNameStr
) -> BackendStatusSchemaOut:
    """
    Get the status of the backend. This follows the qiskit logic.

    Args:
        display_name: The name of the backend

    Returns:
        The status dict of the backend

    Raises:
        FileNotFoundError: If the backend does not exist
    """
    # get the database on which we work
    database = self.client["backends"]
    config_collection = database["configs"]

    # create the filter for the document with display_name that is equal to display_name
    document_to_find = {"display_name": display_name}
    backend_config_dict = config_collection.find_one(document_to_find)

    if not backend_config_dict:
        raise FileNotFoundError(
            f"The backend {display_name} does not exist for the given storageprovider."
        )

    backend_config_dict.pop("_id")
    backend_config_info = BackendConfigSchemaIn(**backend_config_dict)
    qiskit_backend_dict = self.backend_dict_to_qiskit_status(backend_config_info)
    return qiskit_backend_dict

get_backends()

Get a list of all the backends that the provider offers.

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def get_backends(self) -> list[DisplayNameStr]:
    """
    Get a list of all the backends that the provider offers.
    """

    # get the database on which we work
    database = self.client["backends"]
    config_collection = database["configs"]
    # get all the documents in the collection configs and save the disply_name in a list
    backend_names: list[DisplayNameStr] = []
    for config_dict in config_collection.find():
        backend_names.append(config_dict["display_name"])
    return backend_names

get_config(display_name)

The function that downloads the spooler configuration to the storage.

Parameters:

Name	Type	Description	Default
display_name		The name of the backend	required

Raises:

Type	Description
FileNotFoundError	If the backend does not exist

Returns:

Type	Description
BackendConfigSchemaIn	The configuration of the backend in complete form.

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def get_config(self, display_name: DisplayNameStr) -> BackendConfigSchemaIn:
    """
    The function that downloads the spooler configuration to the storage.

    Args:
        display_name : The name of the backend

    Raises:
        FileNotFoundError: If the backend does not exist

    Returns:
        The configuration of the backend in complete form.
    """
    # get the database on which we work
    database = self.client["backends"]
    config_collection = database["configs"]

    # create the filter for the document with display_name that is equal to display_name
    document_to_find = {"display_name": display_name}
    backend_config_dict = config_collection.find_one(document_to_find)

    if not backend_config_dict:
        raise FileNotFoundError("The backend does not exist for the given storage.")

    backend_config_dict.pop("_id")
    return BackendConfigSchemaIn(**backend_config_dict)

get_file_content(storage_path, job_id)

Get the file content from the storage

Parameters:

Name	Type	Description	Default
storage_path	str	the path towards the file, excluding the filename / id	required
job_id	str	the id of the file we are about to look up	required

Returns:

Type	Description
dict	The content of the file

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def get_file_content(self, storage_path: str, job_id: str) -> dict:
    """
    Get the file content from the storage

    Args:
        storage_path: the path towards the file, excluding the filename / id
        job_id: the id of the file we are about to look up

    Returns:
        The content of the file
    """
    try:
        document_to_find = {"_id": ObjectId(job_id)}
    except InvalidId as err:
        raise FileNotFoundError(
            f"The job_id {job_id} is not valid. Please check the job_id."
        ) from err

    document_to_find = {"_id": ObjectId(job_id)}

    # get the database on which we work
    database = self.client[storage_path.split("/")[0]]

    # get the collection on which we work
    collection_name = ".".join(storage_path.split("/")[1:])
    collection = database[collection_name]

    result_found = collection.find_one(document_to_find)

    if not result_found:
        raise FileNotFoundError(
            f"Could not find a file under {storage_path} with the id {job_id}."
        )

    # remove the id from the result dict for further use
    result_found.pop("_id", None)
    return result_found

get_file_queue(storage_path)

Get a list of documents in the collection of all the queued jobs.

Parameters:

Name	Type	Description	Default
storage_path	str	Where are we looking for the files.	required

Returns:

Type	Description
list[str]	A list of files that was found.

Source code in src/sqooler/storage_providers/mongodb.py

def get_file_queue(self, storage_path: str) -> list[str]:
    """
    Get a list of documents in the collection of all the queued jobs.

    Args:
        storage_path: Where are we looking for the files.

    Returns:
        A list of files that was found.
    """
    # strip trailing and leading slashes from the paths
    storage_path = storage_path.strip("/")

    # get the database on which we work
    database = self.client[storage_path.split("/")[0]]

    # get the collection on which we work
    collection_name = ".".join(storage_path.split("/")[1:])
    collection = database[collection_name]

    # now get the id of all the documents in the collection
    results = collection.find({}, {"_id": 1})
    file_list = []
    for result in results:
        file_list.append(str(result["_id"]))
    return file_list

get_job_content(storage_path, job_id)

Get the content of the job from the storage. This is a wrapper around get_file_content and and handles the different ways of identifiying the job.

storage_path: the path towards the file, excluding the filename / id job_id: the id of the file we are about to look up

Returns:

Source code in src/sqooler/storage_providers/mongodb.py

def get_job_content(self, storage_path: str, job_id: str) -> dict:
    """
    Get the content of the job from the storage. This is a wrapper around get_file_content
    and and handles the different ways of identifiying the job.

    storage_path: the path towards the file, excluding the filename / id
    job_id: the id of the file we are about to look up

    Returns:

    """
    job_dict = self.get_file_content(storage_path=storage_path, job_id=job_id)
    job_dict.pop("_id", None)
    return job_dict

get_next_job_in_queue(display_name)

A function that obtains the next job in the queue. If there is no job, it returns an empty dict. If there is a job, it moves the job from the queue to the running folder. It also update the time stamp for when the system last looked into the file queue.

Parameters:

Name	Type	Description	Default
display_name	str	The name of the backend	required

Returns:

Type	Description
NextJobSchema	the job dict

Source code in src/sqooler/storage_providers/mongodb.py

def get_next_job_in_queue(self, display_name: str) -> NextJobSchema:
    """
    A function that obtains the next job in the queue. If there is no job, it returns an empty
    dict. If there is a job, it moves the job from the queue to the running folder.
    It also update the time stamp for when the system last looked into the file queue.

    Args:
        display_name: The name of the backend

    Returns:
        the job dict
    """

    queue_dir = "jobs/queued/" + display_name
    job_dict = {"job_id": 0, "job_json_path": "None"}
    job_list = self.get_file_queue(queue_dir)

    # update the time stamp of the last job
    self.timestamp_queue(display_name)

    # if there is a job, we should move it
    if job_list:
        job_id = job_list[0]
        job_dict["job_id"] = job_id

        # and move the file into the right directory
        self.move_file(queue_dir, "jobs/running", job_id)
        job_dict["job_json_path"] = "jobs/running"
    return NextJobSchema(**job_dict)

get_result(display_name, username, job_id)

This function gets the result file from the backend and returns the result dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
ResultDict	The result dict of the job. If the information is not available, the result dict
ResultDict	has a status of "ERROR".

Source code in src/sqooler/storage_providers/mongodb.py

def get_result(
    self, display_name: DisplayNameStr, username: str, job_id: str
) -> ResultDict:
    """
    This function gets the result file from the backend and returns the result dict.

    Args:
        display_name: The name of the backend to which we want to upload the job
        username: The username of the user that is uploading the job
        job_id: The job_id of the job that we want to upload the status for

    Returns:
        The result dict of the job. If the information is not available, the result dict
        has a status of "ERROR".
    """
    result_json_dir = "results/" + display_name
    try:
        result_dict = self.get_file_content(
            storage_path=result_json_dir, job_id=job_id
        )
    except FileNotFoundError:
        # if the job_id is not valid, we return an error
        return ResultDict(
            display_name=display_name,
            backend_version="",
            job_id=job_id,
            qobj_id=None,
            success=False,
            status="ERROR",
            header={},
            results=[],
        )
    backend_config_info = self.get_backend_dict(display_name)
    result_dict["backend_name"] = backend_config_info.backend_name

    typed_result = ResultDict(**result_dict)
    return typed_result

get_status(display_name, username, job_id)

This function gets the status file from the backend and returns the status dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
StatusMsgDict	The status dict of the job

Source code in src/sqooler/storage_providers/mongodb.py

def get_status(
    self, display_name: DisplayNameStr, username: str, job_id: str
) -> StatusMsgDict:
    """
    This function gets the status file from the backend and returns the status dict.

    Args:
        display_name: The name of the backend to which we want to upload the job
        username: The username of the user that is uploading the job
        job_id: The job_id of the job that we want to upload the status for

    Returns:
        The status dict of the job
    """
    status_json_dir = "status/" + display_name

    try:
        status_dict = self.get_file_content(
            storage_path=status_json_dir, job_id=job_id
        )
        return StatusMsgDict(**status_dict)
    except FileNotFoundError as err:
        # if the job_id is not valid, we return an error
        return StatusMsgDict(
            job_id=job_id,
            status="ERROR",
            detail="The job_id is not valid.",
            error_message=str(err),
        )

move_file(start_path, final_path, job_id)

Move the file from start_path to final_path

start_path: the path where the file is currently stored, but excluding the file name final_path: the path where the file should be stored, but excluding the file name job_id: the name of the file. Is a json file

Returns:

Type	Description
None	None

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def move_file(self, start_path: str, final_path: str, job_id: str) -> None:
    """
    Move the file from start_path to final_path

    start_path: the path where the file is currently stored, but excluding the file name
    final_path: the path where the file should be stored, but excluding the file name
    job_id: the name of the file. Is a json file

    Returns:
        None
    """
    # get the database on which we work
    database = self.client[start_path.split("/")[0]]

    # get the collection on which we work
    collection_name = ".".join(start_path.split("/")[1:])
    collection = database[collection_name]

    document_to_find = {"_id": ObjectId(job_id)}
    result_found = collection.find_one(document_to_find)

    # delete the old file
    collection.delete_one(document_to_find)

    # add the document to the new collection
    database = self.client[final_path.split("/")[0]]
    collection_name = ".".join(final_path.split("/")[1:])
    collection = database[collection_name]
    collection.insert_one(result_found)

update_file(content_dict, storage_path, job_id)

Update the file content. It replaces the old content with the new content.

Parameters:

Name	Type	Description	Default
content_dict	dict	The dictionary containing the new content of the file	required
storage_path	str	The path to the file	required
job_id	str	The id of the job	required

Returns:

Type	Description
None	None

Raises:

Type	Description
FileNotFoundError	If the file is not found

Source code in src/sqooler/storage_providers/mongodb.py

def update_file(self, content_dict: dict, storage_path: str, job_id: str) -> None:
    """
    Update the file content. It replaces the old content with the new content.


    Args:
        content_dict: The dictionary containing the new content of the file
        storage_path: The path to the file
        job_id: The id of the job

    Returns:
        None

    Raises:
        FileNotFoundError: If the file is not found
    """
    # get the database on which we work
    database = self.client[storage_path.split("/")[0]]

    # get the collection on which we work
    collection_name = ".".join(storage_path.split("/")[1:])
    collection = database[collection_name]

    filter_dict = {"_id": ObjectId(job_id)}
    result = collection.replace_one(filter_dict, content_dict)

    if result.matched_count == 0:
        raise FileNotFoundError(f"Could not update file under {storage_path}")

update_in_database(result_dict, status_msg_dict, job_id, display_name)

Upload the status and result to the StorageProvider.

The function checks if the reported status of the job has changed to DONE. If so, it will create a result json file and move the job json file to the finished folder. It will also update the status json file.

Parameters:

Name	Type	Description	Default
result_dict	ResultDict | None	the dictionary containing the result of the job	required
status_msg_dict	StatusMsgDict	the dictionary containing the status message of the job	required
job_id	str	the name of the job	required
display_name	DisplayNameStr	the name of the backend	required

Returns:

Type	Description
None	None

Raises:

Source code in src/sqooler/storage_providers/mongodb.py

def update_in_database(
    self,
    result_dict: ResultDict | None,
    status_msg_dict: StatusMsgDict,
    job_id: str,
    display_name: DisplayNameStr,
) -> None:
    """
    Upload the status and result to the `StorageProvider`.

    The function checks if the reported status of the job has changed to DONE. If so, it will create
    a result json file and move the job json file to the finished folder. It will also update the
    status json file.

    Args:
        result_dict: the dictionary containing the result of the job
        status_msg_dict: the dictionary containing the status message of the job
        job_id: the name of the job
        display_name: the name of the backend

    Returns:
        None

    Raises:

    """

    job_json_start_dir = "jobs/running"
    # check if the job is done or had an error
    if status_msg_dict.status == "DONE":
        # test if the result dict is None
        if result_dict is None:
            raise ValueError(
                "The 'result_dict' argument cannot be None if the job is done."
            )
        # let us create the result json file
        result_json_dir = "results/" + display_name
        self.upload(result_dict.model_dump(), result_json_dir, job_id)

        # now move the job out of the running jobs into the finished jobs
        job_finished_json_dir = "jobs/finished/" + display_name
        self.move_file(job_json_start_dir, job_finished_json_dir, job_id)

    elif status_msg_dict.status == "ERROR":
        # because there was an error, we move the job to the deleted jobs
        deleted_json_dir = "jobs/deleted"
        self.move_file(job_json_start_dir, deleted_json_dir, job_id)

    # TODO: most likely we should raise an error if the status of the job is not DONE or ERROR

    # and create the status json file
    status_json_dir = "status/" + display_name
    self.update_file(status_msg_dict.model_dump(), status_json_dir, job_id)

upload(content_dict, storage_path, job_id)

Upload the file to the storage

content_dict: the content that should be uploaded onto the mongodb base storage_path: the access path towards the mongodb collection job_id: the id of the file we are about to create

Source code in src/sqooler/storage_providers/mongodb.py

@validate_active
def upload(self, content_dict: dict, storage_path: str, job_id: str) -> None:
    """
    Upload the file to the storage

    content_dict: the content that should be uploaded onto the mongodb base
    storage_path: the access path towards the mongodb collection
    job_id: the id of the file we are about to create
    """
    storage_splitted = storage_path.split("/")

    # get the database on which we work
    database = self.client[storage_splitted[0]]

    # get the collection on which we work
    collection_name = ".".join(storage_splitted[1:])
    collection = database[collection_name]

    content_dict["_id"] = ObjectId(job_id)
    collection.insert_one(content_dict)

    # remove the id from the content dict for further use
    content_dict.pop("_id", None)

upload_config(config_dict, display_name)

The function that uploads the spooler configuration to the storage.

Parameters:

Name	Type	Description	Default
config_dict	BackendConfigSchemaIn	The dictionary containing the configuration	required
display_name		The name of the backend	required

Returns:

Type	Description
None	None

Source code in src/sqooler/storage_providers/mongodb.py

def upload_config(
    self, config_dict: BackendConfigSchemaIn, display_name: DisplayNameStr
) -> None:
    """
    The function that uploads the spooler configuration to the storage.

    Args:
        config_dict: The dictionary containing the configuration
        display_name : The name of the backend

    Returns:
        None
    """
    config_path = "backends/configs"

    # first we have to check if the device already exists in the database

    document_to_find = {"display_name": display_name}

    # get the database on which we work
    database = self.client["backends"]

    # get the collection on which we work
    collection = database["configs"]

    result_found = collection.find_one(document_to_find)
    config_dict.display_name = display_name
    if result_found:
        # update the file
        self.update_file(
            content_dict=config_dict.model_dump(),
            storage_path=config_path,
            job_id=result_found["_id"],
        )
        return

    # if the device does not exist, we have to create it

    config_id = uuid.uuid4().hex[:24]
    self.upload(config_dict.model_dump(), config_path, config_id)

upload_job(job_dict, display_name, username)

Upload the job to the storage provider.

Parameters:

Name	Type	Description	Default
job_dict	dict	the full job dict	required
display_name	DisplayNameStr	the name of the backend	required
username	str	the name of the user that submitted the job	required

Returns:

Type	Description
str	The job id of the uploaded job.

Source code in src/sqooler/storage_providers/mongodb.py

def upload_job(
    self, job_dict: dict, display_name: DisplayNameStr, username: str
) -> str:
    """
    Upload the job to the storage provider.

    Args:
        job_dict: the full job dict
        display_name: the name of the backend
        username: the name of the user that submitted the job

    Returns:
        The job id of the uploaded job.
    """

    storage_path = "jobs/queued/" + display_name
    job_id = (uuid.uuid4().hex)[:24]

    self.upload(content_dict=job_dict, storage_path=storage_path, job_id=job_id)
    return job_id

upload_status(display_name, username, job_id)

This function uploads a status file to the backend and creates the status dict.

Parameters:

Name	Type	Description	Default
display_name	DisplayNameStr	The name of the backend to which we want to upload the job	required
username	str	The username of the user that is uploading the job	required
job_id	str	The job_id of the job that we want to upload the status for	required

Returns:

Type	Description
StatusMsgDict	The status dict of the job

Source code in src/sqooler/storage_providers/mongodb.py

def upload_status(
    self, display_name: DisplayNameStr, username: str, job_id: str
) -> StatusMsgDict:
    """
    This function uploads a status file to the backend and creates the status dict.

    Args:
        display_name: The name of the backend to which we want to upload the job
        username: The username of the user that is uploading the job
        job_id: The job_id of the job that we want to upload the status for

    Returns:
        The status dict of the job
    """
    storage_path = "status/" + display_name
    status_draft = {
        "job_id": job_id,
        "status": "INITIALIZING",
        "detail": "Got your json.",
        "error_message": "None",
    }

    # should we also upload the username into the dict ?
    status_dict = StatusMsgDict(**status_draft)
    # now upload the status dict
    self.upload(
        content_dict=status_dict.model_dump(),
        storage_path=storage_path,
        job_id=job_id,
    )
    return status_dict

Comments