Query

Provides models for representing the results of running a query.

QueryColumn (BaseModel) pydantic-model

A model representing a single column from a query response.

Attributes:

Name	Type	Description
name	str	The name of the column.
type	str	The type of the column.

Source code in bdantic/models/query.py

class QueryColumn(BaseModel):
    """A model representing a single column from a query response.

    Attributes:
        name: The name of the column.
        type: The type of the column.
    """

    name: str
    type: str

QueryResult (Base) pydantic-model

A model representing the result from a beancount query.

The constructor of this model accepts the value returned from executing a beancount query using the beancount.query.query.run_query function. The result is a tuple of columns and rows which this model represents in the columns and rows fields accordingly. Like all models, any data types which can be parsed from the result into models are automatically parsed.

Attributes:

Name	Type	Description
columns	List[bdantic.models.query.QueryColumn]	The columns denoting the name and types of the resulting data.
rows	List[Dict[str, Any]]	The data rows returned from the query.

Source code in bdantic/models/query.py

class QueryResult(Base):
    """A model representing the result from a beancount query.

    The constructor of this model accepts the value returned from executing a
    beancount query using the `beancount.query.query.run_query` function. The
    result is a tuple of columns and rows which this model represents in the
    `columns` and `rows` fields accordingly. Like all models, any data types
    which can be parsed from the result into models are automatically parsed.

    Attributes:
        columns: The columns denoting the name and types of the resulting data.
        rows: The data rows returned from the query.
    """

    ty: Literal["QueryResult"] = "QueryResult"
    columns: List[QueryColumn]
    rows: List[QueryRow]

    @classmethod
    def parse(
        cls, obj: Tuple[List[Tuple[str, Type]], List[Any]]
    ) -> QueryResult:
        """Parses a beancount query result into this model

        Args:
            obj: The Beancount query result

        Returns:
            A new instance of this model
        """
        columns: List[QueryColumn] = []
        for column in obj[0]:
            columns.append(
                QueryColumn(name=column[0], type=column[1].__name__)
            )

        rows: List[QueryRow] = []
        for row in obj[1]:
            d = row._asdict()
            for k, v in d.items():
                if type(v) in type_map.keys():
                    d[k] = type_map[type(v)].parse(v)

            rows.append(d)

        return QueryResult(columns=columns, rows=rows)

    def export(self) -> Tuple[List[Tuple[str, Type]], List[Any]]:
        """Exports this model into a beancount query result

        Returns:
            A new instance of a beancount query result
        """
        column_names = [v.name for v in self.columns]
        ResultRow = collections.namedtuple(  # type: ignore
            "ResultRow",
            column_names,
        )

        columns: List[Tuple[str, Type]] = []
        for column in self.columns:
            columns.append((column.name, _map[column.type]))

        rows: List[ResultRow] = []
        for row in self.rows:
            values = []
            for key in column_names:
                if type(row[key]) in type_map.values():
                    values.append(row[key].export())
                else:
                    values.append(row[key])
            rows.append(ResultRow._make(values))

        return (columns, rows)

export(self)

Exports this model into a beancount query result

Returns:

Type	Description
Tuple[List[Tuple[str, Type]], List[Any]]	A new instance of a beancount query result

Source code in bdantic/models/query.py

def export(self) -> Tuple[List[Tuple[str, Type]], List[Any]]:
    """Exports this model into a beancount query result

    Returns:
        A new instance of a beancount query result
    """
    column_names = [v.name for v in self.columns]
    ResultRow = collections.namedtuple(  # type: ignore
        "ResultRow",
        column_names,
    )

    columns: List[Tuple[str, Type]] = []
    for column in self.columns:
        columns.append((column.name, _map[column.type]))

    rows: List[ResultRow] = []
    for row in self.rows:
        values = []
        for key in column_names:
            if type(row[key]) in type_map.values():
                values.append(row[key].export())
            else:
                values.append(row[key])
        rows.append(ResultRow._make(values))

    return (columns, rows)

parse(obj) classmethod

Parses a beancount query result into this model

Parameters:

Name	Type	Description	Default
obj	Tuple[List[Tuple[str, Type]], List[Any]]	The Beancount query result	required

Returns:

Type	Description
QueryResult	A new instance of this model

Source code in bdantic/models/query.py

@classmethod
def parse(
    cls, obj: Tuple[List[Tuple[str, Type]], List[Any]]
) -> QueryResult:
    """Parses a beancount query result into this model

    Args:
        obj: The Beancount query result

    Returns:
        A new instance of this model
    """
    columns: List[QueryColumn] = []
    for column in obj[0]:
        columns.append(
            QueryColumn(name=column[0], type=column[1].__name__)
        )

    rows: List[QueryRow] = []
    for row in obj[1]:
        d = row._asdict()
        for k, v in d.items():
            if type(v) in type_map.keys():
                d[k] = type_map[type(v)].parse(v)

        rows.append(d)

    return QueryResult(columns=columns, rows=rows)