def currency_cell(self, amount: bc_amount._Amount, **attrs: Any) -> odf.table.TableCell:

    MetaKey,

from beancount.core.inventory import Inventory

from beancount.core.position import _Position as Position

QueryExpression = Union[

    bc_query_parser.Column,

    bc_query_parser.Constant,

    bc_query_parser.Function,

    bc_query_parser.UnaryOp,

]

    META_FNAMES = frozenset([

        'any_meta',

        'entry_meta',

        'meta',

        'meta_docs',

        'str_meta',

    ])

    def _generic_cell(self, value: Any) -> odf.table.TableCell:

        if isinstance(value, Iterable) and not isinstance(value, (str, tuple)):

            return self.multiline_cell(value)

        else:

            return self.string_cell('' if value is None else str(value))

    def _inventory_cell(self, value: Inventory) -> odf.table.TableCell:

        return self.balance_cell(core.Balance(pos.units for pos in value))

    def _link_string_cell(self, value: str) -> odf.table.TableCell:

        return self.meta_links_cell(value.split())

    def _metadata_cell(self, value: MetaValue) -> odf.table.TableCell:

        return self._cell_type(type(value))(value)

    def _position_cell(self, value: Position) -> odf.table.TableCell:

        return self.currency_cell(value.units)

        """Return a function to create a cell, for non-metadata row types."""

        if issubclass(row_type, Inventory):

            return self._inventory_cell

        elif issubclass(row_type, Position):

            return self._position_cell

        elif issubclass(row_type, BeancountAmount):

    def _link_cell_type(self, row_type: Type) -> CellFunc:

        """Return a function to create a cell from metadata with documentation links."""

        if issubclass(row_type, str):

            return self._link_string_cell

        elif issubclass(row_type, tuple):

            return self._generic_cell

        elif issubclass(row_type, Iterable):

            return self.meta_links_cell

            return self._generic_cell

    def _meta_target(self, target: QueryExpression) -> Optional[MetaKey]:

        """Return the metadata key looked up by this target, if any

        This function takes a parsed target (i.e., what we're SELECTing) and

        recurses it to see whether it's looking up any metadata. If so, it

        returns the key of that metadata. Otherwise it returns None.

        """

        if isinstance(target, bc_query_parser.UnaryOp):

            return self._meta_target(target.operand)

        elif not isinstance(target, bc_query_parser.Function):

            return None

        try:

            operand = target.operands[0]

        except IndexError:

            return None

        if (target.fname in self.META_FNAMES

            and isinstance(operand, bc_query_parser.Constant)):

            return operand.value  # type:ignore[no-any-return]

        else:

            for operand in target.operands:

                retval = self._meta_target(operand)

                if retval is not None:

                    break

            return retval

    def _cell_types(self, statement: QueryStatement, row_types: RowTypes) -> Iterator[CellFunc]:

        """Return functions to create table cells from result rows

        Given a parsed query and the types of return rows, yields a function

        to create a cell for each column in the row, in order. The returned

        functions vary in order to provide the best available formatting for

        different data types.

        """

        if (isinstance(statement, bc_query_parser.Select)

            and isinstance(statement.targets, Sequence)):

            targets = [t.expression for t in statement.targets]

        else:

            # Synthesize something that makes clear we're not loading metadata.

            targets = [bc_query_parser.Column(name) for name, _ in row_types]

        for target, (_, row_type) in zip(targets, row_types):

            meta_key = self._meta_target(target)

            if meta_key is None:

            elif meta_key in data.LINK_METADATA:

                yield self._link_cell_type(row_type)

            else:

                yield self._metadata_cell

    def write_query(self, statement: QueryStatement, row_types: RowTypes, rows: Rows) -> None:

            if issubclass(row_type, datetime.date):

                col_width = 1.0

            elif issubclass(row_type, (BeancountAmount, Inventory, Position)):

                col_width = 2.0

            self.string_cell(data.Metadata.human_name(name), stylename=self.style_bold)

        cell_funcs = list(self._cell_types(statement, row_types))

        if isinstance(value, Sequence) and not isinstance(value, (str, tuple)):

            render_func(statement, row_types, rows)

    def _render_csv(self, statement: QueryStatement, row_types: RowTypes, rows: Rows) -> None:

    def _render_ods(self, statement: QueryStatement, row_types: RowTypes, rows: Rows) -> None:

        self.ods.write_query(statement, row_types, rows)

        logger.info(

            "%s rows of results saved in sheet %s",

            len(rows),

            self.ods.sheet.getAttribute('name'),

        )

    def _render_text(self, statement: QueryStatement, row_types: RowTypes, rows: Rows) -> None:

    return parser.parse_args(arglist)

from beancount.query import query_parser as bc_query_parser

@pytest.fixture(scope='module')

def qparser():

    return bc_query_parser.Parser()

def test_ods_amount_formatting(qparser):

    statement = qparser.parse('SELECT UNITS(position)')

    ods.write_query(statement, row_types, row_source)

def test_ods_datetime_formatting(qparser):

    statement = qparser.parse('SELECT date')

    ods.write_query(statement, row_types, row_source)

@pytest.mark.parametrize('meta_key,meta_func', [

    ('check', 'ANY_META'),

    ('purchase-order', 'META'),

    ('rt-id', 'META_DOCS'),

def test_ods_link_formatting(qparser, rt, meta_key, meta_func):

    meta_func_returns_list = meta_func == 'META_DOCS'

    statement = qparser.parse(f'SELECT {meta_func}({meta_key!r}) AS docs')

    row_types = [('docs', list if meta_func_returns_list else str)]

    row_source = [

        (s.split() if meta_func_returns_list else s,)

        for s in ['rt:1/5', 'rt:3 Checks/9.pdf']

    ]

    ods.write_query(statement, row_types, row_source)

    assert next(rows).text == 'Docs'

def test_ods_meta_formatting(qparser):

    statement = qparser.parse('SELECT ANY_META("entity") AS entity')

    row_types = [('entity', object)]

    ods.write_query(statement, row_types, row_source)

    assert next(actual)[0].text == 'Entity'

def test_ods_multicolumn_write(qparser, rt):

    statement = qparser.parse(

        'SELECT MIN(date) AS date, SET(META_DOCS("rt-id")) AS tix, STR_META("entity") AS entity',

    )

    row_types = [('date', datetime.date), ('tix', set), ('entity', str)]

        (testutil.PAST_DATE, {'rt:1'}, 'AA'),

        (testutil.FY_START_DATE, {'rt:2'}, 'BB'),

        (testutil.FUTURE_DATE, {'rt:3', 'rt:4'}, 'CC'),

    ods.write_query(statement, list(row_types), list(row_source))

    for expected, _ in row_types:

        assert next(actual) == expected.title()

    assert next(actual) == 'AA'

    assert next(actual) == 'BB'

    assert frozenset(next(actual).split('\0')) == row_source[-1][1]

    assert next(actual) == 'CC'

def test_ods_is_empty(qparser):

    statement = qparser.parse('SELECT * WHERE date < 1900-01-01')

    ods.write_query(statement, [], [])

    with stdout:

        stdout.seek(0)

        ods_doc = odf.opendocument.load(stdout)

def test_ods_aggregate_output():

    books_path = testutil.test_path(f'books/books/2020.beancount')

    config = testutil.TestConfig(books_path=books_path)

    arglist = [

        '-O', '-',

        '-f', 'ods',

        'SELECT account, SET(narration), SUM(UNITS(position))',

        'WHERE date >= 2020-04-01 AND date <= 2020-04-02',

        'GROUP BY account ORDER BY account ASC',

    ]

    returncode, stdout, stderr = pipe_main(arglist, config, io.BytesIO)

    assert returncode == 0

    with stdout:

        stdout.seek(0)

        ods_doc = odf.opendocument.load(stdout)

    rows = iter(ods_doc.spreadsheet.firstChild.getElementsByType(odf.table.TableRow))

    next(rows)  # Skip header row

    actual = {}

    for row in rows:

        acct, descs, balance = row.childNodes

        actual[acct.text] = (frozenset(descs.text.split('\0')), balance.text)

    in_desc = {'2020 donation'}

    ex_desc = {'2020 bank maintenance fee'}

    assert actual['Income:Donations'] == (in_desc, '$20.20')

    assert actual['Expenses:BankingFees'] == (ex_desc, '$1.00')

    assert actual['Assets:Checking'] == (in_desc | ex_desc, '($21.20)')