CCClient

chemcloud.client.CCClient

CCClient(
    *,
    chemcloud_username: Optional[str] = None,
    chemcloud_password: Optional[str] = None,
    profile: Optional[str] = None,
    chemcloud_domain: Optional[str] = None,
    settings: Settings = settings,
    queue: Optional[str] = None,
)

Main client object to perform computations using ChemCloud.

Parameters:

Name	Type	Description	Default
chemcloud_username	Optional[str]	Your ChemCloud username (full email address).	None
chemcloud_password	Optional[str]	Your ChemCloud password.	None
profile	Optional[str]	A named profile for authentication with ChemCloud. No value needs to be passed and most users will only have one login with ChemCloud. CCClient will access the profile by default without a specific name being passed. Pass a value if you have multiple logins to ChemCloud.	None
chemcloud_domain	Optional[str]	The domain for the ChemCloud server. Defaults to https://chemcloud.mtzlab.com.	None
settings	Settings	An instance of the Settings class. Defaults to the global settings object.	settings
queue	Optional[str]	The name of a desired compute queue. If None, default queue is used from settings.	None

Responsibilities

• Expose domain-specific methods (e.g., compute, output) that operate with Python objects.
• Translate raw JSON responses into domain objects (e.g., FutureOutput).
• Handle OpenAPI specification caching, parameter validation, etc.

Source code in chemcloud/client.py

def __init__(
    self,
    *,
    chemcloud_username: Optional[str] = None,
    chemcloud_password: Optional[str] = None,
    profile: Optional[str] = None,
    chemcloud_domain: Optional[str] = None,
    settings: Settings = settings,
    queue: Optional[str] = None,
):
    self._http_client = _HttpClient(
        chemcloud_username=chemcloud_username,
        chemcloud_password=chemcloud_password,
        profile=profile,
        chemcloud_domain=chemcloud_domain,
        settings=settings,
    )
    self.queue = queue
    self._settings = settings
    self._openapi_spec: Optional[dict[str, Any]] = None

version property

version: str

Returns chemcloud client version

supported_programs property

supported_programs: list[str]

Sync wrapper for supported_programs_async.

compute_async async

compute_async(
    program: str,
    inp_obj: Union[Any, list[Any]],
    *,
    collect_stdout: bool = True,
    collect_files: bool = False,
    collect_wfn: bool = False,
    rm_scratch_dir: bool = True,
    propagate_wfn: bool = False,
    queue: Optional[str] = None,
    return_future: bool = False,
) -> Union[
    ProgramOutput, list[ProgramOutput], FutureOutput
]

Asynchronously submit a computation to ChemCloud.

Parameters:

Name	Type	Description	Default
program	str	A program name matching one of the self.supported_programs	required
inp_obj	Union[Any, list[Any]]	The input object to be used for the computation. This can be a single input object or a list of input objects.	required
collect_stdout	bool	Whether to collect stdout/stderr from the program as output. Failed computations will always collect stdout/stderr.	True
collect_files	bool	Collect all files generated by the QC program as output.	False
collect_wfn	bool	Collect the wavefunction file(s) from the calculation. Not every program will support this. Use collect_files to collect all files including the wavefunction.	False
rm_scratch_dir	bool	Delete the scratch directory after the program exits. Should only be set to False for debugging purposes.	True
propagate_wfn	bool	For any adapter performing a sequential task, such as a geometry optimization, propagate the wavefunction from the previous step to the next step. This is useful for accelerating convergence by using a previously computed wavefunction as a starting guess. This will be ignored if the adapter for a given qc program does not support it.	False
queue	Optional[str]	The name of a private compute queue. If None, default queue is used from settings.	None
return_future	bool	If True, return a FutureOutput object. If False, block and return the ProgramOutput object(s) directly.	False

Returns:

Type	Description
Union[ProgramOutput, list[ProgramOutput], FutureOutput]	Object providing access to a computation's eventual result. You can check a
Union[ProgramOutput, list[ProgramOutput], FutureOutput]	computation's status by running .status on the FutureOutput object or
Union[ProgramOutput, list[ProgramOutput], FutureOutput]	.get() to block and retrieve the computation's final result.

Source code in chemcloud/client.py

async def compute_async(
    self,
    program: str,
    inp_obj: Union[Any, list[Any]],
    *,
    collect_stdout: bool = True,
    collect_files: bool = False,
    collect_wfn: bool = False,
    rm_scratch_dir: bool = True,
    propagate_wfn: bool = False,
    queue: Optional[str] = None,
    return_future: bool = False,
) -> Union[ProgramOutput, list[ProgramOutput], FutureOutput]:
    """Asynchronously submit a computation to ChemCloud.

    Parameters:
        program: A program name matching one of the self.supported_programs
        inp_obj: The input object to be used for the computation. This can be a
            single input object or a list of input objects.
        collect_stdout: Whether to collect stdout/stderr from the program as output.
            Failed computations will always collect stdout/stderr.
        collect_files: Collect all files generated by the QC program as output.
        collect_wfn: Collect the wavefunction file(s) from the calculation.
            Not every program will support this. Use collect_files to collect
            all files including the wavefunction.
        rm_scratch_dir: Delete the scratch directory after the program exits. Should
            only be set to False for debugging purposes.
        propagate_wfn: For any adapter performing a sequential task, such
            as a geometry optimization, propagate the wavefunction from the previous
            step to the next step. This is useful for accelerating convergence by
            using a previously computed wavefunction as a starting guess. This will
            be ignored if the adapter for a given qc program does not support it.
        queue: The name of a private compute queue. If None, default queue is used
            from settings.
        return_future: If True, return a FutureOutput object. If False, block and
            return the ProgramOutput object(s) directly.

    Returns:
        Object providing access to a computation's eventual result. You can check a
        computation's status by running .status on the FutureOutput object or
        .get() to block and retrieve the computation's final result.
    """
    if not inp_obj:
        raise ValueError("Please provide input objects for the computation.")

    logger.info(
        f"Submitting compute job for program {program} with inputs {inp_obj}."
    )
    supported_programs = await self.supported_programs_async()
    if program not in supported_programs:
        raise UnsupportedProgramError(
            f"Please use one of the following programs: {supported_programs}"
        )
    url_params = {
        "program": program,
        "collect_stdout": collect_stdout,
        "collect_files": collect_files,
        "collect_wfn": collect_wfn,
        "rm_scratch_dir": rm_scratch_dir,
        "propagate_wfn": propagate_wfn,
        "queue": queue or self.queue or self._settings.chemcloud_queue,
    }

    # Normalize inputs to a list.
    inp_list = [inp_obj] if not isinstance(inp_obj, list) else inp_obj

    # Create a list of coroutines to submit compute requests.
    coroutines = [
        self._http_client._authenticated_request_async(
            "post", "/compute", data=inp, params=url_params
        )
        for inp in inp_list
    ]
    # Use asyncio.gather to run them concurrently.
    task_ids = await asyncio.gather(*coroutines)

    future = FutureOutput(
        task_ids=task_ids,
        inputs=inp_list,
        program=program,
        client=self,
        return_single_output=not isinstance(inp_obj, list),
    )
    if return_future:
        return future
    return await future.get_async()

compute

compute(
    *args, **kwargs
) -> Union[
    ProgramOutput, list[ProgramOutput], FutureOutput
]

Synchronous wrapper for compute_async.

Source code in chemcloud/client.py

def compute(
    self, *args, **kwargs
) -> Union[ProgramOutput, list[ProgramOutput], FutureOutput]:
    """Synchronous wrapper for compute_async."""
    return self.run(self.compute_async(*args, **kwargs))

fetch_output_async async

fetch_output_async(
    task_id: str, delete: bool = True
) -> tuple[TaskStatus, Optional[ProgramOutput]]

Get the status and output (if it is complete) of a task.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to check.	required
delete	bool	Whether to delete the output from the server after fetching.	True

Returns:

Type	Description
tuple[TaskStatus, Optional[ProgramOutput]]	A tuple of the task status and the output object if available.

Source code in chemcloud/client.py

async def fetch_output_async(
    self, task_id: str, delete: bool = True
) -> tuple[TaskStatus, Optional[ProgramOutput]]:
    """
    Get the status and output (if it is complete) of a task.

    Parameters:
        task_id: The ID of the task to check.
        delete: Whether to delete the output from the server after fetching.

    Returns:
        A tuple of the task status and the output object if available.
    """
    response = await self._http_client._authenticated_request_async(
        "get", f"/compute/output/{task_id}"
    )
    status = TaskStatus(response.get("status", TaskStatus.PENDING))
    output = response.get("program_output")
    if output is not None:
        output = ProgramOutput(**output)
    if status in READY_STATES and delete:
        # Fire-and-forget the deletion task
        asyncio.create_task(self.delete_output_async(task_id))
    return status, output

fetch_output

fetch_output(
    task_id: str,
) -> tuple[
    TaskStatus,
    Optional[Union[ProgramOutput, list[ProgramOutput]]],
]

Sync wrapper for fetch_output_async.

Source code in chemcloud/client.py

def fetch_output(
    self, task_id: str
) -> tuple[TaskStatus, Optional[Union[ProgramOutput, list[ProgramOutput]]]]:
    """Sync wrapper for `fetch_output_async`."""
    return self.run(self.fetch_output_async(task_id))

delete_output_async async

delete_output_async(task_id: str) -> None

Delete a task's output from the ChemCloud server.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to delete.	required

Source code in chemcloud/client.py

async def delete_output_async(self, task_id: str) -> None:
    """
    Delete a task's output from the ChemCloud server.

    Parameters:
        task_id: The ID of the task to delete.
    """
    logger.debug(f"Deleting output for task {task_id}")
    await self._http_client._authenticated_request_async(
        "delete", f"/compute/output/{task_id}"
    )
    logger.debug(f"Output deleted for task {task_id}")

delete_output

delete_output(task_id: str) -> None

Sync wrapper for delete_output_async.

Source code in chemcloud/client.py

def delete_output(self, task_id: str) -> None:
    """Sync wrapper for `delete_output_async`."""
    return self.run(self.delete_output_async(task_id))

run

run(coro: Coroutine[Any, Any, Any]) -> Any

Synchronous runner for async methods. Akin to asyncio.run() but centralized on CCClient since we have to modify the ._http_client._async_client and semaphore.

Source code in chemcloud/client.py

def run(self, coro: Coroutine[Any, Any, Any]) -> Any:
    """
    Synchronous runner for async methods. Akin to asyncio.run() but centralized
    on CCClient since we have to modify the ._http_client._async_client and
    semaphore.
    """
    return asyncio.run(self._run_helper(coro))

hello_world

hello_world(name: Optional[str] = None) -> str

A simple endpoint to check connectivity to ChemCloud.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Your name	None

Returns:

Type	Description
str	A message from ChemCloud if the client was able to successfully
str	connect.

Source code in chemcloud/client.py

def hello_world(self, name: Optional[str] = None) -> str:
    """A simple endpoint to check connectivity to ChemCloud.

    Parameters:
        name: Your name

    Returns:
        A message from ChemCloud if the client was able to successfully
        connect.
    """
    logger.info(f"Sending hello_world request with name: {name}")

    # Run a single asynchronous request synchronously via run_parallel_requests.
    return self.run(
        self._http_client._request_async(
            "get", "/hello-world", params={"name": name}, api_call=False
        )
    )

openapi_spec_async async

openapi_spec_async() -> dict[str, Any]

Asynchronously retrieves and caches the OpenAPI specification from the ChemCloud server.

Source code in chemcloud/client.py

async def openapi_spec_async(self) -> dict[str, Any]:
    """
    Asynchronously retrieves and caches the OpenAPI specification from the ChemCloud server.
    """
    if self._openapi_spec is None:
        result = await self._http_client._request_async(
            "get", "/openapi.json", api_call=False
        )
        self._openapi_spec = result
    return self._openapi_spec

supported_programs_async async

supported_programs_async() -> list[str]

Asynchronously returns the list of supported programs from the OpenAPI specification.

Source code in chemcloud/client.py

async def supported_programs_async(self) -> list[str]:
    """
    Asynchronously returns the list of supported programs from the OpenAPI specification.
    """
    spec = await self.openapi_spec_async()
    try:
        programs = spec["components"]["schemas"]["SupportedPrograms"]["enum"]
    except (KeyError, IndexError):
        logger.warning("Cannot locate currently supported programs.")
        programs = [""]
    return programs

setup_profile

setup_profile(profile: Optional[str] = None) -> None

Setup profiles for authentication with ChemCloud.

Parameters:

Name	Type	Description	Default
profile	Optional[str]	Optional value to create a named profile for use with QC Cloud. No value needs to be passed and most users will only have one login with ChemCloud. CCClient will access the profile by default without a specific name being passed. Pass a value if you have multiple logins to ChemCloud.	None

Note: Configures chemcloud to use the passed credentials automatically in the future. You only need to run this method once per profile. Credentials will be loaded automatically from the credentials file in the future.

Source code in chemcloud/client.py

def setup_profile(self, profile: Optional[str] = None) -> None:
    """Setup profiles for authentication with ChemCloud.

    Parameters:
        profile: Optional value to create a named profile for use with QC
            Cloud. No value needs to be passed and most users will only have one
            login with ChemCloud. CCClient will access the profile by
            default without a specific name being passed. Pass a value if you have
            multiple logins to ChemCloud.
    Note:
        Configures `chemcloud` to use the passed credentials automatically in the
        future. You only need to run this method once per profile. Credentials will
        be loaded automatically from the credentials file in the future.
    """
    profile = profile or self._settings.chemcloud_credentials_profile
    print(
        f"✅ If you don't have an account, please signup at: {self._http_client._chemcloud_domain}/signup"
    )
    # Use the async _set_tokens_from_user_input wrapped via run_parallel_requests
    access_token, refresh_token = self.run(
        self._http_client._set_tokens_from_user_input()
    )
    self._http_client.write_tokens_to_credentials_file(
        access_token, refresh_token, profile=profile
    )
    print(
        f"'{profile}' profile configured! Username/password not required for future use."
    )