BaseStructuredView

Tip

• To understand the general idea better, visit Concept: Structured Views
• To learn how to support custom data sources, visit How-to: create your custom view page

dbally.views.structured.BaseStructuredView

Bases: BaseView

Base class for all structured Views. All classes implementing this interface has to be able to list all available filters, apply them and execute queries.

list_few_shots

list_few_shots() -> List[FewShotExample]

List all examples to be injected into few-shot prompt.

RETURNS	DESCRIPTION
List[FewShotExample]	List of few-shot examples

Source code in src/dbally/views/base.py

def list_few_shots(self) -> List[FewShotExample]:
    """
    List all examples to be injected into few-shot prompt.

    Returns:
        List of few-shot examples
    """
    return []

get_iql_generator

get_iql_generator() -> IQLGenerator

Returns the IQL generator for the view.

RETURNS	DESCRIPTION
IQLGenerator	IQL generator for the view.

Source code in src/dbally/views/structured.py

def get_iql_generator(self) -> IQLGenerator:
    """
    Returns the IQL generator for the view.

    Returns:
        IQL generator for the view.
    """
    return IQLGenerator()

ask async

ask(query: str, llm: LLM, event_tracker: Optional[EventTracker] = None, n_retries: int = 3, dry_run: bool = False, llm_options: Optional[LLMOptions] = None) -> ViewExecutionResult

Executes the query and returns the result. It generates the IQL query from the natural language query and applies the filters to the view. It retries the process in case of errors.

PARAMETER	DESCRIPTION
query	The natural language query to execute. TYPE: str
llm	The LLM used to execute the query. TYPE: LLM
event_tracker	The event tracker used to audit the query execution. TYPE: Optional[EventTracker] DEFAULT: None
n_retries	The number of retries to execute the query in case of errors. TYPE: int DEFAULT: 3
dry_run	If True, the query will not be used to fetch data from the datasource. TYPE: bool DEFAULT: False
llm_options	Options to use for the LLM. TYPE: Optional[LLMOptions] DEFAULT: None

RETURNS	DESCRIPTION
ViewExecutionResult	The result of the query.

RAISES	DESCRIPTION
ViewExecutionError	When an error occurs while executing the view.

Source code in src/dbally/views/structured.py

async def ask(
    self,
    query: str,
    llm: LLM,
    event_tracker: Optional[EventTracker] = None,
    n_retries: int = 3,
    dry_run: bool = False,
    llm_options: Optional[LLMOptions] = None,
) -> ViewExecutionResult:
    """
    Executes the query and returns the result. It generates the IQL query from the natural language query\
    and applies the filters to the view. It retries the process in case of errors.

    Args:
        query: The natural language query to execute.
        llm: The LLM used to execute the query.
        event_tracker: The event tracker used to audit the query execution.
        n_retries: The number of retries to execute the query in case of errors.
        dry_run: If True, the query will not be used to fetch data from the datasource.
        llm_options: Options to use for the LLM.

    Returns:
        The result of the query.

    Raises:
        ViewExecutionError: When an error occurs while executing the view.
    """
    filters = self.list_filters()
    examples = self.list_few_shots()
    aggregations = self.list_aggregations()

    iql_generator = self.get_iql_generator()
    iql = await iql_generator(
        question=query,
        filters=filters,
        aggregations=aggregations,
        examples=examples,
        llm=llm,
        event_tracker=event_tracker,
        llm_options=llm_options,
        n_retries=n_retries,
    )

    if iql.failed:
        raise ViewExecutionError(
            view_name=self.__class__.__name__,
            iql=iql,
        )

    if iql.filters:
        await self.apply_filters(iql.filters)

    if iql.aggregation:
        await self.apply_aggregation(iql.aggregation)

    result = self.execute(dry_run=dry_run)
    result.context["iql"] = {
        "filters": str(iql.filters) if iql.filters else None,
        "aggregation": str(iql.aggregation) if iql.aggregation else None,
    }
    return result

list_filters abstractmethod

list_filters() -> List[ExposedFunction]

Lists all available filters for the View.

RETURNS	DESCRIPTION
List[ExposedFunction]	Filters defined inside the View.

Source code in src/dbally/views/structured.py

@abc.abstractmethod
def list_filters(self) -> List[ExposedFunction]:
    """
    Lists all available filters for the View.

    Returns:
        Filters defined inside the View.
    """

list_aggregations abstractmethod

list_aggregations() -> List[ExposedFunction]

Lists all available aggregations for the View.

RETURNS	DESCRIPTION
List[ExposedFunction]	Aggregations defined inside the View.

Source code in src/dbally/views/structured.py

@abc.abstractmethod
def list_aggregations(self) -> List[ExposedFunction]:
    """
    Lists all available aggregations for the View.

    Returns:
        Aggregations defined inside the View.
    """

apply_filters abstractmethod async

apply_filters(filters: IQLFiltersQuery) -> None

Applies the chosen filters to the view.

PARAMETER	DESCRIPTION
filters	IQLQuery object representing the filters to apply. TYPE: IQLFiltersQuery

Source code in src/dbally/views/structured.py

@abc.abstractmethod
async def apply_filters(self, filters: IQLFiltersQuery) -> None:
    """
    Applies the chosen filters to the view.

    Args:
        filters: IQLQuery object representing the filters to apply.
    """

apply_aggregation abstractmethod async

apply_aggregation(aggregation: IQLAggregationQuery) -> None

Applies the chosen aggregation to the view.

PARAMETER	DESCRIPTION
aggregation	IQLQuery object representing the aggregation to apply. TYPE: IQLAggregationQuery

Source code in src/dbally/views/structured.py

@abc.abstractmethod
async def apply_aggregation(self, aggregation: IQLAggregationQuery) -> None:
    """
    Applies the chosen aggregation to the view.

    Args:
        aggregation: IQLQuery object representing the aggregation to apply.
    """

execute abstractmethod

execute(dry_run: bool = False) -> ViewExecutionResult

Executes the query and returns the result.

PARAMETER	DESCRIPTION
dry_run	if True, should only generate the query without executing it. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
ViewExecutionResult	The view execution result.

Source code in src/dbally/views/structured.py

@abc.abstractmethod
def execute(self, dry_run: bool = False) -> ViewExecutionResult:
    """
    Executes the query and returns the result.

    Args:
        dry_run: if True, should only generate the query without executing it.

    Returns:
        The view execution result.
    """

list_similarity_indexes

list_similarity_indexes() -> Dict[AbstractSimilarityIndex, List[IndexLocation]]

Lists all the similarity indexes used by the view.

RETURNS	DESCRIPTION
Dict[AbstractSimilarityIndex, List[IndexLocation]]	Mapping of similarity indexes to their locations in the (view_name, filter_name, argument_name) format.

Source code in src/dbally/views/structured.py

def list_similarity_indexes(self) -> Dict[AbstractSimilarityIndex, List[IndexLocation]]:
    """
    Lists all the similarity indexes used by the view.

    Returns:
        Mapping of similarity indexes to their locations in the (view_name, filter_name, argument_name) format.
    """
    indexes = defaultdict(list)
    filters = self.list_filters()
    for filter_ in filters:
        for param in filter_.parameters:
            if param.similarity_index:
                indexes[param.similarity_index].append((self.__class__.__name__, filter_.name, param.name))
    return indexes

dbally.views.exposed_functions.ExposedFunction dataclass

ExposedFunction(name: str, description: str, parameters: List[MethodParamWithTyping])

Represents a function exposed to the AI model.

name instance-attribute

name: str

description instance-attribute

description: str

parameters instance-attribute

parameters: List[MethodParamWithTyping]

dbally.views.exposed_functions.MethodParamWithTyping dataclass

MethodParamWithTyping(name: str, type: Union[type, _GenericAlias])

Represents a method parameter with its type.

name instance-attribute

name: str

type instance-attribute

type: Union[type, _GenericAlias]

similarity_index property

similarity_index: Optional[AbstractSimilarityIndex]

Returns the SimilarityIndex object if the type is annotated with it.

dbally.views.methods_base.MethodsBaseView

Bases: BaseStructuredView, ABC

Base class for views that use view methods to expose filters.

HIDDEN_ARGUMENTS class-attribute instance-attribute

HIDDEN_ARGUMENTS = ['cls', 'self', 'return']

ask async

ask(query: str, llm: LLM, event_tracker: Optional[EventTracker] = None, n_retries: int = 3, dry_run: bool = False, llm_options: Optional[LLMOptions] = None) -> ViewExecutionResult

Executes the query and returns the result. It generates the IQL query from the natural language query and applies the filters to the view. It retries the process in case of errors.

PARAMETER	DESCRIPTION
query	The natural language query to execute. TYPE: str
llm	The LLM used to execute the query. TYPE: LLM
event_tracker	The event tracker used to audit the query execution. TYPE: Optional[EventTracker] DEFAULT: None
n_retries	The number of retries to execute the query in case of errors. TYPE: int DEFAULT: 3
dry_run	If True, the query will not be used to fetch data from the datasource. TYPE: bool DEFAULT: False
llm_options	Options to use for the LLM. TYPE: Optional[LLMOptions] DEFAULT: None

RETURNS	DESCRIPTION
ViewExecutionResult	The result of the query.

RAISES	DESCRIPTION
ViewExecutionError	When an error occurs while executing the view.

Source code in src/dbally/views/structured.py

async def ask(
    self,
    query: str,
    llm: LLM,
    event_tracker: Optional[EventTracker] = None,
    n_retries: int = 3,
    dry_run: bool = False,
    llm_options: Optional[LLMOptions] = None,
) -> ViewExecutionResult:
    """
    Executes the query and returns the result. It generates the IQL query from the natural language query\
    and applies the filters to the view. It retries the process in case of errors.

    Args:
        query: The natural language query to execute.
        llm: The LLM used to execute the query.
        event_tracker: The event tracker used to audit the query execution.
        n_retries: The number of retries to execute the query in case of errors.
        dry_run: If True, the query will not be used to fetch data from the datasource.
        llm_options: Options to use for the LLM.

    Returns:
        The result of the query.

    Raises:
        ViewExecutionError: When an error occurs while executing the view.
    """
    filters = self.list_filters()
    examples = self.list_few_shots()
    aggregations = self.list_aggregations()

    iql_generator = self.get_iql_generator()
    iql = await iql_generator(
        question=query,
        filters=filters,
        aggregations=aggregations,
        examples=examples,
        llm=llm,
        event_tracker=event_tracker,
        llm_options=llm_options,
        n_retries=n_retries,
    )

    if iql.failed:
        raise ViewExecutionError(
            view_name=self.__class__.__name__,
            iql=iql,
        )

    if iql.filters:
        await self.apply_filters(iql.filters)

    if iql.aggregation:
        await self.apply_aggregation(iql.aggregation)

    result = self.execute(dry_run=dry_run)
    result.context["iql"] = {
        "filters": str(iql.filters) if iql.filters else None,
        "aggregation": str(iql.aggregation) if iql.aggregation else None,
    }
    return result

list_similarity_indexes

list_similarity_indexes() -> Dict[AbstractSimilarityIndex, List[IndexLocation]]

Lists all the similarity indexes used by the view.

RETURNS	DESCRIPTION
Dict[AbstractSimilarityIndex, List[IndexLocation]]	Mapping of similarity indexes to their locations in the (view_name, filter_name, argument_name) format.

Source code in src/dbally/views/structured.py

def list_similarity_indexes(self) -> Dict[AbstractSimilarityIndex, List[IndexLocation]]:
    """
    Lists all the similarity indexes used by the view.

    Returns:
        Mapping of similarity indexes to their locations in the (view_name, filter_name, argument_name) format.
    """
    indexes = defaultdict(list)
    filters = self.list_filters()
    for filter_ in filters:
        for param in filter_.parameters:
            if param.similarity_index:
                indexes[param.similarity_index].append((self.__class__.__name__, filter_.name, param.name))
    return indexes

list_few_shots

list_few_shots() -> List[FewShotExample]

List all examples to be injected into few-shot prompt.

RETURNS	DESCRIPTION
List[FewShotExample]	List of few-shot examples

Source code in src/dbally/views/base.py

def list_few_shots(self) -> List[FewShotExample]:
    """
    List all examples to be injected into few-shot prompt.

    Returns:
        List of few-shot examples
    """
    return []

get_iql_generator

get_iql_generator() -> IQLGenerator

Returns the IQL generator for the view.

RETURNS	DESCRIPTION
IQLGenerator	IQL generator for the view.

Source code in src/dbally/views/structured.py

def get_iql_generator(self) -> IQLGenerator:
    """
    Returns the IQL generator for the view.

    Returns:
        IQL generator for the view.
    """
    return IQLGenerator()

apply_filters abstractmethod async

apply_filters(filters: IQLFiltersQuery) -> None

Applies the chosen filters to the view.

PARAMETER	DESCRIPTION
filters	IQLQuery object representing the filters to apply. TYPE: IQLFiltersQuery

Source code in src/dbally/views/structured.py

@abc.abstractmethod
async def apply_filters(self, filters: IQLFiltersQuery) -> None:
    """
    Applies the chosen filters to the view.

    Args:
        filters: IQLQuery object representing the filters to apply.
    """

apply_aggregation abstractmethod async

apply_aggregation(aggregation: IQLAggregationQuery) -> None

Applies the chosen aggregation to the view.

PARAMETER	DESCRIPTION
aggregation	IQLQuery object representing the aggregation to apply. TYPE: IQLAggregationQuery

Source code in src/dbally/views/structured.py

@abc.abstractmethod
async def apply_aggregation(self, aggregation: IQLAggregationQuery) -> None:
    """
    Applies the chosen aggregation to the view.

    Args:
        aggregation: IQLQuery object representing the aggregation to apply.
    """

execute abstractmethod

execute(dry_run: bool = False) -> ViewExecutionResult

Executes the query and returns the result.

PARAMETER	DESCRIPTION
dry_run	if True, should only generate the query without executing it. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
ViewExecutionResult	The view execution result.

Source code in src/dbally/views/structured.py

@abc.abstractmethod
def execute(self, dry_run: bool = False) -> ViewExecutionResult:
    """
    Executes the query and returns the result.

    Args:
        dry_run: if True, should only generate the query without executing it.

    Returns:
        The view execution result.
    """

list_methods_by_decorator classmethod

list_methods_by_decorator(decorator: Callable) -> List[ExposedFunction]

Lists all methods decorated with the given decorator.

PARAMETER	DESCRIPTION
decorator	The decorator to filter the methods TYPE: Callable

RETURNS	DESCRIPTION
List[ExposedFunction]	List of exposed methods

Source code in src/dbally/views/methods_base.py

@classmethod
def list_methods_by_decorator(cls, decorator: Callable) -> List[ExposedFunction]:
    """
    Lists all methods decorated with the given decorator.

    Args:
        decorator: The decorator to filter the methods

    Returns:
        List of exposed methods
    """
    methods = []
    for method_name in dir(cls):
        method = getattr(cls, method_name)
        if (
            hasattr(method, "_methodDecorator")
            and method._methodDecorator == decorator  # pylint: disable=protected-access
        ):
            annotations = method.__annotations__.items()
            methods.append(
                ExposedFunction(
                    name=method_name,
                    description=textwrap.dedent(method.__doc__).strip() if method.__doc__ else "",
                    parameters=[
                        MethodParamWithTyping(n, t) for n, t in annotations if n not in cls.HIDDEN_ARGUMENTS
                    ],
                )
            )
    return methods

list_filters

list_filters() -> List[ExposedFunction]

List filters in the given view

RETURNS	DESCRIPTION
List[ExposedFunction]	Filters defined inside the View and decorated with decorators.view_filter.

Source code in src/dbally/views/methods_base.py

def list_filters(self) -> List[ExposedFunction]:
    """
    List filters in the given view

    Returns:
        Filters defined inside the View and decorated with `decorators.view_filter`.
    """
    return self.list_methods_by_decorator(decorators.view_filter)

list_aggregations

list_aggregations() -> List[ExposedFunction]

List aggregations in the given view

RETURNS	DESCRIPTION
List[ExposedFunction]	Aggregations defined inside the View and decorated with decorators.view_aggregation.

Source code in src/dbally/views/methods_base.py

def list_aggregations(self) -> List[ExposedFunction]:
    """
    List aggregations in the given view

    Returns:
        Aggregations defined inside the View and decorated with `decorators.view_aggregation`.
    """
    return self.list_methods_by_decorator(decorators.view_aggregation)

call_filter_method async

call_filter_method(func: FunctionCall) -> Any

Converts a IQL FunctonCall filter to a method call. If the method is a coroutine, it will be awaited.

PARAMETER	DESCRIPTION
func	IQL FunctionCall node TYPE: FunctionCall

RETURNS	DESCRIPTION
Any	The result of the method call

Source code in src/dbally/views/methods_base.py

async def call_filter_method(self, func: syntax.FunctionCall) -> Any:
    """
    Converts a IQL FunctonCall filter to a method call. If the method is a coroutine, it will be awaited.

    Args:
        func: IQL FunctionCall node

    Returns:
        The result of the method call
    """
    method, args = self._method_with_args_from_call(func, decorators.view_filter)
    return await self._call_method(method, args)

call_aggregation_method async

call_aggregation_method(func: FunctionCall) -> Any

Converts a IQL FunctonCall aggregation to a method call. If the method is a coroutine, it will be awaited.

PARAMETER	DESCRIPTION
func	IQL FunctionCall node TYPE: FunctionCall

RETURNS	DESCRIPTION
Any	The result of the method call

Source code in src/dbally/views/methods_base.py

async def call_aggregation_method(self, func: syntax.FunctionCall) -> Any:
    """
    Converts a IQL FunctonCall aggregation to a method call. If the method is a coroutine, it will be awaited.

    Args:
        func: IQL FunctionCall node

    Returns:
        The result of the method call
    """
    method, args = self._method_with_args_from_call(func, decorators.view_aggregation)
    return await self._call_method(method, args)