steps.api.cloud_objects

Submodules

• steps.api.cloud_objects.cloud_base
• steps.api.cloud_objects.cloud_job
• steps.api.cloud_objects.cloud_platform
• steps.api.cloud_objects.cloud_token
• steps.api.cloud_objects.cloud_user

Classes

TokenType	Enum defining different token types used for authentication.
JobToken	Job token class for job-specific authentication.
LoginToken	Login token class for user authentication.
ApiKey	API key class for API authentication.
User	User class representing a user in the cloud system.
Platform	Platform class representing a worker in the cloud system.
Job	Job class representing a computational job in the cloud system.

Package Contents

class steps.api.cloud_objects.TokenType(*args, **kwds)

Bases: 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.

Login = 0

Job = 1

API_Key = 2

class steps.api.cloud_objects.JobToken(label: str = None, duration: int = 99999, token_id: int = -1, user_id: int = -1, token: str = '', **kwargs)

Bases: 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.

Parameters:

• label – Label for the token

• duration – Token duration in seconds

• token_id – Token ID (defaults to -1 for new tokens)

• user_id – ID of the user the token belongs to

• token – Token string

• kwargs – Additional attributes to set on the token

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

DURATION_KEY = 'duration'

LABEL_KEY = 'label'

label: str = None

token_id: int

user_id: int = -1

duration: int = 99999

token: str = ''

create(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Create this job token in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication and owning the token

Raises:

• AssertionError – If the user doesn’t have a login token

• RuntimeError – If job token creation fails

read(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Read this job token’s information from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with token data from response

Raises:

• AssertionError – If the user doesn’t have job tokens or the token wasn’t created

• RuntimeError – If reading token data fails

delete(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

Delete this job token from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Raises:

• AssertionError – If the token wasn’t created

• RuntimeError – If deleting the token fails

classmethod list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) → list[JobToken]

List all job tokens in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Returns:

List of JobToken objects

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If listing tokens fails

class steps.api.cloud_objects.LoginToken

Login token class for user authentication.

This class manages the tokens used for user login and general API access.

ENDPOINT = '/api/login'

token: str = None

create(context: steps.api.utils.CloudContext, user=None) → None

Create a login token for the user.

Parameters:

• context – Cloud context for API configuration

• user – User to create the login token for

Raises:

RuntimeError – If login token creation fails

class steps.api.cloud_objects.ApiKey(label: str = None, duration: int = 99999, api_key: str = '', api_key_id: int = -1, **_)

Bases: 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.

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

DURATION_KEY = 'duration'

LABEL_KEY = 'label'

api_key: str = ''

api_key_id: int = -1

label: str = None

duration: int = 99999

property token: str

Get the token string for this API key.

Returns:

API key string

create(context: steps.api.utils.CloudContext, admin=None) → None

Create this API key in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If API key creation fails

class steps.api.cloud_objects.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: 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.

Parameters:

• first_name – User’s first name

• last_name – User’s last name

• user_id – User’s ID (defaults to -1 for new users)

• email – User’s email address

• password – User’s password

• platform_ids – List of platform IDs the user has access to

• is_admin – Whether the user has admin privileges

• priority – User’s priority level (default: 2)

• company_id – User’s company ID

• kwargs – Additional attributes to set on the user

EMAIL_KEY = 'email'

PASSWORD_KEY = 'password'

FIRST_NAME_KEY = 'first_name'

LAST_NAME_KEY = 'last_name'

PLATFORM_IDS_KEY = 'platform_ids'

IS_ADMIN_KEY = 'is_admin'

COMPANY_ID_KEY = 'company_id'

PRIORITY_KEY = 'priority'

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

get_token(token_type: steps.api.cloud_objects.cloud_base.TokenType) → None

Get a token of the specified type for this user.

Parameters:

token_type – Type of token to retrieve

Returns:

Token string for the requested token type

Raises:

NotImplementedError – If the token type is not supported

first_name: str = ''

last_name: str = ''

email: str = ''

password: str = ''

user_id: int

is_admin: bool = False

priority: int = 2

company_id: int = None

platform_ids: list[str] = []

job_tokens: list[steps.api.cloud_objects.cloud_token.JobToken] = []

login_token: steps.api.cloud_objects.cloud_token.LoginToken

api_key: steps.api.cloud_objects.cloud_token.ApiKey = None

keys() → list[str]

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.

create(context: steps.api.utils.CloudContext, user: User) → None

Create this user in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication for the creation

Raises:

• RuntimeError – If user creation fails

• AssertionError – If trying to create an admin user

read(context: steps.api.utils.CloudContext, user=None) → dict[str]

Read this user’s information from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication (defaults to self)

Returns:

Dictionary with user data from response

Raises:

• RuntimeError – If reading user data fails

• AssertionError – If the user has not been created yet

update(context: steps.api.utils.CloudContext, user, update_dict: dict[str]) → None

Update this user’s information in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication (defaults to self)

• update_dict – Dictionary with fields to update

Raises:

• RuntimeError – If updating user data fails

• AssertionError – If the user has not been created yet

delete(context: steps.api.utils.CloudContext, user: User) → None

Delete this user from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Raises:

• RuntimeError – If deleting the user fails

• AssertionError – If the user has not been created or is an admin

log(context: steps.api.utils.CloudContext) → None

Login the user and obtain a login token.

Parameters:

context – Cloud context for API configuration

Raises:

AssertionError – If the user is not an admin and hasn’t been created

create_job_token(context: steps.api.utils.CloudContext, token_dict: dict[str]) → None

Create a job token for this user.

Parameters:

• context – Cloud context for API configuration

• token_dict – Dictionary with token parameters

Raises:

AssertionError – If the user hasn’t been created

create_api_key(context: steps.api.utils.CloudContext, api_key_dict: dict[str]) → None

Create an API key for this user.

Parameters:

• context – Cloud context for API configuration

• api_key_dict – Dictionary with API key parameters

Raises:

AssertionError – If the user hasn’t been created

classmethod list_all(context: steps.api.utils.CloudContext, admin: User) → list[User]

List all users in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Returns:

List of User objects

Raises:

• RuntimeError – If listing users fails

• AssertionError – If the authenticating user is not an admin

class steps.api.cloud_objects.Platform(name: str = '', type: str = 'simulator', platform_id: str = None, worker_id: str = str(uuid4()), specs: dict[str] = {'available_commands': ['probs']}, **kwargs)

Bases: 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.

Parameters:

• name – Name of the platform

• type – Type of platform (default: “simulator”)

• platform_id – Platform ID (None for new platforms)

• worker_id – Worker ID (defaults to a new UUID)

• specs – Platform specifications

• kwargs – Additional attributes to set on the platform

Raises:

AssertionError – If Platform.ADMIN is None or not an admin

ADMIN = None

REGISTER_ENDPOINT = '/admin/platforms/register'

JOB_ENDPOINT = '/admin/job'

ASSIGN_ENDPOINT = '/api/platform-permissions'

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

PLATFORM_NAME_KEY = 'platform_name'

DEFAULT_NAME = 'sim:qa'

name = ''

type = 'simulator'

worker_id: str = ''

platform_id: str

specs

register(context: steps.api.utils.CloudContext, register_dict: dict[str] = None) → None

Register this platform in the cloud.

Parameters:

• context – Cloud context for API configuration

• register_dict – Dictionary with registration parameters (optional)

Raises:

RuntimeError – If platform registration fails

read(context: steps.api.utils.CloudContext) → dict[str]

Read this platform’s information from the cloud.

Parameters:

context – Cloud context for API configuration

Returns:

Dictionary with platform data from response

Raises:

RuntimeError – If reading platform data fails

update(context: steps.api.utils.CloudContext, update_dict: dict) → None

Update this platform’s information in the cloud.

Parameters:

• context – Cloud context for API configuration

• update_dict – Dictionary with fields to update

Raises:

RuntimeError – If updating platform data fails

delete(context: steps.api.utils.CloudContext) → None

Delete this platform from the cloud.

Parameters:

context – Cloud context for API configuration

Raises:

RuntimeError – If deleting the platform fails

fetch_job(context: steps.api.utils.CloudContext) → steps.api.cloud_objects.cloud_job.Job

Fetch a job for this platform to process.

Parameters:

context – Cloud context for API configuration

Returns:

Job object for the fetched job

Raises:

RuntimeError – If fetching a job fails

run_job(context: steps.api.utils.CloudContext, job) → None

progress_job(context: steps.api.utils.CloudContext, job, progress_dict: dict[str]) → None

finish_job(context: steps.api.utils.CloudContext, job, finish_dict: dict[str]) → None

assign_user(context: steps.api.utils.CloudContext, user: steps.api.cloud_objects.cloud_user.User) → None

assign_token(context: steps.api.utils.CloudContext, job_token) → None

assign_company(context: steps.api.utils.CloudContext, company_id: int) → None

classmethod list_all(context: steps.api.utils.CloudContext) → list[Platform]

List all platforms in the cloud.

Parameters:

context – Cloud context for API configuration

Returns:

List of Platform objects

Raises:

• AssertionError – If ADMIN is not an admin

• RuntimeError – If listing platforms fails

class steps.api.cloud_objects.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: 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.

Parameters:

• platform_name – Name of the platform to run the job on

• job_name – Name of the job

• payload – Job payload containing command, circuit, parameters, etc.

• pcvl_version – Version of the PCVL library to use

• process_id – Unique ID for the process (defaults to a new UUID)

• job_id – Job ID (empty for new jobs)

• _ – Additional attributes to set on the job

JOB_NAME_KEY = 'job_name'

PLATFORM_NAME_KEY = 'platform_name'

PAYLOAD_KEY = 'payload'

PCVL_VERSION_KEY = 'pcvl_version'

PROCESS_ID_KEY = 'process_id'

DEFAULT_PAYLOAD

classmethod get_object_name() → str

classmethod get_endpoint() → str

classmethod get_api(context: steps.api.utils.CloudContext) → steps.api.utils.Api

classmethod get_token_method() → str

DEFAULT_PLATFORM_NAME = 'sim:qa'

platform_name: str = 'sim:qa'

job_name: str = 'qa_job'

payload: dict[str]

pcvl_version: str = '0.12.0'

process_id: str = ''

job_id = ''

create(context: steps.api.utils.CloudContext, user, create_dict: dict = {}) → None

Create this job in the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

• create_dict – Additional parameters for job creation

Raises:

RuntimeError – If job creation fails

read(context: steps.api.utils.CloudContext, user) → dict[str]

Read this job’s information from the cloud.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job data from response

Raises:

RuntimeError – If reading job data fails

cancel(context: steps.api.utils.CloudContext, user) → dict[str]

Cancel this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Raises:

RuntimeError – If canceling the job fails

get_status(context: steps.api.utils.CloudContext, user) → dict[str]

Get the current status of this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job status information

Raises:

RuntimeError – If retrieving job status fails

get_results(context: steps.api.utils.CloudContext, user) → dict[str]

Get the results of this job.

Parameters:

• context – Cloud context for API configuration

• user – User providing authentication

Returns:

Dictionary with job results

Raises:

RuntimeError – If retrieving job results fails

classmethod cancel_many(context: steps.api.utils.CloudContext, jobs: list, admin: steps.api.cloud_objects.cloud_user.User) → None

Cancel multiple jobs at once.

Parameters:

• context – Cloud context for API configuration

• jobs – List of Job objects to cancel

• admin – Admin user providing authentication

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If canceling the jobs fails

classmethod list_all(context: steps.api.utils.CloudContext, admin: steps.api.cloud_objects.cloud_user.User) → list[Job]

List all jobs in the cloud.

Parameters:

• context – Cloud context for API configuration

• admin – Admin user providing authentication

Returns:

List of Job objects

Raises:

• AssertionError – If the user is not an admin

• RuntimeError – If listing jobs fails