API reference#

class MIDataflowIntegration(use_https=True, verify_ssl=True, certificate_file=None)#

Represents a MI Data Flow step at the point at which the Python script is triggered.

When this class is instantiated, it parses the data provided by MI Data Flow, enabling Granta MI API client sessions to be created.

Parameters:
use_httpsbool, default True

Whether to use HTTPS if supported by the Granta MI application server.

verify_sslbool, default True

Whether to verify the SSL certificate CA. Has no effect if use_https is set to False.

certificate_filestr | pathlib.Path | None, default None

The CA certificate file, provided as either a string or pathlib.Path object. This paraemter can be provided in the following ways:

  • The filename of the certificate provided as a string. In this case, the certificate must be added to the workflow definition as a supporting file.

  • The filename or relative path of the certificate provided as a pathlib.Path object. In this case, the certificate must be added to the workflow definition as a supporting file.

  • The absolute path to the certificate. In this case, the certificate can be stored anywhere on disk, but it is recommended to store it in a location that will not be modified between workflows.

  • None. In this case, the certifi public CA store will be used.

If specified, the certificate will be used to verify PyGranta and MI Data Flow requests. Has no effect if use_https or verify_ssl are set to False.

Raises:
json.JSONDecodeError

If the string read from stdin is invalid JSON.

KeyError

If the JSON read from stdin does not conform to the correct data structure.

Warns:
UserWarning

If use_https is set to True and the server does not support HTTPS.

Notes

When a workflow is configured to call a Python script, the workflow execution will be suspended whilst the Python script executes. To enable the workflow to continue, call the resume_bookmark method.

Examples

If HTTPS is not configured on the server, disable HTTPS.

>>> data_flow = MIDataflowIntegration(use_https=False)

If HTTPS is configured on the server with an internal certificate and the private CA certificate is not available, either disable HTTPS or disable certificate verification.

>>> data_flow = MIDataflowIntegration(use_https=False)
>>> data_flow = MIDataflowIntegration(use_https=True, verify_ssl=False)

If HTTPS is configured on the server with an internal certificate and the private CA certificate is available, provide the private CA certificate to use this certificate for verification. If the filename only is provided, then the certificate must be added to the workflow definition file in Data Flow Designer.

>>> data_flow = MIDataflowIntegration(certificate_file="my_cert.crt")

If the certificate is stored somewhere else on disk, it can be specified by using a pathlib.Path object. In this case, the certificate should not be added to the workflow definition file in Data Flow Designer.

>>> cert = pathlib.Path(r"C:\dataflow_files\certificates\my_cert.crt")
>>> data_flow = MIDataflowIntegration(certificate_file=cert)

If HTTPS is configured on the server with a public certificate, use the default configuration to enable HTTPS and certificate verification against public CAs.

>>> data_flow = MIDataflowIntegration()
classmethod from_dict_payload(dataflow_payload, **kwargs)#

Instantiate an MIDataflowIntegration object with a static payload provided as a Python dictionary.

Can be used for testing purposes to avoid needing to trigger the Python script from within Data Flow. See get_payload_as_dict() for information on generating a suitable payload.

Parameters:
dataflow_payloadDict[str, Any]

A Python dictionary containing a copy of a Data Flow data payload used for testing purposes.

**kwargs

Additional keyword arguments are passed to the MIDataflowIntegration constructor.

Returns:
MIDataflowIntegration

The instantiated class.

Examples

>>> dataflow_payload = {"WorkflowId": "67eb55ff-363a-42c7-9793-df363f1ecc83", ...: ...}
>>> df = MIDataflowIntegration.from_dict_payload(dataflow_payload)

Additional parameters are passed through to the MIDataflowIntegration constructor

>>> dataflow_payload = {"WorkflowId": "67eb55ff-363a-42c7-9793-df363f1ecc83", ...: ...}
>>> df = MIDataflowIntegration.from_dict_payload(dataflow_payload, verify_ssl=False)
classmethod from_string_payload(dataflow_payload, **kwargs)#

Instantiate an MIDataflowIntegration object with a static payload.

provided as a JSON formatted string.

Can be used for testing purposes to avoid needing to trigger the Python script from within Data Flow. See get_payload_as_string() for information on generating a suitable payload.

Parameters:
dataflow_payloadstr

A JSON-formatted static copy of a Data Flow data payload used for testing purposes.

**kwargs

Additional keyword arguments are passed to the MIDataflowIntegration constructor.

Returns:
MIDataflowIntegration

The instantiated class.

Raises:
ValueError

If the dataflow_payload argument is not valid JSON.

Examples

>>> dataflow_payload = '{"WorkflowId": "67eb55ff-363a-42c7-9793-df363f1ecc83", ...: ...}'
>>> df = MIDataflowIntegration.from_string_payload(dataflow_payload)

Additional parameters are passed through to the MIDataflowIntegration constructor

>>> dataflow_payload = '{"WorkflowId": "67eb55ff-363a-42c7-9793-df363f1ecc83", ...: ...}'
>>> df = MIDataflowIntegration.from_string_payload(dataflow_payload, verify_ssl=False)
get_payload_as_dict(include_credentials=False)#

Get the payload used to instantiate this class as a Python dictionary.

This can be stored and provided to the from_dict_payload() method to test independently of MI Data Flow.

Parameters:
include_credentialsbool, default False

Whether to include the Basic or OIDC token header in the result.

Returns:
str

A static copy of a Data Flow data payload used for testing purposes.

Notes

By default the basic and OIDC authentication header AuthorizationHeader is replaced with the string "<HeaderRemoved>" to avoid leaking credentials. To construct the appropriate header manually:

  • For basic authentication, combine the username and password with a colon (:), Base64 encode the resulting string, and then prepend the result with “Basic “. For example, for the username Alice and password s3cr3t, these are combined to give "Alice:s3cr3t" and Base64 encoded to "QWxpY2U6czNjcjN0", which gives the final AuthorizationHeader value of "Basic QWxpY2U6czNjcjN0".

  • For OIDC authentication, generate a valid access token and prepend with "Bearer ". For example, for the token gaUDsgUrOiJSUzI, the final AuthorizationHeader value would be "Bearer gaUDsgUrOiJSUzI".

Alternatively, you can invoke this method with include_credentials=True, however you must ensure that the result is stored securely to avoid leaking credentials.

get_payload_as_string(indent=False, **kwargs)#

Get the payload used to instantiate this class and serialize to a JSON string.

This can be stored and provided to the from_string_payload() method to test independently of MI Data Flow.

This method uses the MIDataflowIntegration.get_payload_as_dict() method to prepare the dictionary. See the MIDataflowIntegration.get_payload_as_dict() documentation for more details and additional keyword arguments.

Parameters:
indentbool, default False

Whether to indent the JSON representation of the payload. Useful if displaying the result.

**kwargs

Additional keyword arguments are passed to the MIDataflowIntegration.get_payload_as_dict() method.

Returns:
str

A static copy of a Data Flow data payload used for testing purposes.

property service_layer_url: str#

The URL to the Granta MI service layer.

The URL scheme is set to https if both the server supports HTTPS and use_https = True was specified in the constructor. Otherwise, the URL scheme is set to http.

Returns:
str

URL to the service layer.

property mi_session: mpy.Session#

An MI Scripting Toolkit session which can be used to interact with Granta MI.

Deprecated since version v0.2: This property is deprecated. Use get_scripting_toolkit_session() instead.

Requires a supported version of MI Scripting Toolkit to be installed.

Returns:
mpy.Session

MI Scripting Toolkit session.

Raises:
MissingClientModuleException

If Scripting Toolkit cannot be imported.

get_scripting_toolkit_session(timeout=None, max_retries=None)#

Create an MI Scripting Toolkit session which can be used to interact with Granta MI.

Requires a supported version of MI Scripting Toolkit to be installed.

Parameters:
timeoutint, optional

The maximum time in milliseconds for the Scripting Toolkit session to wait for a response from Granta MI. See the Scripting Toolkit documentation for default behavior.

max_retriesint, optional

The maximum number of times for the Scripting Toolkit to retry a request before failing. See the Scripting Toolkit documentation for default behavior.

Returns:
mpy.Session

MI Scripting Toolkit session.

Raises:
MissingClientModuleException

If Scripting Toolkit cannot be imported.

property supporting_files_dir: Path#

The directory containing the supporting files added to the workflow definition.

Will always include the script executed by the workflow, but may contain additional scripts, CA certificates, and any other files as required by the business logic.

Returns:
pathlib.Path

The directory containing supporting files added to the workflow definition.

configure_pygranta_connection(pygranta_connection_class, session_configuration=<ansys.openapi.common._util.SessionConfiguration object>)#

Configure a PyGranta connection object with credentials provided by Data Flow.

Parameters:
pygranta_connection_classType[PyGranta_Connection_Class]

The Connection class to use to create the client object. Must be a class, not an instance of a class. Must be a PyGranta connection class, which is defined as a subclass of the base ApiClientFactory class.

session_configurationSessionConfiguration, optional

Configure the connection to the Granta MI server. The SessionConfiguration arguments verify_ssl and cert_store_path are overridden based on the values specified when instantiating this class.

Returns:
PyGranta_Connection_Class

A configured Connection object corresponding to the provided class. Call the .connect() method to finalize the connection.

Raises:
TypeError

If the class provided to this method is not a subclass of SessionConfiguration.

Examples

>>> from ansys.grantami.jobqueue import Connection
>>> data_flow = MIDataflowIntegration()
>>> connection = data_flow.configure_pygranta_connection(Connection)
>>> client = connection.connect()
>>> client
<JobQueueApiClient: url: http://my_mi_server/mi_servicelayer>
resume_bookmark(exit_code)#

Call the Data Flow API to allow the MI Data Flow step to continue.

Parameters:
exit_codestr | int

An exit code to inform Data Flow of success or otherwise of the business logic script.

class MissingClientModuleException#

Raised when a client API module is expected but could not be imported.

On this page