gvm.protocols.gmpv208.system.aggregates.AggregatesMixin.get_aggregates()

Complexity

Conditions

30

Size

Total Lines	162
Code Lines	106

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

# -*- coding: utf-8 -*-
    def get_aggregates(
        self,
        resource_type: EntityType,
        *,
        filter_string: Optional[str] = None,
        filter_id: Optional[str] = None,
        sort_criteria: Optional[list] = None,
        data_columns: Optional[list] = None,
        group_column: Optional[str] = None,
        subgroup_column: Optional[str] = None,
        text_columns: Optional[list] = None,
        first_group: Optional[int] = None,
        max_groups: Optional[int] = None,
        mode: Optional[int] = None,
        **kwargs,
    ) -> Any:
        """Request aggregated information on a resource / entity type

        Additional arguments can be set via the kwargs parameter for backward
        compatibility with older versions of python-gvm, but are not validated.

        Arguments:
            resource_type: The entity type to gather data from
            filter_string: Filter term to use for the query
            filter_id: UUID of an existing filter to use for the query
            sort_criteria: List of sort criteria (dicts that can contain
                a field, stat and order)
            data_columns: List of fields to aggregate data from
            group_column: The field to group the entities by
            subgroup_column: The field to further group the entities
                inside groups by
            text_columns: List of simple text columns which no statistics
                are calculated for
            first_group: The index of the first aggregate group to return
            max_groups: The maximum number of aggregate groups to return,
                -1 for all
            mode: Special mode for aggregation

        Returns:
            The response. See :py:meth:`send_command` for details.
        """
        if not resource_type:
            raise RequiredArgument(
                function=self.get_aggregates.__name__, argument='resource_type'
            )

        if not isinstance(resource_type, EntityType):
            raise InvalidArgumentType(
                function=self.get_aggregates.__name__,
                argument='resource_type',
                arg_type=EntityType.__name__,
            )

        cmd = XmlCommand('get_aggregates')

        _actual_resource_type = resource_type
        if resource_type.value == EntityType.AUDIT.value:
            _actual_resource_type = EntityType.TASK
            cmd.set_attribute('usage_type', 'audit')
        elif resource_type.value == EntityType.POLICY.value:
            _actual_resource_type = EntityType.SCAN_CONFIG
            cmd.set_attribute('usage_type', 'policy')
        elif resource_type.value == EntityType.SCAN_CONFIG.value:
            cmd.set_attribute('usage_type', 'scan')
        elif resource_type.value == EntityType.TASK.value:
            cmd.set_attribute('usage_type', 'scan')
        cmd.set_attribute('type', _actual_resource_type.value)

        add_filter(cmd, filter_string, filter_id)

        if first_group is not None:
            if not isinstance(first_group, int):
                raise InvalidArgumentType(
                    function=self.get_aggregates.__name__,
                    argument='first_group',
                    arg_type=int.__name__,
                )
            cmd.set_attribute('first_group', str(first_group))

        if max_groups is not None:
            if not isinstance(max_groups, int):
                raise InvalidArgumentType(
                    function=self.get_aggregates.__name__,
                    argument='max_groups',
                    arg_type=int.__name__,
                )
            cmd.set_attribute('max_groups', str(max_groups))

        if sort_criteria is not None:
            if not isinstance(sort_criteria, list):
                raise InvalidArgumentType(
                    function=self.get_aggregates.__name__,
                    argument='sort_criteria',
                    arg_type=list.__name__,
                )
            for sort in sort_criteria:
                if not isinstance(sort, dict):
                    raise InvalidArgumentType(
                        function=self.get_aggregates.__name__,
                        argument='sort_criteria',
                    )

                sort_elem = cmd.add_element('sort')
                if sort.get('field'):
                    sort_elem.set_attribute('field', sort.get('field'))

                if sort.get('stat'):
                    if isinstance(sort['stat'], AggregateStatistic):
                        sort_elem.set_attribute('stat', sort['stat'].value)
                    else:
                        stat = get_aggregate_statistic_from_string(sort['stat'])
                        sort_elem.set_attribute('stat', stat.value)

                if sort.get('order'):
                    if isinstance(sort['order'], SortOrder):
                        sort_elem.set_attribute('order', sort['order'].value)
                    else:
                        so = get_sort_order_from_string(sort['order'])
                        sort_elem.set_attribute('order', so.value)

        if data_columns is not None:
            if not isinstance(data_columns, list):
                raise InvalidArgumentType(
                    function=self.get_aggregates.__name__,
                    argument='data_columns',
                    arg_type=list.__name__,
                )
            for column in data_columns:
                cmd.add_element('data_column', column)

        if group_column is not None:
            cmd.set_attribute('group_column', group_column)

        if subgroup_column is not None:
            if not group_column:
                raise RequiredArgument(
                    '{} requires a group_column argument'
                    ' if subgroup_column is given'.format(
                        self.get_aggregates.__name__
                    ),
                    function=self.get_aggregates.__name__,
                    argument='subgroup_column',
                )
            cmd.set_attribute('subgroup_column', subgroup_column)

        if text_columns is not None:
            if not isinstance(text_columns, list):
                raise InvalidArgumentType(
                    function=self.get_aggregates.__name__,
                    argument='text_columns',
                    arg_type=list.__name__,
                )
            for column in text_columns:
                cmd.add_element('text_column', column)

        if mode is not None:
            cmd.set_attribute('mode', mode)

        # Add additional keyword args as attributes for backward compatibility.
        cmd.set_attributes(kwargs)

        return self._send_xml_command(cmd)