PineXQ access from ProCon

When running in a PineXQ environment ProCon can acquire a pinexq-client to interact with PineXQ, e.g. to create subjobs.

Get the current job

To get a ready-to-use Job class (pinexq-client) for the current executed function call get_job(context: ExecutionContextType) -> Job

e.g. job = get_job(self.step_context).

Get API EntryPoint

For more elaborate use of the PineXQ API you can get a preconfigured EntryPointHco by calling get_entrypoint_hco(context: ExecutionContextType) -> EntryPointHco

e.g. entrypoint: EntryPointHco = get_entrypoint_hco(self.step_context)

Get a configured client

To get a fully configured HTTPX client to access the API, call:

get_client(exec_context: ExecutionContextType | None = None) -> httpx.Client

e.g. client = get_client(self.step_context)

This will retrieve required security headers from the PineXQ job offer (for the executing user) for remote integration.

If the remote feature is not active then the required information is read from the environment variables:

• JMC_API_HOST_URL: The base URL of the PineXQ JobManagement API, without entry point.
• JMC_API_KEY: (header, optional) The API-key you get from the login portal.
• JMC_ACCESS_TOKEN: (header, optional) An access token provided by PineXQ.

Access to given Grants

For permission checks in ProCon, you can request the list of grants for the job’s user by calling `get_grants(context: ExecutionContextType) -> list[str]“ These grants will only be available if Access Token security is used.

e.g. grants = get_grants(self.step_context)

Example

from pinexq.procon.jobmanagement.api_helpers import get_client, job_from_step_context
from pinexq.client.job_management.hcos.entrypoint_hco import EntryPointHco
from pinexq.client.job_management.enterjma import enter_jma
from procon.step import version, Step
# ...

class MySteps(Step):
  # ...
  @version("0.1.0-dev1")
  def my_function(self) -> int
    client = get_client(self.step_context)
    entrypoint: EntryPointHco = enter_jma(client) # access PineXQ entry point

    current_job = job_from_step_context(self, client)
    state = current_job.get_state() # access the current job's state
    #...