Skip to content

Server-Side Python API Reference

Introducing a new Python-based backend engine designed to elevate the functionality and capabilities of our application. In terms of custom RMP function this engine is feature-equivalent to the older Freemarker engine. However, developers can use the more powerful Python language compared to the more limited scope of Freemarker.

This section will walk you through the essential steps and concepts, and help you streamline your workflow, whether you're looking to automate tasks, extend functionality, or solve complex problems. ecated.

Limitations

  • Scripts executed Python are executed in a resource-isolated environment which is more strict in terms of time and memory limits.
  • IO operations are not permitted in the Python sandbox.
  • You cannot add external libraries to the execution environment.
  • There may be an initial delay of ca. 5 secs when executing a python script. This should be rare and is related to auto-scaling.
  • Multi-account users (e.g. third-party support users) are not supported.
  • The Python engine supports the pure Python 3.11 language scope.

Basic Script Setup

The new engine can be triggered when specifying a script like this (e.g. as input/output parameters of a process):


<@script env="python">
... python code here ...
</@script>

Please note that compared to the existing engine, you have to specify <@script env="python"> for Python rather than <@script env="js"> used by the Javascript engine.

Besides the standard Javascript language, there are a number of RMP Objects which provide access to custom functions. e.g. for creating a custom MongoDB collection, you'd choose the RMPCollection class with the corresponding function name like: 


<@script env="python">
def someResponse = RMPCollection.createCollection("some collection name", false, false);
</@script>

Broadly, these custom functions resemble those from the Freemarker environment.

Response Handling

The response object has to be an Object that can be translated into a String. Namely:

  • String
  • Number
  • Boolean
  • Map (JSON Object)
  • Array (JSON Array)

You can specify the response explicitly via the custom function below. All output objects will be transformed into text format.


<@script env="python">
RMPData.setOutput(someResponseObject);
</@script>

Error Handling

The functions will produce an exception with a reasonable error message, and the execution fails.

Saving Data into the Process Context

  
    <@script env="python">
      ...
      RMPData.setContextVariable(someKey, someValue);
      ...
    </@script>

Python vs Freemarker

Freemarker:

  
    <#assign result = append_file("73f01850-1a85-11ec-8413-8c859067f180","coucou")>
    ${result}

So you use the <#assign x=y> for assigning a value to a variable.

${x} is used to return the variable value to the workflow / app.

You may also spot the following variation:

  
    <#assign result>
      append_file("73f01850-1a85-11ec-8413-8c859067f180","coucou")
    </#assign>
    ${result}

Python(see also Response Handling):

  
    <@script env="python">
      def result = RMPFile.append("73f01850-1a85-11ec-8413-8c859067f180","coucou");
      RMPData.setOutput(result);
    </@script>

var x = y; is a standard JS value assignment.

User RMPData.setOutput(x); for returning the variable value to the workflow / app.

Python Reference Guide (External)

  • Python 3 Reference

    • RunMyProcess-specific Custom Functions for Workflows

    Please refer to the following pages for the function documentation:

    RMPApplication

    class RMPApplication()

    Namespace for RMPApplication.

    attachFile

    @staticmethod
def attachFile(file_ids: List[str], widget_id: str) -> List[Dict]

    Attaches files to an application.

    Arguments:

    • file_ids List[Dict] - A list of file identifiers.
    • widget_id str - The identifier of the uploaded file widget to which the files should be attached.

    Returns:

    List[Dict] A list of descriptors of the attached files.

    Example:

    file_ids = ["fd58e98b-447e-42f8-8a78-24b55d928f3e"]
result = RMPApplication.attachFile(file_ids, "uploadFile_1")
RMPData.setOutput(result);

    detachFile

    @staticmethod
def detachFile(file_ids: List[str], instance_id: str) -> List[Dict]

    Detaches one or more files from an application instance.

    Arguments:

    • file_ids - A list of file identifiers.
    • instance_id - The application instance from which the files should be detached.

    Returns:

    • List[Dict] - A list of descriptors of the detached files.

    Example:

    file_ids = ["fd58e98b-447e-42f8-8a78-24b55d928f3e"]
result = RMPApplication.detachFile(file_ids, instance_id)
RMPData.setOutput(result);

    listAttachedFiles

    @staticmethod
def listAttachedFiles(instance_id: str) -> List[Dict]

    Lists files attached to an application instance.

    Arguments:

    • instance_id - The identifier of the application instance.

    Returns:

    • List[Dict] - A list of attached file descriptors.

    Example:

    result = RMPApplication.listAttachedFiles("4ab793b0-6157-49b7-b95f-3112360a8f9a")
RMPData.setOutput(result);

    getHistory

    @staticmethod
def getHistory() -> List[Dict]

    Retrieves the modification history of an application instance.

    Returns:

    • List[Dict] - A list of metadata describing historical changes.

    Example:

    history = RMPApplication.getHistory()
RMPData.setOutput(history)

    getInstanceInfo

    @staticmethod
def getInstanceInfo(instance_id: str) -> dict

    Retrieves metadata for an application instance.

    Arguments:

    • instance_id - The identifier of the application instance.

    Returns:

    • dict - A dictionary containing application metadata.

    Example:

    info = RMPApplication.getInstanceInfo("2518ba82-b617-4e5b-8673-6ba414b19203")
RMPData.setOutput(info)

    getI18n

    @staticmethod
def getI18n(key: str,
            default_value: Optional[str] = "",
            language: Optional[str] = "en") -> str

    Retrieves a translated string based on a given key and language.

    Arguments:

    • key - The text key to translate.
    • default_value - (Optional) The default value if no translation is found.
    • language - (Optional) The language code (two-letter format).

    Returns:

    The translated string or the default value if no translation is found.

    Example:

    result = RMPApplication.getI18n("Text", "defaultVal")
RMPData.setOutput(result);

    RMPCollection

    class RMPCollection()

    Namespace for RMCollection

    createCollection

    @staticmethod
def createCollection(collection_name: str, is_public: bool,
                     is_readonly: bool) -> bool

    Creates a new Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • is_public - Whether Mongo collection can be accessed publicly, without checking rights.
    • is_readonly - Whether the Mongo collection can only be modified by an administrator of the customer account.

    Returns:

    • bool - Collection successfully created or not.

    Examples:

    result1 = RMPCollection.createCollection('private_collection', False, True)
RMPData.setOutput(result1)
result2 = RMPCollection.createCollection('public_collection', True, False)
RMPData.setOutput(result2)

    drop

    @staticmethod
def drop(collection_name: str) -> bool

    Drops a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.

    Returns:

    • bool - Collection successfully removed or not.

    Examples:

    f = RMPCollection.drop('col_rms_v1_new')
RMPData.setOutput(f)

    renameCollection

    @staticmethod
def renameCollection(old_collection_name: str,
                     new_collection_name: str) -> bool

    Renames a Mongo collection.

    Arguments:

    • old_collection_name - Current name of the Mongo collection.
    • new_collection_name - New name of the Mongo collection.

    Returns:

    • bool - Renaming successful or not.

    Examples:

    is_success = RMPCollection.renameCollection('col_rms_v1', 'col_rms_v1_new')
RMPData.setOutput(is_success)

    createIndex

    @staticmethod
def createIndex(collection_name: str,
                index_name: str,
                index_array: List[str],
                unique: Optional[bool] = False,
                ascending: Optional[bool] = False) -> bool

    Creates indexes on Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • index_name - Name of the Index.
    • index_array - Index Array.
    • unique - Optional: Whether the index shall observe uniqueness.
    • ascending - Optional: The ordering.

    Returns:

    • bool - Creation of the index was successful or not.

    Examples:

    idx = ["attr1","attr2"]
is_success = RMPCollection.createIndex('rms_car', 'rms_car_idx', idx)
RMPData.setOutput(is_success)
is_success = RMPCollection.createIndex('rms_car', 'rms_car_idx', idx, True)
RMPData.setOutput(is_success)
is_success = RMPCollection.createIndex('rms_car', 'rms_car_idx', idx, False, False)
RMPData.setOutput(is_success)

    dropIndex

    @staticmethod
def dropIndex(collection_name: str, index_name: str) -> bool

    Drops an index from a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • index_name - Name of the Index.

    Returns:

    • bool - Creation of the index was dropped successful or not.

    Examples:

    f = RMPCollection.dropIndex('rms_car', 'rms_car_idx')
RMPData.setOutput(f)

    listIndexes

    @staticmethod
def listIndexes(collection_name: str) -> List[Dict]

    List of all indexes on a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.

    Returns:

    • List[Dict] - The list of indexes of a collection containing name, keys, unique and ascending fields.

    Examples:

    f = RMPCollection.listIndexes('rms_car')
RMPData.setOutput(f)

    aggregate

    @staticmethod
def aggregate(collection_name: str, pipeline: List[Dict]) -> List[Dict]

    Runs an aggregation query over a Mongo collection.

    Note: Please make sure that you don't exceed the maximum memory limit when using the function.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • pipeline - The aggregation pipeline in MongoDB BSON format.

    Returns:

    • List[Dict] - JSON result of the aggregate operation.

    Examples:

    aggr = [
        {'$match': { 'brand_id': 'bogdan_group' } },
    {'$project': { 'nameStr': { '$convert': { 'input': '$name', 'to': 'string' } } } },
        {'$group': { '_id': '$nameStr', 'totaldocs': { '$sum': 1 } } }
];
result = RMPCollection.aggregate('Cars_info', aggr );
for doc in result:
    _id = doc.get("_id")
    if isinstance(_id, dict) and "$oid" in _id:
        doc["name"] = _id["$oid"]
    else:
        doc["name"] = _id
    del doc["_id"]
RMPData.setOutput(result);

    See Also:

    Details about the aggregation definition

    find

    @staticmethod
def find(collection_name: str,
         query: Dict,
         offset: Optional[int] = 0,
         limit: Optional[int] = 100) -> List[Dict]

    Retrieve objects from a Mongo collection.

    Note: Please make sure that you don't exceed the maximum memory limit when using the function.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • query - The actual query. "$query": query in MongoDB BSON object format.
    • "$orderby" - order by attributes in JSON array format.
    • offset - Optional, Number of leading items to be skipped.
    • limit - Optional, Max number of items returned.
    • projectionKeys - Optional, Attributes included in the query.
    • first - Optional, Attribute to sort, ASC or DESC.
    • removeId - Optional, Remove Id attribute in the query.

    Returns:

    List of objects.

    Examples:

    1.RMPCollection.find(collectionName, query)

    f = RMPCollection.find('Cars_info', {})
for doc in f:
    if isinstance(doc, dict) and '_id' in doc and isinstance(doc['_id'], dict) and '$oid' in doc['_id']:
        doc['id_str'] = doc['_id']['$oid']
        del doc['_id']
RMPData.setOutput(f)

    2.RMPCollection.find(collectionName, query, offset)

    f = RMPCollection.find('Cars_info', {}, 1)
for doc in f:
    if isinstance(doc, dict) and '_id' in doc and isinstance(doc['_id'], dict) and '$oid' in doc['_id']:
        doc['id_str'] = doc['_id']['$oid']
        del doc['_id']
RMPData.setOutput(f)

    3. RMPCollection.find(collectionName, query, offset, limit)

    f = RMPCollection.find('Cars_info', {}, 1, 1)
for doc in f:
    if isinstance(doc, dict) and '_id' in doc and isinstance(doc['_id'], dict) and '$oid' in doc['_id']:
        doc['id_str'] = doc['_id']['$oid']
        del doc['_id']
RMPData.setOutput(f)

    4. RMPCollection.find(collectionName, query, projectionKeys, first, limit)

    f = RMPCollection.find('Cars_info', {}, ['brand_label', 'Alfa Romeo'], 0, 1)
for doc in f:
    if isinstance(doc, dict) and '_id' in doc and isinstance(doc['_id'], dict) and '$oid' in doc['_id']:
        doc['id_str'] = doc['_id']['$oid']
        del doc['_id']
RMPData.setOutput(f)

    5. RMPCollection.find(collectionName, query, projectionKeys, orderBy, first, limit, removeId)

    f = RMPCollection.find('Cars_info', {}, ['brand_label', 'Alfa Romeo'], {'name': 'DESC'}, 0, 10, True)
for doc in f:
    if isinstance(doc, dict) and '_id' in doc and isinstance(doc['_id'], dict) and '$oid' in doc['_id']:
        doc['id_str'] = doc['_id']['$oid']
        del doc['_id']
RMPData.setOutput(f)

    See Also:

    Details on how to define a MongoDB query

    countDocuments

    @staticmethod
def countDocuments(collection_name: str, query: Dict) -> int

    Count documents verifying a condition in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • query - The actual query in JSON format.

    Returns:

    The number of documents matching the query

    Example:

    query = {}
f = RMPCollection.countDocuments('car_rms', query)
RMPData.setOutput(f)

    See Also:

    Details on how to define a MongoDB query

    insertOne

    @staticmethod
def insertOne(collection_name: str, doc: Dict) -> Dict

    Insert a JSON document in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • doc - The document to be inserted in JSON format

    Returns:

    • Dict - The document

    Example:

    doc = {"engine":"e1","brand_label":"Brand-1","name":"Name-1","brand_id":"b1"}
f = RMPCollection.insertOne('car_rms', doc)
RMPData.setOutput(f)

    insertMany

    @staticmethod
def insertMany(collection_name: str, docs: List[Dict]) -> List[Dict]

    Insert an array of JSON documents in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • docs - The document to be inserted in JSON format

    Returns:

    The array of documents

    Example:

    docs = [
    {
    "engine": "e2",
    "brand_label": "Brand-2",
    "name": "Name-2",
    "brand_id": "b2"
    },
    {
    "engine": "e3",
    "brand_label": "Brand-3",
    "name": "Name-3",
    "brand_id": "b3"
    }
]
f = RMPCollection.insertMany('car_rms', docs)
RMPData.setOutput(f)

    deleteMany

    @staticmethod
def deleteMany(collection_name: str, query: Dict) -> int

    Deletes documents that matches with the given query.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • query - The actual query in JSON format.

    Returns:

    The number of objects deleted

    Example:

    query = {'brand_id':'b3'}
f = RMPCollection.deleteMany('car_rms', query)
RMPData.setOutput(f)

    See Also:

    Details on how to define a MongoDB query

    updateMany

    @staticmethod
def updateMany(collection_name: str,
               upd: Dict,
               query: Dict,
               multi: Optional[bool] = False,
               update_operator: Optional[bool] = False) -> int

    Update documents with a setter according to a given query in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • upd - The setter for the document fields to be updated in BSON format.
    • query - The actual query in BSON format.
    • multi - Whether one or multiple documents shall be updated
    • update_operator - Whether to update or replace a document if multi=false

    Returns:

    • int - The number of documents updated

    Example:

    query = {'brand_id':'b2'}
upd = {'$set':{'engine':'e2','brand_label':'Brand-2','name':'Name-2'}}
f = RMPCollection.updateMany('car_rms', upd, query, True)
RMPData.setOutput(f)

    See Also:

    Details on how to define a MongoDB updates

    updateOne

    @staticmethod
def updateOne(collection_name: str, upd: dict, query: dict) -> int

    Updates the first document with a setter according to the query in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • upd - The setter for the document fields to be updated in BSON format.
    • query - The actual query in BSON format.

    Returns:

    • int - The number of documents updated

    Example:

    query = {'brand_id':'b2'}
upd = {'$set':{'engine':'e2','brand_label':'Brand-2','name':'Name-2'}}
f = RMPCollection.updateOne('car_rms', upd, query)
RMPData.setOutput(f)

    See Also:

    Details on how to define a MongoDB updates

    updateField

    @staticmethod
def updateField(collection_name: str,
                upd: dict,
                query: dict,
                multi: Optional[bool] = False) -> int

    Update fields on a subset of the documents in a Mongo collection.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • upd - The value in JSON format.
    • query - The actual query in JSON format.
    • multi - Optional, Whether one or multiple documents shall be updated

    Returns:

    • int - The number of updated field

    Example:

    query = {'brand_id':'b2'}
upd = {"engine":"engine_3"}
f = RMPCollection.updateField('car_rms', upd, query, True)
RMPData.setOutput(f)

    deleteField

    @staticmethod
def deleteField(collection_name: str,
                query: dict,
                multi: Optional[bool] = False) -> int

    Deletes one or more fields from documents that matches with the given query.

    Arguments:

    • collection_name - Name of the Mongo collection.
    • query - The actual query in JSON format.
    • multi - Optional, Whether one or multiple documents shall be updated

    Returns:

    The number of updated documents

    Example:

    query = {"name":"delete_me!"}
f = RMPCollection.deleteField('car_rms', query, 'engine')
RMPData.setOutput(f)

    importCSV

    @staticmethod
def importCSV(file_id: str,
              collection_name: str,
              options: dict,
              drop_collection: Optional[bool] = False) -> int

    Import objects to a collection from a CSV file.

    Arguments:

    • file_id - Reference of an uploaded file.
    • collection_name - Name of the Mongo collection.
    • options - Options to be used:
    • separator str - Separator between two values (e.g. ; or ,). Default: ,
    • delimiter str - Used to allow values containing a separator character (e.g. " or '). Default: "
    • empty str - Value to be used to represent an empty value. Default: ""
    • charset str - Character set to be used to read the file content. Default: UTF-8
    • parse_numbers list[str] - Names/indices of columns containing numerical values to be stored as Double.
    • parse_longs list[str] - Names/indices of columns containing long values to be stored as Long.
    • trim bool - Trim values before inserting them in the collection.
    • drop_collection - Whether to drop the existing content of a collection before the import

    Returns:

    The number of imported objects.

    Example:

    options = {}
f = RMPCollection.importCSV("30ad9a42-8a4e-48b3-b857-660979eb77dd", "myCollection", options)
RMPData.setOutput(f)

    importJSON

    @staticmethod
def importJSON(file_id: str,
               collection_name: str,
               options: dict,
               drop_collection: Optional[bool] = False) -> int

    Import objects to a Mongo collection from a JSON file.

    Arguments:

    • file_id - Reference of an uploaded file.
    • collection_name - Name of the Mongo collection.
    • options - Options to be used:
    • separator str - Separator between two values (e.g. ; or ,). Default: ,
    • delimiter str - Used to allow values containing a separator character (e.g. " or '). Default: "
    • empty str - Value to be used to represent an empty value. Default: ""
    • charset str - Character set to be used to read the file content. Default: UTF-8
    • parse_numbers list[str] - Names/indices of columns containing numerical values to be stored as Double.
    • parse_longs list[str] - Names/indices of columns containing long values to be stored as Long.
    • trim bool - Trim values before inserting them in the collection.
    • drop_collection - Whether to drop the existing content of a collection before the import

    Returns:

    The number of imported objects.

    Example:

    options = {}
f = RMPCollection.importJSON("30ad9a42-8a4e-48b3-b857-660979eb77dd", "myCollection", options)
RMPData.setOutput(f)

    RMPData

    class RMPData()

    Namespace for RMPData.

    setOutput

    @staticmethod
def setOutput(some_result: Any)

    Stores the result of a script execution in the workflow variable P_result.

    Arguments:

    • some_result - The result of the script execution, which can be structured data or a simple value.

    Example:

    RMPData.setOutput(someResult);

    User Objects

    class User(TypedDict)

    Preferences

    Represents user information.

    getCurrentUser

    @staticmethod
def getCurrentUser() -> User

    Retrieves information about the currently active user.

    Returns:

    A dictionary containing user details such as name, login, and profile.

    Example:

    user_info = RMPData.getCurrentUser()
RMPData.setOutput(user_info)

    getExecutionMode

    @staticmethod
def getExecutionMode() -> str

    Retrieves the current execution mode of the workflow.

    Returns:

    A string indicating the mode: TEST, ACCEPTANCE, or LIVE.

    Example:

    mode = RMPData.getExecutionMode()
RMPData.setOutput(mode)

    getAllContextVariables

    @staticmethod
def getAllContextVariables() -> dict

    Retrieves all variables available in the current process context.

    Returns:

    A dictionary of key-value pairs representing context variables.

    Example:

    variables = RMPData.getAllContextVariables()
RMPData.setOutput(variables)

    getContextVariable

    @staticmethod
def getContextVariable(key: str) -> Any

    Retrieves a specific variable from the process context.

    Arguments:

    • key - The name of the variable to retrieve.

    Returns:

    The value associated with the given key.

    Example:

    variable_value = RMPData.getContextVariable("theKey")
RMPData.setOutput(variable_value)

    getAppInstanceId

    @staticmethod
def getAppInstanceId() -> str

    Retrieves the identifier of the app instance mapped to the current request.

    Returns:

    A string representing the app instance ID.

    Example:

    app_id = RMPData.getAppInstanceId()
RMPData.setOutput(app_id)

    getProjectId

    @staticmethod
def getProjectId() -> int

    Retrieves the project ID for the current request.

    Returns:

    An integer representing the project ID.

    Example:

    project_id = RMPData.getProjectId()
RMPData.setOutput(projectId)

    log

    @staticmethod
def log(log_message: str) -> bool

    Sends a custom log message to the RunMyProcess logging tool.

    Arguments:

    • log_message - The message to log.

    Returns:

    True if the log was successfully recorded, otherwise False.

    Example:

    RMPData.log("Custom log message")

    log(with level)

    @staticmethod
def log(level: str, log_message: str) -> bool

    Sends a log message with a specific severity level to RunMyProcess.

    Arguments:

    • level - Severity level (SEVERE, WARNING, INFO, CONFIG, FINE, FINER, FINEST).
    • log_message - The message to be logged (max 3 KB).

    Returns:

    True if the log was successfully recorded, otherwise False.

    Example:

    RMPData.log("WARNING", "Potential issue detected")

    raiseEvent

    @staticmethod
def raiseEvent(custom_key, custom_message, is_failure, custom_data)

    Sends a custom event to the Event Management tool.

    Arguments:

    • custom_key - The name of the event.
    • custom_message - A message describing the event.
    • is_failure - Whether the event indicates a failure.
    • custom_data - Additional event-related data.

    Returns:

    True if the event was successfully raised, otherwise False.

    Example:

    event_data = {"error": "Timeout"}
RMPData.raiseEvent("ProcessFailure", "Timeout occurred", True, event_data)

    setContextVariable

    @staticmethod
def setContextVariable(key, value)

    Add a variable with a key to the context of a process request.

    Arguments:

    • key - Key to Object.
    • value - Object to be set to the context.

    Example:

    RMPData.setContextVariable("var2","20")

    RMPFile

    class RMPFile()

    Namespace for RMPFile operations.

    create

    @staticmethod
def create(name: str, content: str) -> str

    Create a new file. Works only for uploaded files.

    Arguments:

    • name str - Name of the file.
    • content str - The data to be added into the file.

    Returns:

    • str - Identifier of the created file.

    Example:

    createFile = RMPFile.create('test2.txt', 'abcdef')
RMPData.setOutput(createFile)

    delete

    @staticmethod
def delete(file_id: str) -> bool

    Delete an existing file. Works only for uploaded files.

    Arguments:

    • file_id str - Identifier of the file to be deleted.

    Returns:

    • bool - The result of the delete operation (True, False).

    Example:

    file_id = "54fedbb0-b9ae-11ed-b944-d68b202471ff"
RMPFile.delete(file_id)

    getContent

    @staticmethod
def getContent(file_id: str,
               encoding: Optional[str] = "None",
               charset: Optional[str] = "UTF-8") -> str

    Read the content of a file.

    Arguments:

    • file_id str - Identifier of the file.
    • encoding str, optional - "BASE64" or "NONE".
    • charset str, optional - Character-set, e.g., "UTF-8".

    Returns:

    • str - The content of the file.

    Example:

    file_id = "3146acd0-b0e8-11ed-ac99-ea641a55cdec"
content = RMPFile.getContent(file_id)
RMPData.setOutput(content)

    SaveFileOptions

    class SaveFileOptions()

    charset

    Optional parameters for saving files.

    Attributes:

    • encoding str, optional - "BASE64" (default) or "NONE".
    • charset str, optional - Character set to be used to read the file content. Default: "UTF-8".

    save

    @staticmethod
def save(file_id: str, content: str,
         options: Optional[SaveFileOptions]) -> str

    Overwrite an existing file. Works only for uploaded files.

    Arguments:

    • file_id str - Identifier of the file to be overridden.
    • content str - New content of the file.
    • options SaveFileOptions - Optional parameters.

    Returns:

    • str - Identifier of the file.

    Example:

    file_id = RMPFile.create('text.txt', 'abcdef')
RMPFile.save(file_id, 'hirsch')
RMPData.setOutput(file_id)

    append

    @staticmethod
def append(file_id: str,
           content: str,
           charset: Optional[str] = "UTF-8") -> str

    Append data to an existing file. Works only for uploaded files.

    Arguments:

    • file_id str - Identifier of the file to be appended. A new file is created if 'file_id' is empty.
    • content str - The data to be added into the file.
    • charset str, optional - Character-set, e.g., "UTF-8".

    Returns:

    • str - Identifier of the appended file.

    Example:

    RMPFile.append("3146acd0-b0e8-11ed-ac99-ea641a55cdec", 'defghi')

    ReplaceInFileOptions

    class ReplaceInFileOptions()

    charset

    Optional parameters for replacing text in files.

    Attributes:

    • flag str - 'f' for first occurrence only, 'r' for all occurrences.
    • algorithm str, optional - File encoding type: "BASE64", "SHA-1", "SHA-256", "MD5", or "NONE".
    • charset str, optional - Character-set, e.g., "UTF-8".

    replaceInFile

    @staticmethod
def replaceInFile(file_id: str,
                  old_string: str,
                  replacement: str,
                  options: Optional[ReplaceInFileOptions] = {}) -> str

    Replace occurrences of a string in a file and save it. Works only for uploaded files.

    Arguments:

    • file_id str - Identifier of the file.
    • old_string str - The data to be replaced.
    • replacement str - The data to insert as replacement.
    • options ReplaceInFileOptions, optional - Optional parameters.

    Returns:

    • str - Identifier of the modified file.

    Example:

    file_id = "3146acd0-b0e8-11ed-ac99-ea641a55cdec"
RMPFile.replaceInFile(file_id, 'abc', 'efg')

    getSize

    @staticmethod
def getSize(file_id: str) -> Dict

    Get the size in KBytes of a file on the platform.

    Arguments:

    • file_id str - Identifier of the file.

    Returns:

    • Dict - Size of the file in KB along with its description.

    Example:

    file_id = "3146acd0-b0e8-11ed-ac99-ea641a55cdec"
file_size = RMPFile.getSize(file_id)
RMPData.setOutput(file_size)

    IndexOfOptions

    class IndexOfOptions()

    charset

    Optional parameters for finding string indices in a file.

    Attributes:

    • fromIndex int - Start index.
    • algorithm str, optional - File encoding type: "BASE64" or "NONE".
    • charset str, optional - Character-set, e.g., "UTF-8".

    indexOf

    @staticmethod
def indexOf(file_id: str,
            substring: str,
            options: Optional[IndexOfOptions] = {}) -> int

    Find the index of the first occurrence of a string in a file.

    Arguments:

    • file_id str - Identifier of the file to be searched.
    • substring str - Text to be matched with the file content.
    • options RMPFile.IndexOfOptions - Optional parameters.

    Returns:

    • int - Index position of the given string.

    Example:

    file_id = "3146acd0-b0e8-11ed-ac99-ea641a55cdec"
idx = RMPFile.indexOf(file_id, 'ab')
RMPData.setOutput(idx)

    getDescription

    @staticmethod
def getDescription(file_id: str) -> dict

    Get the metadata describing a file on the platform.

    Arguments:

    • file_id str - Identifier of the file.

    Returns:

    • dict - Description of the file.

    Example:

    file_id = "3146acd0-b0e8-11ed-ac99-ea641a55cdec"
description = RMPFile.getDescription(file_id)
RMPData.setOutput(description)

    FileMetadata

    class FileMetadata()

    visibility

    Metadata for describing a file.

    Attributes:

    • name str - File name.
    • type str - File type.
    • visibility str, optional - "PRIVATE" (default) or "PUBLIC".

    saveDescription

    @staticmethod
def saveDescription(file_id: str, metadata: FileMetadata) -> FileMetadata

    Modify the metadata describing a file on the platform.

    Arguments:

    • file_id str - Identifier of the file.
    • metadata FileMetadata - New metadata of the file.

    Returns:

    • FileMetadata - New metadata of the file.

    Example:

    file_id = RMPFile.create('text.txt', 'abcdef')
payload = {'desc': 'file desc'}
description = RMPFile.saveDescription(file_id, payload)
RMPData.setOutput(description)

    unzip

    @staticmethod
def unzip(file_id: str) -> List[str]

    Unzip a file and create a new file with the unzipped data. Works only for uploaded files.

    Arguments:

    • file_id str - Identifier of the file to be unzipped.

    Returns:

    • list[str] - List of unzipped file identifiers.

    Example:

    file_id = "bd98d36e-fae7-40e6-98b6-9c115305110f"
unzip_file_id = RMPFile.unzip(file_id)
RMPData.setOutput(unzip_file_id)

    zip

    @staticmethod
def zip(name: str, file_ids: List[str]) -> str

    Zip a file and create a new file with the zipped data. Works only for uploaded files.

    Arguments:

    • name str - Name for the ZIP file to be created.
    • file_ids list[str] - Identifiers of the files to be zipped.

    Returns:

    • str - ZIP File Identifier.

    Example:

    file_id = "54fedbb0-b9ae-11ed-b944-d68b202471ff"
zip_file_id = RMPFile.zip('test.zip', [file_id])
RMPData.setOutput(zip_file_id)

    readExcelSpreadsheet

    @staticmethod
def readExcelSpreadsheet(file_id: str, sheet_name: str) -> List[list]

    Read cells from an Excel worksheet. Works for XLSX format only.

    Arguments:

    • file_id str - Identifier of the Excel file uploaded on the platform.
    • sheet_name str - Worksheet descriptor listing the cells to be read.

    Returns:

    • list[list] - Content of the spreadsheet.

    Example:

    sheet = RMPFile.readExcelSpreadsheet('54cfc512-2307-494c-a5f7-2e262c3e1add', 'Sheet1')
for row in sheet:
    for item in row:
        print(item)
RMPData.setOutput(sheet)

    RMPProject

    class RMPProject()

    RMPProject contains various static methods to interact with custom lists, project vault, and OAuth2 tokens, among other functionalities.

    getCustomList

    @staticmethod
def getCustomList(list_id: str) -> List[Dict]

    Returns the content of an existing custom list.

    Arguments:

    • list_id str - Identifier of a custom list.

    Returns:

    • List - An array of custom list elements.

    Raises:

    • ValueError - If the listId is empty or invalid.

    Example:

    custom_list = RMPProject.getCustomList("b1eeebf5-d26a-4374-bf4c-03d340f17596")
RMPData.setOutput(custom_list)

    saveCustomList

    @staticmethod
def saveCustomList(list_id: str, array_of_objects: List[Dict]) -> List[Dict]

    Saves entries into an existing custom list.

    Arguments:

    • list_id str - Identifier of a custom list.
    • array_of_objects List - An array of custom list elements.

    Returns:

    • list - An array of custom list elements.

    Raises:

    • ValueError - If the custom list doesn't exist or the payload can't be parsed.

    Example:

    payload = [{"label":"ab","value":"c"},{"label":"cd","value":"d"}]
custom_list = RMPProject.saveCustomList("b1eeebf5-d26a-4374-bf4c-03d340f17596", payload)
RMPData.setOutput(custom_list)

    getProjectVault

    @staticmethod
def getProjectVault() -> Dict

    Retrieve the project vault associated with the project containing the current request. The vault can be used to store secrets.

    Returns:

    • dict - Project vault data in Map format. A project vault value depends on the current request execution mode.

    Example:

    project_vault_data = RMPProject.getProjectVault()
RMPData.setOutput(project_vault_data)

    saveProjectVault

    @staticmethod
def saveProjectVault(vault_data: Dict) -> Dict

    Save data into the project vault. The vault can be used to store secrets.

    Arguments:

    • vault_data dict - Data to be saved in the vault.

    Returns:

    • dict - The updated project vault.

    Example:

    payload = {"project":"TEST","value":"Value test"}
project_vault_data = RMPProject.saveProjectVault(payload)
RMPData.setOutput(project_vault_data)

    deleteProjectVault

    @staticmethod
def deleteProjectVault()

    Delete a project vault.

    Returns:

    • boolean - True if the project vault has been found and deleted, false otherwise.

    Example:

    f = RMPProject.deleteProjectVault()
RMPData.setOutput(f)

    getAccessUrl

    @staticmethod
def getAccessUrl(include_protocol: bool) -> str

    Returns the Live URL of the current RunMyProcess instance.

    Arguments:

    • include_protocol bool - Whether to include the protocol in the URL.

    Returns:

    • str - The regional URL to your current RMP instance.

    Example:

    live_url = RMPProject.getAccessUrl(True)
RMPData.setOutput(live_url)

    getProjectMetadata

    @staticmethod
def getProjectMetadata(project_id: Optional[int] = Any) -> Dict

    Returns the metadata of a project.

    Arguments:

    • project_id int, optional - Identifier of the project. Defaults to the current project.

    Returns:

    • dict - The project's metadata in map format.

    Example:

    project_metadata = RMPProject.getProjectMetadata(23324452)
RMPData.setOutput(project_metadata)

    saveProjectMetadata

    @staticmethod
def saveProjectMetadata(project_id: int, key: str, meta_data: Dict) -> Dict

    Saves the metadata of a project.

    Arguments:

    • project_id int - Identifier of the project.
    • key str - Identifier of the metadata entry.
    • meta_data Dict - Metadata to be saved.

    Returns:

    • Dict - Metadata in map format.

    Example:

    project_metadata = RMPProject.saveProjectMetadata("testProjectMeta", {"name": "test3"})
RMPData.setOutput(project_metadata)

    getOAuth2Token

    @staticmethod
def getOAuth2Token(key: str) -> Dict

    Returns a stored OAuth2 service token.

    Arguments:

    • key str - Name/key of the OAuth2 service whose token is to be retrieved.

    Returns:

    • Dict - The stored OAuth2 token in Map format.

    Example:

    oauth_token = RMPProject.getOAuth2Token("docuSignOauth")
RMPData.setOutput(oauth_token)

    saveOAuth2Token

    @staticmethod
def saveOAuth2Token(key: str, token: Dict) -> Dict

    Save an OAuth2 service token.

    Arguments:

    • key str - Name/key of the OAuth2 service whose token is to be saved.
    • token Dict - The OAuth2 token to be saved.

    Returns:

    • Dict - The stored OAuth2 token in Map format.

    Example:

    oauth_token = RMPProject.saveOAuth2Token("docuSignOauth", token)
RMPData.setOutput(oauth_token)

    RMPRequest

    class RMPRequest()

    Namespace for handling requests in the RMP system.

    addUserToLane

    @staticmethod
def addUserToLane(lane_id: int, user_id: int) -> bool

    Add a user to the given runtime lane.

    Arguments:

    • lane_id int - Identifier of the lane.
    • user_id int - Id or login of the user.

    Returns:

    • bool - Status if adding a user to the lane.

    Example:

    add_user_to_lane_request = RMPRequest.addUserToLane(53877772, 65914367)
RMPData.setOutput(add_user_to_lane_request)

    getNextUrl

    @staticmethod
def getNextUrl(step_id: int,
               is_public: bool,
               app_ctx: Optional[str] = Any) -> str

    Return the URL of a manual task. Provides the URL of a manual task that will be generated when reaching the step step_id. This method can be called outside a manual task's input parameters. In a loop, the returned URL will be set for the next iteration.

    Arguments:

    • step_id int - The id of the step as defined in the process design.
    • is_public bool - Use a public task template or not.
    • app_ctx str, optional - If context is 'mobile', the generated URL will be adapted to the RunMyApp environment. Default is 'web'.

    Returns:

    • str - The Task URL.

    Example:

    next_url = RMPRequest.getNextUrl(2, False, "mobile")
RMPData.setOutput(next_url)
next_url_2 = RMPRequest.getNextUrl(2, False)
RMPData.setOutput(next_url_2)

    getRequestInfo

    @staticmethod
def getRequestInfo(request_id: Optional[str] = Any) -> Dict

    Describe a process request.

    Arguments:

    • request_id str, optional - Identifier of the process request to describe. Defaults to current process request.

    Returns:

    • Dict - Process request descriptor.

    Example:

    request_data = RMPRequest.getRequestInfo("fcb7a691-38bd-4b26-b1a7-82cca9d6ecf0")
RMPData.setOutput(request_data)

    getTaskUrl

    @staticmethod
def getTaskUrl(is_public: bool) -> str

    Return the URL of the current manual task. When used within a manual task's input parameters, provides the URL of the manual task to be generated.

    Arguments:

    • is_public bool - Use a public task template or not.

    Returns:

    • str - The Task URL.

    Example:

    task_url = RMPRequest.getTaskUrl(True)
RMPData.setOutput(task_url)

    injectParams

    @staticmethod
def injectParams() -> str

    Add the result of the last executed task to the context of a process request.

    Returns:

    • str - Result of the last executed task.

    Example:

    inject_params = RMPRequest.injectParams()
RMPData.setOutput(inject_params)

    lock

    @staticmethod
def lock(lock_name: str,
         timeout_in_ms: Optional[int] = 40000,
         expiration_in_secs: Optional[int] = 3600) -> bool

    Lock a resource.

    Create a global lock on a customer account to avoid concurrency when accessing a shared resource. This waits for the acquisition of the lock if a lock with the same name already exists.

    Arguments:

    • lock_name str - Name of the lock.
    • timeout_in_ms int, optional - Maximum time to wait to get the lock in milliseconds up to maximum of 40000ms. Defaults to 40000.
    • expiration_in_secs int, optional - Maximum time the lock can be kept if not released programmatically (in seconds). Default is 1 hour.

    Returns:

    • bool - Status of the lock, lock acquired or not.

    Raises:

    • TimeoutError - If the request doesn't get the lock.

    Example:

    state = "ok"
lock_test = RMPRequest.lock("test", 10000)
if lock_test:
    RMPRequest.unlock("test")
else:
    state = "nok"
RMPData.setOutput(state)

    unlock

    @staticmethod
def unlock(lock_name: str) -> None

    Release a lock acquired with RMPRequest.lock().

    Arguments:

    • lock_name str - Name of the lock.

    Returns:

    None

    Example:

    state = "ok"
f = RMPRequest.lock("test", 10000)
if f:
    RMPRequest.unlock("test")
else:
    state = "nok"
RMPData.setOutput(state)

    raiseError

    @staticmethod
def raiseError(error_message: str) -> str

    Raise an error with a given message.

    Arguments:

    • error_message str - The error message.

    Returns:

    • str - The error message.

    Raises:

    • RuntimeError - An error with the given message.

    Example:

    error = RMPRequest.raiseError("An error has occurred")
RMPData.setOutput(error)

    readFileAddLane

    @staticmethod
def readFileAddLane(file_ids: List[str],
                    lane_ids: Optional[List[int]] = Any) -> Dict

    Add lane(s) allowed to READ the file(s).

    Arguments:

    • file_ids List[str] - A list of file identifiers.
    • lane_ids List[int] - Optional, A list of lane identifiers. Defaults to current lane

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    j = RMPRequest.readFileAddLane(["fb090250-b571-4aba-a25a-77a996fe72ee"], [53762810])
RMPData.setOutput(j)

    readFileAddUser

    @staticmethod
def readFileAddUser(file_ids: List[str],
                    user_ids: Optional[List[int]] = Any) -> List[Dict]

    Add user(s) allowed to READ the file(s).

    Arguments:

    • file_ids List[str] - A list of file identifiers.
    • user_ids List[int], optional - A list of either user identifiers. Defaults to Current User.

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    add_user = RMPRequest.readFileAddUser(["fb090250-b571-4aba-a25a-77a996fe72ee"], [66021605])
RMPData.setOutput(add_user)

    readFileRemoveLane

    @staticmethod
def readFileRemoveLane(file_ids: List[str],
                       lane_ids: Optional[List[int]] = Any) -> List[Dict]

    Remove lane(s) allowed to READ the file(s).

    Arguments:

    • file_ids List[str] - A list of file identifiers.
    • lane_ids List[int], optional - A list of lane identifiers. Defaults to current lane.

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    remove_lane = RMPRequest.readFileRemoveLane(["03acf9f2-609f-4cb4-bcbe-17e4005e1fea"], [53818190])
RMPData.setOutput(remove_lane)

    readFileRemoveUser

    @staticmethod
def readFileRemoveUser(file_ids: List[str],
                       user_ids: Optional[List[int]] = Any) -> List[Dict]

    Remove connected/login user(s) allowed to READ the file(s).

    Arguments:

    • file_ids List[str] - A list of file identifiers.
    • userIds List[int], optional - A list of user Ids. Defaults to the current user

    Returns:

    • Dict - The list of lanes and login users with the appropriate right for each file in JSON.

    Example:

    remove_user = RMPRequest.readFileRemoveUser(["fb090250-b571-4aba-a25a-77a996fe72ee"])
RMPData.setOutput(remove_user)

    removeUserFromLane

    @staticmethod
def removeUserFromLane(lane_id: int, user_id: int) -> boolean

    Remove user(s) from the given runtime lane.

    Arguments:

    • lane_id int - ID of the lane.
    • user_id int - ID of the user.

    Returns:

    • boolean - status if adding a user to the lane.

    Example:

    fromLane = RMPRequest.removeUserFromLane(53877772,65914367)
RMPData.setOutput(fromLane)

    updateFileAddLane

    @staticmethod
def updateFileAddLane(file_ids: List[str],
                      lane_ids: Optional[List[int]] = Any) -> List[Dict]

    Add lane(s) allowed to WRITE/DELETE the file(s).

    Arguments:

    • file_ids List[str] - a list of file identifiers.
    • lane_ids List[int] - a list of lane identifiers. Defaults to current lane

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    addLane = RMPRequest.updateFileAddLane(["fb090250-b571-4aba-a25a-77a996fe72ee"],[53762810])
RMPData.setOutput(addLane)

    updateFileAddUser

    @staticmethod
def updateFileAddUser(file_ids: List[str],
                      user_ids: Optional[List[int]] = Any) -> List[Dict]

    Update users(s) allowed to WRITE/DELETE the file(s).

    Arguments:

    • file_ids List[str] - a list of file identifiers.
    • user_ids List[int] - a list of either user identifiers or user logins. Defaults to current user

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    addUser = RMPRequest.updateFileAddUser(["fb090250-b571-4aba-a25a-77a996fe72ee"],[66021605])
RMPData.setOutput(addUser)

    updateFileRemoveLane

    @staticmethod
def updateFileRemoveLane(file_ids: List[str],
                         lane_ids: Optional[List[int]] = Any) -> List[Dict]

    Remove lanes allowed to WRITE/DELETE the file(s).

    Arguments:

    • file_ids List[str] - a list of file identifiers.
    • lane_ids List[int] - a list of either user identifiers or user logins. Defaults to current user

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON.

    Example:

    removeLane = RMPRequest.updateFileRemoveLane(["03acf9f2-609f-4cb4-bcbe-17e4005e1fea"],[53818190])
RMPData.setOutput(removeLane)

    updateFileRemoveUser

    @staticmethod
def updateFileRemoveUser(file_ids: List[str],
                         user_ids: Optional[List[int]] = Any) -> List[Dict]

    Remove lanes allowed to WRITE/DELETE the file(s).

    Arguments:

    • file_ids List[str] - a list of file identifiers.
    • user_ids List[int] - a list of either user identifiers or user logins. Defaults to current user

    Returns:

    • List[Dict] - The list of lanes and users with the appropriate right for each file in JSON

    Example:

    removeUser = RMPRequest.updateFileRemoveUser(["fb090250-b571-4aba-a25a-77a996fe72ee"],[66021605])
RMPData.setOutput(removeUser)

    setRequestStatus

    @staticmethod
def setRequestStatus(status_code: int) -> bool

    Set the status of the current request to the status code.

    Arguments:

    • status_code int - Status code to set (e.g., 102, 201, 301, 302, 400, 401).

    Returns:

    • bool - True if status is updated, else False.

    Example:

    request_status = RMPRequest.setRequestStatus(200)
RMPData.setOutput(request_status)

    RMPUser

    class RMPUser()

    RMPUser contains various static methods to interact with user-related data, applications, lanes, and user preferences.

    getApplications

    @staticmethod
def getApplications(login: str = None,
                    app_context: Optional[str] = Any,
                    mode: Optional[str] = Any,
                    is_tagged: Optional[bool] = False) -> List[Application]

    Returns the list of applications authorized to a user.

    Arguments:

    • login str, optional - User login/email address.
    • app_context str, optional - Context in which the application can be used ('web', 'mobile', 'tablet').
    • mode str, optional - Execution mode ('LIVE', 'ACCEPTANCE', 'TEST').
    • is_tagged bool, optional - Return the list of user interfaces grouped by tag if True.

    Returns:

    • List[Application] - List of applications.

    Example:

    list_of_applications = RMPUser.getApplications("user@runmyprocess.com")
RMPData.setOutput(list_of_applications)

    getLanes

    @staticmethod
def getLanes(login: str) -> List[Lane]

    Returns the list of lanes a user pertains to in the current execution mode context.

    Arguments:

    • login str - User login/email address.

    Returns:

    • List[Lane] - List of lanes.

    Example:

    list_of_lanes = RMPUser.getLanes("user@runmyprocess.com")
RMPData.setOutput(list_of_lanes)

    getLaneUsers

    @staticmethod
def getLaneUsers(pool_id: int, lane_id: int, page_size: int,
                 skip: int) -> List[User]

    Returns the list of users pertaining to a lane in the current execution mode context.

    Arguments:

    • pool_id int - Pool/Organization identifier.
    • lane_id int - Lane/Role identifier.
    • page_size int - Number of users returned.
    • skip int - Start pointer for pagination purposes.

    Returns:

    • list[User] - List of users.

    Example:

    list_of_lane_users = RMPUser.getLaneUsers(112, 5701, 5, 20)
RMPData.setOutput(list_of_lane_users)

    getManager

    @staticmethod
def getManager(login: str, level: Optional[int] = 1) -> str

    Returns the login of a manager N levels above the designated user.

    Arguments:

    • login str - User login/email address.
    • level int, optional - Number of levels in the hierarchy. Default is 1.

    Returns:

    • str - Login/email address of the manager.

    Raises:

    • ValueError - If no manager is found.

    Example:

    the_users_manager = RMPUser.getManager("user@rmp.com", 1)
RMPData.setOutput(the_users_manager)

    getUserData

    @staticmethod
def getUserData(login: Optional[str]) -> User

    Returns a user's basic information (id, name, profile) from their login.

    Arguments:

    • login str - User login/email address.

    Returns:

    • User - User's basic data.

    Example:

    the_user = RMPUser.getUserData("john@doe.com")
RMPData.setOutput(the_user)

    getUserLanes

    @staticmethod
def getUserLanes(login: str) -> list[Lane]

    Returns the list of lanes a user belongs to.

    Arguments:

    • login str - User login/email address.

    Returns:

    • list[Lane] - List of lanes.

    Example:

    user_lanes = RMPUser.getUserLanes("john@doe.com")
RMPData.setOutput(user_lanes)

    hasRightInLane

    @staticmethod
def hasRightInLane(login: str, lane_id: int) -> bool

    Checks if a user belongs to a lane in the current execution mode context.

    Arguments:

    • login str - User login/email address.
    • lane_id int - Lane/Role identifier.

    Returns:

    • bool - True if the user belongs to the indicated lane, False otherwise.

    Example:

    does_belong_to_lane = RMPUser.hasRightInLane("user@runmyprocess.com", 5701)
RMPData.setOutput(does_belong_to_lane)

    getUserMetaData

    @staticmethod
def getUserMetaData(login: Optional[str] = Any) -> Dict

    Returns the metadata of a user.

    Arguments:

    • login str, optional - User login/email address. Defaults to current user.

    Returns:

    • Dict - User's metadata.

    Example:

    user_meta_data = RMPUser.getUserMetaData("user@runmyprocess.com")
RMPData.setOutput(user_meta_data)

    saveUserMetaData

    @staticmethod
def saveUserMetaData(login: str, metadata: Dict) -> str

    Updates the metadata of a user.

    Arguments:

    • login str - User login/email address.
    • metadata Dict - New metadata settings.

    Returns:

    • str - Login of the user.

    Example:

    new_metadata = {"name": "user", "age": 22, "car": None}
login = RMPUser.saveUserMetaData("user@runmyprocess.com", new_metadata)
RMPData.setOutput(login)

    getUserPreferences

    @staticmethod
def getUserPreferences(login: Optional[str] = Any) -> Dict

    Returns the preferences associated with a user.

    Arguments:

    • login str, optional - User login/email address. Defaults to current user.

    Returns:

    • Dict - User's preferences.

    Example:

    preferences = RMPUser.getUserPreferences("user@runmyprocess.com")
RMPData.setOutput(preferences)

    saveUserPreferences

    @staticmethod
def saveUserPreferences(preferences: Dict) -> str

    Updates the preferences of a user.

    Arguments:

    • preferences Dict - New preferences settings.

    Returns:

    • str - Login of the user.

    Example:

    new_preferences = {"elementTest": "DT", "Login": "user"}
login = RMPUser.saveUserPreferences(new_preferences)
RMPData.setOutput(login)

    impersonate

    @staticmethod
def impersonate(login: str) -> str

    Impersonates another user on the same account (only allowed when the process user is an account's admin).

    Arguments:

    • login str - User login.

    Returns:

    • str - Login of the impersonated user.

    Example:

    login = RMPUser.impersonate("someUserEmail@runmyprocess.com")
metadata = RMPUser.getUserMetaData()
RMPData.setOutput(metadata)

    User

    class User(TypedDict)

    Preferences

    Represents user information.

    Lane

    class Lane(TypedDict)

    name

    Represents lane information.

    Application

    class Application(TypedDict)

    Tags

    Represents application information.

    RMPUtilities

    class RMPUtilities()

    Namespace for RMPUtilities.

    csvToJson

    @staticmethod
def csvToJson(csv_data: str, option_list: CSVOptions) -> str

    Converts a CSV string into a JSON array of objects.

    Arguments:

    • csv_data - The input CSV formatted string.
    • option_list - Options to be used for the transformation

    Returns:

    A string containing the CSV data in JSON format where the keys of the object correspond to the column headers.

    Raises:

    • ValueError - If the CSV data is empty or invalid.

    Example:

    jsonResult = RMPUtilities.csvToJson('1,true,hello,world', {})
RMPData.setOutput(jsonResult)

    jsonToCsv

    @staticmethod
def jsonToCsv(json_data: List[Dict], option_list: CSVOptions) -> str

    Converts a JSON object or array into a CSV formatted string.

    This method takes a JSON array of objects and converts it into a comma-separated values (CSV) string. The function assumes that all objects in the input array have the same set of keys.

    If the input is invalid or empty, this method may return an empty string or throw an error depending on implementation.

    Arguments:

    • json_data - The JSON data to be converted into CSV format.
    • option_list - Options to be used for the transformation

    Returns:

    A string representing the data in CSV format.

    Raises:

    • ValueError - If the JSON data is empty or invalid.

    Example:

    csvResult = RMPUtilities.jsonToCsv([["1","true","hello"],["2","false","world"]], {})
RMPData.setOutput(csvResult)

    CSVOptions

    class CSVOptions()

    trim

    Options for CSV parsing and formatting.

    Attributes:

    • separator - Separator between two values (e.g. ; or ,). Default: ,
    • delimiter - Used to allow values containing a separator character (e.g. " or '). To include a delimiter in a value, use two delimiters (e.g. ""). Default: "
    • empty - Value to be used to represent an empty value (two separators in a row). Default: ""
    • charset - Character set to be used to read the file content. Default: UTF-8
    • parse_numbers - Names/indices of columns containing numerical values to be stored as Double. e.g ["age",3] will parse columns header "age" and the 4th column
    • parse_longs - Names/indices of columns containing long values to be stored as Long.
    • e.g - ["age",3] will parse column header "age" and the 4th column.
    • trim - Trim values before inserting them in the collection. Trimming also occurs before decoding numerical values.

    decrypt

    @staticmethod
def decrypt(encrypted_data: str,
            vector: str,
            key: str,
            algorithm: Optional[str] = "AES/CBC/PKCS5Padding") -> str

    Decrypts an encrypted string using the predefined decryption mechanism.

    This function takes an encrypted string and processes it to return the original plaintext, utilizing the specified cryptographic algorithm and a pre-configured secret key or configuration.

    Arguments:

    • encrypted_data - The encrypted string that needs to be decrypted.
    • vector - The initialization vector
    • key - Key to decrypt the ciphered text in AES format
    • algorithm - Cipher algorithm (e.g. AES, AES/CBC/PKCS5Padding, DES, RSA, Blowfish).
    • Default - "AES/CBC/PKCS5Padding"

    Returns:

    The decrypted plaintext string.

    Raises:

    • ValueError - If the decryption fails, such as when the encrypted data is corrupted or the configuration is invalid.

    Example:

    result = RMPUtilities.decrypt("HtcLaNk0xk/G+DjClefgdA==",
                              "LqME74+sChTIcYsaUMkBPw==",
                              "0ckdURPMf4P4ismngaFgZbX3CGzJCyuR6hTgEAL9RTc=")
RMPData.setOutput(result)

    encrypt

    @staticmethod
def encrypt(input_str: str,
            key: str,
            algorithm: Optional[str] = "AES/CBC/PKCS5Padding") -> Dict

    Encrypts a given input string using a specified encryption algorithm.

    This function takes a plain text input string and applies an encryption process to convert it into a secure, encoded format. It may use a default or provided encryption algorithm depending on the implementation. The output will typically be a non-readable string that can only be decrypted using the corresponding decryption method.

    Arguments:

    • input_str - The plain text input to be encrypted.
    • key - Key to encrypt the ciphered text in AES format
    • algorithm - Cipher algorithm (e.g. AES, AES/CBC/PKCS5Padding, DES, RSA, Blowfish).
    • Default - "AES/CBC/PKCS5Padding"

    Returns:

    A dictionary containing the encrypted string and initialization vector.

    Raises:

    • ValueError - If the encryption fails.

    Example:

    result = RMPUtilities.encrypt('hello world', 'c7ZOu5rr3fMHTOAAwwlFD049PBVAZ6SU8UJuU3A9lVM=')
RMPData.setOutput(result)

    rsaHash

    @staticmethod
def rsaHash(input_str: str,
            key: str,
            hash_algorithm: Optional[str] = "SHA256") -> str

    Computes an RSA hash for the given input data and returns the hash value.

    This function is used for cryptographic operations such as secure ingestion and validation of data. The RSA hash is usually used in scenarios where data integrity and security are critical.

    Arguments:

    • input_str - The input data to be hashed.
    • key - The private key readers supports PEM files with PKCS#8 or PKCS#1 encodings. It doesn't support encrypted PEM files
    • hash_algorithm - e.g. SHA256 (default), MD5, SHA1

    Returns:

    The generated RSA hash as a Base64-encoded string.

    Raises:

    • ValueError - If the hashing operation or key processing fails.

    Example:

    key = '-----BEGIN RSA PRIVATE KEY-----....'
result = RMPUtilities.rsaHash('hello world', key, 'SHA256', 'HEXA')
RMPData.setOutput(result)  # e.g. {"encrypted": "{Base64String}", "iv": "{Base64String}"}

    uuid

    @staticmethod
def uuid() -> str

    Generates a universally unique identifier (UUID).

    This function creates a UUID string based on the Version 4 specification. UUIDs are 128-bit identifiers used to uniquely identify information in distributed systems. It ensures randomness and uniqueness, making it suitable for use in various applications requiring unique identifiers, such as databases or tokens.

    Returns:

    A randomly generated UUID in the standard format (e.g., "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx"), where x is a hexadecimal digit and y represents one of several possible values as per the specification.

    Example:

    result = RMPUtilities.uuid()
RMPData.setOutput(result)

    generateJWT

    @staticmethod
def generateJWT(header: Dict, claims: Dict, expiration_in_secs: int,
                operation: str, signature_algorithm: str) -> str

    Generates a JSON Web Token (JWT) based on the provided claims, header, and options.

    This method creates a signed token using the specified payload and secret key. It is commonly used for authentication and authorization purposes in web applications.

    Arguments:

    • header - JWT header
    • claims - JWT claims (Note: the "iat" (current time) is added per default)
    • expiration_in_secs - The expiration time in seconds
    • operation - Reference to the private key that you uploaded via API
    • signature_algorithm - none, HS256, HS384, HS512, RS256, RS512, ES256, ES384, PS256, PS384, PS512

    Returns:

    The generated JWT as a signed token string.

    Raises:

    • ValueError - If the payload or secret key validation fails or the token generation process encounters an issue.

    Example:

    header = {}
claims = {
      "iss": ".....",
      "sub": ".....",
      "aud": "account-d.docusign.com",
      "scope": "impersonation signature"
}
jwt = RMPUtilities.generateJWT(header, claims, 3600, 'DOCUSIGN', 'RS256')
RMPData.setOutput(jwt)