Skip to content

InspectStats

InspectStats.

InspectStats

Bases: UtilsInspectStats

Class for inspection of statistics.

Source code in src/qtasks/stats/inspect/inspect.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class InspectStats(UtilsInspectStats):
    """Class for inspection of statistics."""

    def __init__(self, app: Union["QueueTasks", "aioQueueTasks"]):
        """
        Initializing the statistics inspection.

        Args:
            app (QueueTasks): Application instance.
        """
        self._app = app

    def app(self, json: bool = False):
        """
        Getting information about the application.

        Args:
            json (bool, optional): Flag to return in JSON format. Default: `False`.

        Returns:
            str: Application information.
        """
        return self._app_parser(self._app, json=json)

    def task(self, task_name: str, json: bool = False):
        """
        Obtaining information about a task.

        Args:
            task_name (str): The name of the task.
            json (bool, optional): Flag to return in JSON format. Default: `False`.

        Returns:
            TaskExecSchema: Schema of the task function.
        """
        if json:
            return self._parser_json(self._app.tasks[task_name])
        return self._tasks_parser((self._app.tasks[task_name],))

    def tasks(self, *tasks: tuple[str], json: bool = False):
        """
        Obtaining information about tasks.

        Returns:
            List[TaskExecSchema]: Task function schemas.
            json (bool, optional): Flag to return in JSON format. Default: `False`.
        """
        if not tasks:
            result = self._app.tasks.values()
        else:
            result = [
                self._app.tasks[task] for task in tasks if task in self._app.tasks
            ]

        if json:
            return self._parser_json(result)

        return self._tasks_parser(result)

__init__(app)

Initializing the statistics inspection.

Parameters:

Name Type Description Default
app QueueTasks

Application instance.

 required
Source code in src/qtasks/stats/inspect/inspect.py 
15
16
17
18
19
20
21
22
def __init__(self, app: Union["QueueTasks", "aioQueueTasks"]):
    """
    Initializing the statistics inspection.

    Args:
        app (QueueTasks): Application instance.
    """
    self._app = app

app(json=False)

Getting information about the application.

Parameters:

Name Type Description Default
json bool

Flag to return in JSON format. Default: False.

 False

Returns:

Name Type Description
str

Application information.
Source code in src/qtasks/stats/inspect/inspect.py 
24
25
26
27
28
29
30
31
32
33
34
def app(self, json: bool = False):
    """
    Getting information about the application.

    Args:
        json (bool, optional): Flag to return in JSON format. Default: `False`.

    Returns:
        str: Application information.
    """
    return self._app_parser(self._app, json=json)

task(task_name, json=False)

Obtaining information about a task.

Parameters:

Name Type Description Default
task_name str

The name of the task.

 required
json bool

Flag to return in JSON format. Default: False.

 False

Returns:

Name Type Description
TaskExecSchema

Schema of the task function.
Source code in src/qtasks/stats/inspect/inspect.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def task(self, task_name: str, json: bool = False):
    """
    Obtaining information about a task.

    Args:
        task_name (str): The name of the task.
        json (bool, optional): Flag to return in JSON format. Default: `False`.

    Returns:
        TaskExecSchema: Schema of the task function.
    """
    if json:
        return self._parser_json(self._app.tasks[task_name])
    return self._tasks_parser((self._app.tasks[task_name],))

tasks(*tasks, json=False)

Obtaining information about tasks.

Returns:

Name Type Description

List[TaskExecSchema]: Task function schemas.
json (bool, optional)

Flag to return in JSON format. Default: False.
Source code in src/qtasks/stats/inspect/inspect.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def tasks(self, *tasks: tuple[str], json: bool = False):
    """
    Obtaining information about tasks.

    Returns:
        List[TaskExecSchema]: Task function schemas.
        json (bool, optional): Flag to return in JSON format. Default: `False`.
    """
    if not tasks:
        result = self._app.tasks.values()
    else:
        result = [
            self._app.tasks[task] for task in tasks if task in self._app.tasks
        ]

    if json:
        return self._parser_json(result)

    return self._tasks_parser(result)