partials

process_manager.views.partials

View functions for partials.

Classes

Functions

filter_table(search, table)

Filter table data based on search parameter.

If the search parameter is empty, the table data is returned unfiltered. Otherwise, the table data is filtered based on the search parameter. The search parameter can be a string or a string with a column name and search string separated by a colon. If the search parameter is a column name, the search string is matched against the values in that column only. Otherwise, the search string is matched against all columns.

Parameters:

Name	Type	Description	Default
search	str	The search string to filter the table data.	required
table	list[dict[str, str | int]]	The table data to filter.	required

Returns:

Type	Description
list[dict[str, str | int]]	The filtered table data.

Source code in process_manager/views/partials.py

def filter_table(
    search: str, table: list[dict[str, str | int]]
) -> list[dict[str, str | int]]:
    """Filter table data based on search parameter.

    If the search parameter is empty, the table data is returned unfiltered. Otherwise,
    the table data is filtered based on the search parameter. The search parameter can
    be a string or a string with a column name and search string separated by a colon.
    If the search parameter is a column name, the search string is matched against the
    values in that column only. Otherwise, the search string is matched against all
    columns.

    Args:
        search: The search string to filter the table data.
        table: The table data to filter.

    Returns:
        The filtered table data.
    """
    if not search or not table:
        return table

    all_cols = list(table[0].keys())
    column, _, search = search.partition(":")
    if not search:
        # No column-based filtering
        search = column
        columns = all_cols
    elif column not in all_cols:
        # If column is unknown, search all columns
        columns = all_cols
    else:
        # Search only the specified column
        columns = [column]
    search = search.lower()
    return [row for row in table if any(search in str(row[k]).lower() for k in columns)]

handle_errors(view_func)

Decorator to handle errors.

Parameters:

Name	Type	Description	Default
view_func	Callable[[HttpRequest], HttpResponse]	The view function to be wrapped.	required

Returns:

Type	Description
Callable[[HttpRequest], HttpResponse]	The wrapped view function.

Source code in process_manager/views/partials.py

def handle_errors(
    view_func: Callable[[HttpRequest], HttpResponse],
) -> Callable[[HttpRequest], HttpResponse]:
    """Decorator to handle errors.

    Args:
        view_func: The view function to be wrapped.

    Returns:
        The wrapped view function.
    """
    logger = logging.getLogger("django")

    def wrapped_view(request, *args, **kwargs) -> HttpResponse:  # type: ignore
        try:
            return view_func(request, *args, **kwargs)
        except Exception as e:
            logger.exception(e)
            return render(request, "process_manager/partials/error_message.html")

    return wrapped_view

messages(request)

Search and render Kafka messages from the database.

Source code in process_manager/views/partials.py

@login_required
@handle_errors
def messages(request: HttpRequest) -> HttpResponse:
    """Search and render Kafka messages from the database."""
    search = request.GET.get("search", "")
    records = DruncMessage.objects.filter(
        topic__regex=settings.KAFKA_TOPIC_REGEX["PROCMAN"], message__icontains=search
    )

    # Time is stored as UTC. localtime(t) converts this to our configured timezone.
    messages = [
        f"{localtime(record.timestamp).strftime('%Y-%m-%d %H:%M:%S')}: {record.message}"
        for record in records
    ]

    return render(
        request=request,
        context=dict(messages=messages[::-1]),
        template_name="process_manager/partials/message_items.html",
    )

process_table(request)

Renders the process table.

This view may be called using either GET or POST methods. GET renders the table with no check boxes selected. POST renders the table with checked boxes for any table row with a uuid provided in the select key of the request data.

Source code in process_manager/views/partials.py

@login_required
@handle_errors
def process_table(request: HttpRequest) -> HttpResponse:
    """Renders the process table.

    This view may be called using either GET or POST methods. GET renders the table with
    no check boxes selected. POST renders the table with checked boxes for any table row
    with a uuid provided in the select key of the request data.
    """
    session_info = get_session_info()

    status_enum_lookup = dict(item[::-1] for item in ProcessInstance.StatusCode.items())

    table_data = []
    process_instances = session_info.data.values
    for process_instance in process_instances:
        metadata = process_instance.process_description.metadata
        uuid = process_instance.uuid.uuid
        table_data.append(
            {
                "uuid": uuid,
                "name": metadata.name,
                "user": metadata.user,
                "session": metadata.session,
                "status_code": status_enum_lookup[process_instance.status_code],
                "exit_code": process_instance.return_code,
            }
        )
    # Filter table data based on search parameter
    table_data = filter_table(request.GET.get("search", ""), table_data)
    table = ProcessTable(table_data)

    # sort table data based on request parameters
    table_configurator = django_tables2.RequestConfig(request)
    table_configurator.configure(table)

    return render(
        request=request,
        context=dict(table=table),
        template_name="process_manager/partials/process_table.html",
    )