SqlAlchemyBaseView

Tip

To learn how to use db-ally with databases see How-To: Use SQL databases with db-ally.

dbally.SqlAlchemyBaseView

SqlAlchemyBaseView(sqlalchemy_engine: Engine)

Bases: MethodsBaseView

Base class for views that use SQLAlchemy to generate SQL queries.

Creates a new instance of the SQL view.

PARAMETER	DESCRIPTION
sqlalchemy_engine	SQLAlchemy engine to use for executing the queries. TYPE: Engine

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

def __init__(self, sqlalchemy_engine: sqlalchemy.Engine) -> None:
    """
    Creates a new instance of the SQL view.

    Args:
        sqlalchemy_engine: SQLAlchemy engine to use for executing the queries.
    """
    super().__init__()
    self.select = self.get_select()
    self._sqlalchemy_engine = sqlalchemy_engine

HIDDEN_ARGUMENTS class-attribute instance-attribute

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

select instance-attribute

select = get_select()

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()

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)

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

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)

get_select abstractmethod

get_select() -> Select

Creates the initial SqlAlchemy select object which will be used to build the query.

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

@abc.abstractmethod
def get_select(self) -> sqlalchemy.Select:
    """
    Creates the initial
    [SqlAlchemy select object
    ](https://docs.sqlalchemy.org/en/20/core/selectable.html#sqlalchemy.sql.expression.Select)
    which will be used to build the query.
    """

apply_filters 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/sqlalchemy_base.py

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

    Args:
        filters: IQLQuery object representing the filters to apply.
    """
    self.select = self.select.where(await self._build_filter_node(filters.root))

apply_aggregation 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/sqlalchemy_base.py

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

    Args:
        aggregation: IQLQuery object representing the aggregation to apply.
    """
    self.select = await self.call_aggregation_method(aggregation.root)

execute

execute(dry_run: bool = False) -> ViewExecutionResult

Executes the generated SQL query and returns the results.

PARAMETER	DESCRIPTION
dry_run	If True, only adds the SQL query to the context field without executing the query. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
ViewExecutionResult	Results of the query where results will be a list of dictionaries representing retrieved rows or an empty list if dry_run is set to True. Inside the context field the generated sql will be stored.

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

def execute(self, dry_run: bool = False) -> ViewExecutionResult:
    """
    Executes the generated SQL query and returns the results.

    Args:
        dry_run: If True, only adds the SQL query to the context field without executing the query.

    Returns:
        Results of the query where `results` will be a list of dictionaries representing retrieved rows or an empty\
        list if `dry_run` is set to `True`. Inside the `context` field the generated sql will be stored.
    """
    results = []
    sql = str(self.select.compile(bind=self._sqlalchemy_engine, compile_kwargs={"literal_binds": True}))

    if not dry_run:
        with self._sqlalchemy_engine.connect() as connection:
            rows = connection.execute(self.select).fetchall()
            # The underscore is used by sqlalchemy to avoid conflicts with column names
            # pylint: disable=protected-access
            results = [dict(row._mapping) for row in rows]

    return ViewExecutionResult(
        results=results,
        context={"sql": sql},
    )