ProcessingStep implementation

Starting a new project

Your first step, when implementing a new ProcessingStep, is to initialize your project. Our command-line tool, pinexq, makes this process simple.

We recommend you run pinexq using uvx without installation; the command below will always run the latest version available. Alternatively, you can install the pinexq-cli via uv (or pipx); then, run pinexq init in your project directory.

uvx

uvx --from pinexq-cli@latest pinexq init

uv install

uv tool install pinexq-cli
pinexq init

pipx install

pipx install pinexq-cli
pinexq init

The tool will ask you some questions, and set up a suitable template project. (Note that Docker requires your project name to be lower case.) The template implements a function called hello_world. You can try running it with uv run.

PS C:\Users\Me\MyProject> uv run main.py run -f hello_world
21-01-26 10:18:10 INFO Platform and package information:
OS: Windows-11-10.0.26200-SP0
Python: CPython 3.14.1
pinexq-procon: 2.3.0
pinexq-client: 1.0.0
21-01-26 10:18:10 INFO Hello World!
Hello World!

Your first ProcessingStep

To implement your first ProcessingStep in Python, you need to create a new class that inherits from the pinexq.procon.Step base class which provides the internal SDK functionalities. All methods of this class are then exposed as ProcessingSteps of this container. To exclude a method from being exposed, prefix its name with an underscore (e.g. def _private_method(...)).

Note

For all ProcessingSteps, it is mandatory to have parameters and return values with type annotations, default values and a docstring following standard Python conventions.

Versions

It is also strongly recommended to add a version decorator e.g. @version("0.1.0-dev1") to your functions. This version will also be placed into the functions manifest file.

Note that Semantic Versioning is mandatory. In particular:

• The version string must have three integer versions, separated by ..
• Versions are ordered lexicographically as integers (i.e. 0.1.9 < 0.1.10 < 0.2.0).
• The version string may optionally end in a - followed by a prerelease identifier.
• A version with a prerelease identifier is considered lower than the same major.minor.patch version without, e.g. 0.1.9-any < 0.1.9.
• The prerelease identifier must be composed only of ASCII alphanumerics and hyphens; underscores are heavily discouraged.

Recommendations for version handling

It is recommended to follow semantic versioning guidelines to make use of wildcards when specifying versions. E.g. 1.* should keep the expected results unchanged besides bugfixes and additional feature which do not break the behavior.

Rule of thumb:

• If parameters or dataslots of a ProcessingStep changes, the major version should be increased. This avoids breaking current uses.
• If the fundamental behavior changes (also for same input) the major should be increase to avoid surprises
• If ProcessingStep is improved but old configuration still is acceptable, the minor version should be increased
• If a bug is fixed the patch version should be increase.

Running a function

Finally, start the container by calling the new class and creating an instance of it. It’s recommended to do this inside of a script-guard to allow importing the script as a module. The default behavior of the Step class is to spawn a CLI interface to manage a control the container. This can be avoided by passing the use_cli=False parameter to the constructor (e.g. for unit-tests or to use the container as a module).

A minimal example

from pinexq.procon.step import Step, version # import package

class MyStepCollection(Step):  # define the container class
    @version("0.1.0-dev1")
    def calculate_square(self, x: float) -> float:  # define a step function
        """Calculate the square of x

        :param x: a float number
        :returns: the square of x
        """
        return x ** 2

    # More step functions can go in the same class

if __name__ == '__main__':  # add script guard
    MyStepCollection()  # run the container - this will spawn the cli

Comments

If a step function contains a docstring it is picked up by the manifest generation. Any common docstring format will be recognized (Google, reST, NumPy). It is recommended to have meaningful parameter descriptions in the comments, as to the outside of the container they are only visible by their name and description. Markdown syntax and formulas using MathJax is supported. Please escape \ with \\, or use a raw string (r"""); for more information, refer to here.

Specifically a ProcessingStep should have:

• a general function comment (first line)
• comments for input parameters (e.g. : :param x:)
• comments for result values or options (e.g. :returns: the square of x)

Type annotations

ProCon extracts a functions parameters, return values and their corresponding types from the functions type annotations. Type hints can be provided as standard types from the typing or pydantic module. The only restriction is that a type has to be JSON-serializable, which includes all the standard Python types.

Generic object types like Any or object are represented in JSON as dict. Conversion might fail for types where that is not applicable. If you must pass other data types, consider converting them a standard type first (e.g. converting a numpy.ndarray to a list) or pass them via a DataSlot.

Caution

Parameters and return values are validated against their annotated types during execution and the container will raise an Exception if there is a mismatch!

Parameters

For simple types, including those found in typing, type hints can be provided directly. More complex types, such as StrEnums, should be passed through via pydantic models. For more information, refer to Parameters.

Additionally, default parameters can be provided, making the parameter optional and adding those as a default to the manifest.

Example with multiple parameters:

from typing import Any
from pydantic import conlist
from pinexq.procon.step import version

# inside a Step...
@version("0.1.0-dev1")
# ...
def many_parameters(
            self,
            a: dict,
            b: int,
            c: conlist(int, max_length=3) = (1, 2, 3),
            d: bool = True
) -> int:
    ...

Caution

The parameters used for running a Step are logged. Secrets should not be provided to Steps via Parameter!

Adding DataSlots

DataSlots are the way to declare that a function requires or will produce some external data, usually a file. Perhaps the data is too big to serialize it as a JSON result object, is already present as WorkData, or is needed as a file.

DataSlots are defined with the pinexq.procon.dataslot function decorator. Typical use cases will need the following:

• @dataslot.input: Load input data from file. The data from the slot will be cast to the annotated type of the corresponding parameter, if it’s not explicitly of type DataSlot.
• @dataslot.returns: Indicates the return value of the function should be written to this DataSlot.

Example of a function declaring DataSlots:

import json
from pinexq.procon.dataslots import dataslot, MediaTypes

@dataslot.input('in_list_of_str1', title='In List 1', reader=json.load, media_type=MediaTypes.JSON)
@dataslot.input('in_list_of_str2', title='In List 2', reader=json.load, media_type=MediaTypes.JSON)
@dataslot.returns(title='TheReturnValue', writer=lambda f, d: json.dump(d, f), media_type=MediaTypes.JSON)
def dataslot_many_slots(
        self,
        in_list_of_str1: list[str],
        in_list_of_str2: list[str]
) -> dict[str, list[str]]:
    """Declare i/o slots

    :param in_list_of_str1: Input slot 1
    :param in_list_of_str2: Input slot 2
    :returns: Concatenated content of all slots
    """
    return {'return_value': in_list_of_str1 + in_list_of_str2}

Note that the media_type argument is typically required, as it determines the filetype of the input/output.

For more information and advanced usage, refer to DataSlots.

Errors

ProCon will catch exceptions from functions, log them and report errors back to the host system. In case of an error, just raise an exception to exit execution early. If you want to communicate a reason to the user, e.g. a computation can not be continued or another system is not reachable, raise a ProConJobExecutionError and set the user_message string. Only the content of user_message will be presented to the user through the API; the error message itself can only be seen in the logs.

from pinexq.procon.core.exceptions import ProConJobExecutionError
from cloud_service import RuntimeService

try:
    return RuntimeService(filename=license_filename)
except JSONDecodeError as err:
    raise ProConJobExecutionError(str(err),
        user_message="Invalid JSON in license file")