steps.api.cloud_objects ======================= .. py:module:: steps.api.cloud_objects Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/steps/api/cloud_objects/cloud_base/index /autoapi/steps/api/cloud_objects/cloud_job/index /autoapi/steps/api/cloud_objects/cloud_platform/index /autoapi/steps/api/cloud_objects/cloud_token/index /autoapi/steps/api/cloud_objects/cloud_user/index Classes ------- .. autoapisummary:: steps.api.cloud_objects.TokenType steps.api.cloud_objects.JobToken steps.api.cloud_objects.LoginToken steps.api.cloud_objects.ApiKey steps.api.cloud_objects.User steps.api.cloud_objects.Platform steps.api.cloud_objects.Job Package Contents ---------------- .. py:class:: TokenType(*args, **kwds) Bases: :py:obj:`enum.Enum` Enum defining different token types used for authentication. Attributes: Login (int): Login token type used for general authentication. Job (int): Job token type used for job-specific operations. API_Key (int): API key token type used for worker access. .. py:attribute:: Login :value: 0 .. py:attribute:: Job :value: 1 .. py:attribute:: API_Key :value: 2 .. py:class:: JobToken(label: str = None, duration: int = 99999, token_id: int = -1, user_id: int = -1, token: str = '', **kwargs) Bases: :py:obj:`steps.api.cloud_objects.cloud_base.ACloudObject` Job token class for job-specific authentication. This class handles the creation and management of tokens used for job operations. :param label: Label for the token :param duration: Token duration in seconds :param token_id: Token ID (defaults to -1 for new tokens) :param user_id: ID of the user the token belongs to :param token: Token string :param kwargs: Additional attributes to set on the token .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: DURATION_KEY :value: 'duration' .. py:attribute:: LABEL_KEY :value: 'label' .. py:attribute:: label :type: str :value: None .. py:attribute:: token_id :type: int .. py:attribute:: user_id :type: int :value: -1 .. py:attribute:: duration :type: int :value: 99999 .. py:attribute:: token :type: str :value: '' .. py:method:: create(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Create this job token in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication and owning the token :raises AssertionError: If the user doesn't have a login token :raises RuntimeError: If job token creation fails .. py:method:: read(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Read this job token's information from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with token data from response :raises AssertionError: If the user doesn't have job tokens or the token wasn't created :raises RuntimeError: If reading token data fails .. py:method:: delete(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None Delete this job token from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :raises AssertionError: If the token wasn't created :raises RuntimeError: If deleting the token fails .. py:method:: list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) -> list[JobToken] :classmethod: List all job tokens in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :return: List of JobToken objects :raises AssertionError: If the user is not an admin :raises RuntimeError: If listing tokens fails .. py:class:: LoginToken Login token class for user authentication. This class manages the tokens used for user login and general API access. .. py:attribute:: ENDPOINT :value: '/api/login' .. py:attribute:: token :type: str :value: None .. py:method:: create(context: steps.api.utils.CloudContext, user=None) -> None Create a login token for the user. :param context: Cloud context for API configuration :param user: User to create the login token for :raises RuntimeError: If login token creation fails .. py:class:: ApiKey(label: str = None, duration: int = 99999, api_key: str = '', api_key_id: int = -1, **_) Bases: :py:obj:`steps.api.cloud_objects.cloud_base.ACloudObject` API key class for API authentication. This class handles the creation and management of API keys for worker access. .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: DURATION_KEY :value: 'duration' .. py:attribute:: LABEL_KEY :value: 'label' .. py:attribute:: api_key :type: str :value: '' .. py:attribute:: api_key_id :type: int :value: -1 .. py:attribute:: label :type: str :value: None .. py:attribute:: duration :type: int :value: 99999 .. py:property:: token :type: str Get the token string for this API key. :return: API key string .. py:method:: create(context: steps.api.utils.CloudContext, admin=None) -> None Create this API key in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :raises AssertionError: If the user is not an admin :raises RuntimeError: If API key creation fails .. py:class:: User(first_name: str = '', last_name: str = '', user_id: int = -1, email: str = '', password: str = '', platform_ids: list[str] = [], is_admin: bool = False, priority: int = 2, company_id: int = None, **kwargs) Bases: :py:obj:`steps.api.cloud_objects.cloud_base.ACloudObject` User class representing a user in the cloud system. This class handles all user-related operations including user creation, authentication, token management, and CRUD operations. :param first_name: User's first name :param last_name: User's last name :param user_id: User's ID (defaults to -1 for new users) :param email: User's email address :param password: User's password :param platform_ids: List of platform IDs the user has access to :param is_admin: Whether the user has admin privileges :param priority: User's priority level (default: 2) :param company_id: User's company ID :param kwargs: Additional attributes to set on the user .. py:attribute:: EMAIL_KEY :value: 'email' .. py:attribute:: PASSWORD_KEY :value: 'password' .. py:attribute:: FIRST_NAME_KEY :value: 'first_name' .. py:attribute:: LAST_NAME_KEY :value: 'last_name' .. py:attribute:: PLATFORM_IDS_KEY :value: 'platform_ids' .. py:attribute:: IS_ADMIN_KEY :value: 'is_admin' .. py:attribute:: COMPANY_ID_KEY :value: 'company_id' .. py:attribute:: PRIORITY_KEY :value: 'priority' .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:method:: get_token(token_type: steps.api.cloud_objects.cloud_base.TokenType) -> None Get a token of the specified type for this user. :param token_type: Type of token to retrieve :return: Token string for the requested token type :raises NotImplementedError: If the token type is not supported .. py:attribute:: first_name :type: str :value: '' .. py:attribute:: last_name :type: str :value: '' .. py:attribute:: email :type: str :value: '' .. py:attribute:: password :type: str :value: '' .. py:attribute:: user_id :type: int .. py:attribute:: is_admin :type: bool :value: False .. py:attribute:: priority :type: int :value: 2 .. py:attribute:: company_id :type: int :value: None .. py:attribute:: platform_ids :type: list[str] :value: [] .. py:attribute:: job_tokens :type: list[steps.api.cloud_objects.cloud_token.JobToken] :value: [] .. py:attribute:: login_token :type: steps.api.cloud_objects.cloud_token.LoginToken .. py:attribute:: api_key :type: steps.api.cloud_objects.cloud_token.ApiKey :value: None .. py:method:: keys() -> list[str] .. py:method:: generate_data() -> None Generate email and password data for this user. Creates a standardized email based on first and last name, and generates a secure random password that meets complexity requirements. .. py:method:: create(context: steps.api.utils.CloudContext, user: User) -> None Create this user in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication for the creation :raises RuntimeError: If user creation fails :raises AssertionError: If trying to create an admin user .. py:method:: read(context: steps.api.utils.CloudContext, user=None) -> dict[str] Read this user's information from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication (defaults to self) :return: Dictionary with user data from response :raises RuntimeError: If reading user data fails :raises AssertionError: If the user has not been created yet .. py:method:: update(context: steps.api.utils.CloudContext, user, update_dict: dict[str]) -> None Update this user's information in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication (defaults to self) :param update_dict: Dictionary with fields to update :raises RuntimeError: If updating user data fails :raises AssertionError: If the user has not been created yet .. py:method:: delete(context: steps.api.utils.CloudContext, user: User) -> None Delete this user from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :raises RuntimeError: If deleting the user fails :raises AssertionError: If the user has not been created or is an admin .. py:method:: log(context: steps.api.utils.CloudContext) -> None Login the user and obtain a login token. :param context: Cloud context for API configuration :raises AssertionError: If the user is not an admin and hasn't been created .. py:method:: create_job_token(context: steps.api.utils.CloudContext, token_dict: dict[str]) -> None Create a job token for this user. :param context: Cloud context for API configuration :param token_dict: Dictionary with token parameters :raises AssertionError: If the user hasn't been created .. py:method:: create_api_key(context: steps.api.utils.CloudContext, api_key_dict: dict[str]) -> None Create an API key for this user. :param context: Cloud context for API configuration :param api_key_dict: Dictionary with API key parameters :raises AssertionError: If the user hasn't been created .. py:method:: list_all(context: steps.api.utils.CloudContext, admin: User) -> list[User] :classmethod: List all users in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :return: List of User objects :raises RuntimeError: If listing users fails :raises AssertionError: If the authenticating user is not an admin .. py:class:: Platform(name: str = '', type: str = 'simulator', platform_id: str = None, worker_id: str = str(uuid4()), specs: dict[str] = {'available_commands': ['probs']}, **kwargs) Bases: :py:obj:`steps.api.cloud_objects.cloud_base.ACloudObject` Platform class representing a worker in the cloud system. This class handles worker registration, job management, and platform permissions. :param name: Name of the platform :param type: Type of platform (default: "simulator") :param platform_id: Platform ID (None for new platforms) :param worker_id: Worker ID (defaults to a new UUID) :param specs: Platform specifications :param kwargs: Additional attributes to set on the platform :raises AssertionError: If Platform.ADMIN is None or not an admin .. py:attribute:: ADMIN :value: None .. py:attribute:: REGISTER_ENDPOINT :value: '/admin/platforms/register' .. py:attribute:: JOB_ENDPOINT :value: '/admin/job' .. py:attribute:: ASSIGN_ENDPOINT :value: '/api/platform-permissions' .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: PLATFORM_NAME_KEY :value: 'platform_name' .. py:attribute:: DEFAULT_NAME :value: 'sim:qa' .. py:attribute:: name :value: '' .. py:attribute:: type :value: 'simulator' .. py:attribute:: worker_id :type: str :value: '' .. py:attribute:: platform_id :type: str .. py:attribute:: specs .. py:method:: register(context: steps.api.utils.CloudContext, register_dict: dict[str] = None) -> None Register this platform in the cloud. :param context: Cloud context for API configuration :param register_dict: Dictionary with registration parameters (optional) :raises RuntimeError: If platform registration fails .. py:method:: read(context: steps.api.utils.CloudContext) -> dict[str] Read this platform's information from the cloud. :param context: Cloud context for API configuration :return: Dictionary with platform data from response :raises RuntimeError: If reading platform data fails .. py:method:: update(context: steps.api.utils.CloudContext, update_dict: dict) -> None Update this platform's information in the cloud. :param context: Cloud context for API configuration :param update_dict: Dictionary with fields to update :raises RuntimeError: If updating platform data fails .. py:method:: delete(context: steps.api.utils.CloudContext) -> None Delete this platform from the cloud. :param context: Cloud context for API configuration :raises RuntimeError: If deleting the platform fails .. py:method:: fetch_job(context: steps.api.utils.CloudContext) -> steps.api.cloud_objects.cloud_job.Job Fetch a job for this platform to process. :param context: Cloud context for API configuration :return: Job object for the fetched job :raises RuntimeError: If fetching a job fails .. py:method:: run_job(context: steps.api.utils.CloudContext, job) -> None .. py:method:: progress_job(context: steps.api.utils.CloudContext, job, progress_dict: dict[str]) -> None .. py:method:: finish_job(context: steps.api.utils.CloudContext, job, finish_dict: dict[str]) -> None .. py:method:: assign_user(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) -> None .. py:method:: assign_token(context: steps.api.utils.CloudContext, job_token) -> None .. py:method:: assign_company(context: steps.api.utils.CloudContext, company_id: int) -> None .. py:method:: list_all(context: steps.api.utils.CloudContext) -> list[Platform] :classmethod: List all platforms in the cloud. :param context: Cloud context for API configuration :return: List of Platform objects :raises AssertionError: If ADMIN is not an admin :raises RuntimeError: If listing platforms fails .. py:class:: Job(platform_name: str = DEFAULT_PLATFORM_NAME, job_name: str = 'qa_job', payload: dict[str] = DEFAULT_PAYLOAD, pcvl_version: str = '0.12.0', process_id: str = str(uuid4()), job_id: str = '', **_) Bases: :py:obj:`steps.api.cloud_objects.cloud_base.ACloudObject` Job class representing a computational job in the cloud system. This class handles job creation, monitoring, and management operations. :param platform_name: Name of the platform to run the job on :param job_name: Name of the job :param payload: Job payload containing command, circuit, parameters, etc. :param pcvl_version: Version of the PCVL library to use :param process_id: Unique ID for the process (defaults to a new UUID) :param job_id: Job ID (empty for new jobs) :param _: Additional attributes to set on the job .. py:attribute:: JOB_NAME_KEY :value: 'job_name' .. py:attribute:: PLATFORM_NAME_KEY :value: 'platform_name' .. py:attribute:: PAYLOAD_KEY :value: 'payload' .. py:attribute:: PCVL_VERSION_KEY :value: 'pcvl_version' .. py:attribute:: PROCESS_ID_KEY :value: 'process_id' .. py:attribute:: DEFAULT_PAYLOAD .. py:method:: get_object_name() -> str :classmethod: .. py:method:: get_endpoint() -> str :classmethod: .. py:method:: get_api(context: steps.api.utils.CloudContext) -> steps.api.utils.Api :classmethod: .. py:method:: get_token_method() -> str :classmethod: .. py:attribute:: DEFAULT_PLATFORM_NAME :value: 'sim:qa' .. py:attribute:: platform_name :type: str :value: 'sim:qa' .. py:attribute:: job_name :type: str :value: 'qa_job' .. py:attribute:: payload :type: dict[str] .. py:attribute:: pcvl_version :type: str :value: '0.12.0' .. py:attribute:: process_id :type: str :value: '' .. py:attribute:: job_id :value: '' .. py:method:: create(context: steps.api.utils.CloudContext, user, create_dict: dict = {}) -> None Create this job in the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :param create_dict: Additional parameters for job creation :raises RuntimeError: If job creation fails .. py:method:: read(context: steps.api.utils.CloudContext, user) -> dict[str] Read this job's information from the cloud. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job data from response :raises RuntimeError: If reading job data fails .. py:method:: cancel(context: steps.api.utils.CloudContext, user) -> dict[str] Cancel this job. :param context: Cloud context for API configuration :param user: User providing authentication :raises RuntimeError: If canceling the job fails .. py:method:: get_status(context: steps.api.utils.CloudContext, user) -> dict[str] Get the current status of this job. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job status information :raises RuntimeError: If retrieving job status fails .. py:method:: get_results(context: steps.api.utils.CloudContext, user) -> dict[str] Get the results of this job. :param context: Cloud context for API configuration :param user: User providing authentication :return: Dictionary with job results :raises RuntimeError: If retrieving job results fails .. py:method:: cancel_many(context: steps.api.utils.CloudContext, jobs: list, admin: steps.api.cloud_objects.cloud_user.User) -> None :classmethod: Cancel multiple jobs at once. :param context: Cloud context for API configuration :param jobs: List of Job objects to cancel :param admin: Admin user providing authentication :raises AssertionError: If the user is not an admin :raises RuntimeError: If canceling the jobs fails .. py:method:: list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) -> list[Job] :classmethod: List all jobs in the cloud. :param context: Cloud context for API configuration :param admin: Admin user providing authentication :return: List of Job objects :raises AssertionError: If the user is not an admin :raises RuntimeError: If listing jobs fails