mirror of
https://github.com/SELinuxProject/setools
synced 2025-04-22 15:15:16 +00:00
be63a5be7b
This is to support more complex rendering than the default, str(). Also support configurations where there is no model to update. Signed-off-by: Chris PeBenito <pebenito@ieee.org>
84 lines
2.5 KiB
Python
84 lines
2.5 KiB
Python
# Copyright 2016, Tresys Technology, LLC
|
|
#
|
|
# SPDX-License-Identifier: LGPL-2.1-only
|
|
#
|
|
#
|
|
import logging
|
|
import typing
|
|
|
|
from PyQt5 import QtCore
|
|
import setools
|
|
|
|
from . import models
|
|
|
|
Q = typing.TypeVar("Q", bound=setools.PolicyQuery)
|
|
|
|
# The first parameter is the result counter and second parameter
|
|
# is a single result to render.
|
|
RenderFunction = typing.Callable[[int, typing.Any], str]
|
|
|
|
|
|
class QueryResultsUpdater(QtCore.QObject, typing.Generic[Q]):
|
|
|
|
"""
|
|
Thread for processing basic queries and updating result widgets.
|
|
|
|
Parameters:
|
|
query The query object
|
|
model The model for the results
|
|
|
|
Keyword Parameters:
|
|
render A two parameter function that renders each item returned
|
|
from the query to a string. This is added to the raw output
|
|
widgets. The default is equivalent to str().
|
|
|
|
Qt signals:
|
|
failed (str) The updated failed, with an error message.
|
|
finished (int) The update has completed, with the number of results.
|
|
raw_line (str) A string to be appended to the raw results.
|
|
"""
|
|
|
|
failed = QtCore.pyqtSignal(str)
|
|
finished = QtCore.pyqtSignal(int)
|
|
raw_line = QtCore.pyqtSignal(str)
|
|
|
|
def __init__(self, query: Q,
|
|
model: models.SEToolsTableModel | None = None,
|
|
render: RenderFunction = lambda _, x: str(x)) -> None:
|
|
|
|
super().__init__()
|
|
self.log: typing.Final = logging.getLogger(query.__module__)
|
|
self.query: typing.Final[Q] = query
|
|
self.model = model
|
|
self.render = render
|
|
|
|
def update(self) -> None:
|
|
"""Run the query and update results."""
|
|
results: typing.List = []
|
|
counter = 0
|
|
|
|
try:
|
|
for counter, item in enumerate(self.query.results(), start=1):
|
|
results.append(item)
|
|
|
|
self.raw_line.emit(self.render(counter, item))
|
|
|
|
if QtCore.QThread.currentThread().isInterruptionRequested():
|
|
break
|
|
elif counter % 10 == 0:
|
|
# yield execution every 10 rules
|
|
QtCore.QThread.yieldCurrentThread()
|
|
|
|
if counter % 1000 == 0:
|
|
self.log.info(f"Generated {counter} results so far.")
|
|
|
|
self.log.info(f"Generated {counter} total results.")
|
|
|
|
except Exception as e:
|
|
self.failed.emit(str(e))
|
|
|
|
else:
|
|
if self.model:
|
|
self.model.item_list = results
|
|
self.finished.emit(counter)
|