Open edX Filters Tooling

Here we document the tooling available for working with Open edX filters as a developer.

class openedx_filters.tooling.OpenEdxPublicFilter

Custom class used to create Open edX Filters.

classmethod get_filter_config() → dict[str, Any]

Get filters configuration from Django settings.

Helper function used to get configuration needed for using a filter.

Where:

• pipeline (list): paths where the functions to be executed by the pipeline are defined.

• fail_silently (bool): determines whether the pipeline can raise exceptions while executing.

  If its value is True then common exceptions (TypeError, ImportError…) are caught and the execution continues, if False then exceptions are re-raised and the execution fails.

• The rest of the keys are extra configuration that can be used to customize the filter.

Returns:

configuration for the filter type defined in the class.

Return type:

dict

Example usage:

>>> configuration = get_filter_config('trigger')
>>> configuration
{
    'pipeline':
        [
            'my_plugin.hooks.filters.PipelineStepV1',
            'my_plugin.hooks.filters.PipelineStepV2',
        ],
    'fail_silently': False,
    'log_level': 'debug'
}

classmethod get_pipeline_configuration() → tuple[list[str], bool, dict[str, Any]]

Get pipeline configuration from filter settings.

Helper function used to get the configuration needed to execute the Pipeline Runner. It will take from the hooks configuration the list of functions to execute and how to execute them.

Returns:

pipeline configuration with the following structure:

• list: Paths where functions for the pipeline are defined.

• bool: Indicates whether exceptions are raised while executing the pipeline associated with a filter. Determined by the fail_silently configuration: True means it won’t raise exceptions, and False means the opposite.

• dict: Extra configuration defined in the filter configuration.

Return type:

tuple

Example usage:

>>> pipeline_config = cls.get_pipeline_configuration()
>>> pipeline_config
    (
        [
            'my_plugin.hooks.filters.PipelineStepV1',
            'my_plugin.hooks.filters.PipelineStepV2',
        ],
        False,
        {
            'log_level': 'debug'
        }
    )

classmethod get_steps_for_pipeline(pipeline: list, fail_silently: bool = True) → list[type]

Get pipeline objects from paths.

Helper function that given a pipeline with step paths gets the objects related to each path.

Parameters:

• pipeline (list) – paths where steps are defined.

• fail_silently (bool) – True meaning it won’t raise exceptions and False the opposite.

Returns:

objects related to each path in the pipeline.

Return type:

list

Example usage:

>>> steps = get_steps_for_pipeline(
    [
        '1st_path_to_step',
        ...
    ]
)
>>> steps
[
    <class 1st_class at 0x00000000000>,
    <class 2nd_class at 0x00000000001>,
    ...
]

classmethod run_pipeline(**kwargs: Any) → dict[str, Any] | Any

Execute filters in order based on the pipeline configuration.

Given a list of pipeline steps, this function will execute them using the Accumulative Pipeline pattern as specified in 3. Filter tooling: pipeline behaviour.

Parameters:

**kwargs – arguments to be passed to the pipeline steps.

Returns:

accumulated outputs of the pipelines that were executed or the return value of a pipeline step

if it’s not a dictionary.

Return type:

dict | Any

Raises:

OpenEdxFilterException – exception re-raised when a pipeline step raises an exception of this type. This behavior is common when using filters to alter the application execution.

This pipeline implementation was inspired by: Social auth core. For more information check their Github repository: python-social-auth/social-core

Example usage:

>>> result = OpenEdxPublicFilter.run_filter(user=user, course=course)
>>> result
{
    'result_1st_function': 1st_object,
    'result_2nd_function': 2nd_object,
}

exception openedx_filters.exceptions.OpenEdxFilterException(message: str = '', redirect_to: str = '', status_code: int | None = None, **kwargs)

Base exception for filters.

It is re-raised by the Pipeline Runner if any filter that is executing raises it.

Parameters:

• message (str) – message describing why the exception was raised.

• redirect_to (str) – redirect URL.

• status_code (int) – HTTP status code.

• arguments (keyword) – extra arguments used to customize

• exception.

class openedx_filters.filters.PipelineStep(filter_type, running_pipeline, **extra_config)

Defines each step of the pipeline to be executed by the pipeline runner.

Example usage:

Let’s say we want a filter step that changes enrollment modes:

1. If the enrollment mode is honor, then changes it to no-id-professional.

2. If it’s other mode, then stop pipeline execution.

This pipeline step can be used in conjunction with PreEnrollmentFilter, that’s why the runner method accepts user, course_key and mode as arguments.

>>> class MyFilterStep(PipelineStep):
        def run_filter(self, user, course_key, mode):
            if mode != "honor":
                return
            return {
                "user": user,
                "course_key": course_key,
                "mode": "no-id-professional",
            }

Another version would be:

>>> class MyFilterStep(PipelineStep):
        def run_filter(self, user, course_key, mode):
            if mode != "honor":
                return
            return {"mode": "no-id-professional"}

abstract run_filter(**kwargs)

Abstract pipeline step runner.

Used to implement custom code that’ll be executed by OpenEdxPublicFilter’s pipeline runner. It must be implemented by child classes.