orchestrata_tools.query.evaluator

Reference › orchestrata_tools › query › evaluator

QueryEvaluator runs a Query from dsl against an ASTIndex, optionally with a TestMatcher from matcher.

evaluate selects a pool by the query target (methods, classes, modules or functions; any other target raises ValueError), keeps the objects for which _matches is true for every filter, and truncates to the limit when one is set. _matches handles __gt, __lt and __contains suffixes and plain equality on attributes, and an object missing the attribute does not match. Two filters are special: public compares with the object's is_public, and tested asks _is_tested, which computes the set of tested method keys once with the matcher and caches it. Without a matcher, no method counts as tested.

At a glance

Class diagram

Dependency diagram

API

Classes

QueryEvaluator

Line 20

Evaluate Query objects against an indexed codebase.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_tools/query/evaluator.py

"""Execute Query objects against an ASTIndex."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from orchestrata_tools.models.ast_models import MethodInfo
from orchestrata_tools.query.dsl import Query

if TYPE_CHECKING:
    from orchestrata_tools.models.ast_models import (
        ClassInfo,
        FunctionInfo,
        ModuleInfo,
    )
    from orchestrata_tools.parser.indexer import ASTIndex
    from orchestrata_tools.test_analyzer.matcher import TestMatcher


class QueryEvaluator:
    """Evaluate Query objects against an indexed codebase."""

    def __init__(self, index: ASTIndex, matcher: TestMatcher | None = None):
        self.index = index
        self.matcher = matcher
        self._tested_keys: set[str] | None = None

    def _is_tested(self, method: MethodInfo) -> bool:
        if self.matcher is None:
            return False
        if self._tested_keys is None:
            self._tested_keys = self.matcher.find_tested_methods(self.index)
        key = f"{method.module_path}.{method.class_name}.{method.name}"
        return key in self._tested_keys

    def _matches(self, obj: Any, filters: dict[str, Any]) -> bool:
        for key, val in filters.items():
            if key.endswith("__gt"):
                attr = key[:-4]
                if not hasattr(obj, attr) or getattr(obj, attr) <= val:
                    return False
            elif key.endswith("__lt"):
                attr = key[:-4]
                if not hasattr(obj, attr) or getattr(obj, attr) >= val:
                    return False
            elif key.endswith("__contains"):
                attr = key[:-10]
                if not hasattr(obj, attr) or val not in getattr(obj, attr):
                    return False
            else:
                if key == "tested":
                    if isinstance(obj, MethodInfo) and self._is_tested(obj) != val:
                        return False
                    continue
                if key == "public":
                    is_public = hasattr(obj, "is_public") and getattr(obj, "is_public")
                    if is_public != val:
                        return False
                    continue
                if not hasattr(obj, key) or getattr(obj, key) != val:
                    return False
        return True

    def evaluate(self, query: Query) -> list[Any]:
        """Execute a query and return matching objects."""
        pool: list[MethodInfo] | list[ClassInfo] | list[ModuleInfo] | list[FunctionInfo]
        if query.target == "methods":
            pool = self.index.codebase.all_methods
        elif query.target == "classes":
            pool = self.index.codebase.all_classes
        elif query.target == "modules":
            pool = self.index.codebase.modules
        elif query.target == "functions":
            pool = self.index.codebase.all_functions
        else:
            raise ValueError(f"Unknown target: {query.target}")

        results = [obj for obj in pool if self._matches(obj, query.filters)]
        if query.limit > 0:
            results = results[: query.limit]
        return results