python - 在 sqlalchemy select 对象中测试 where 子句

Question

我正在尝试编写一些函数来构建 sqlalchemy select 语句。例如:

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata

t_test = sa.Table(
    'test', metadata,
    sa.Column('col_1', sa.Text),
    sa.Column('col_2', sa.Float)
)

def create_test_select():
    sa_select = sa.select([t_test.c.col_1, t_test.c.col_2])
    return sa_select

def add_test_col1_where_clause(sa_select, x ):
    sa_select = sa_select.where(t_test.c.col_1 == x)
    return sa_select

我想测试这些功能。

要测试 create_test_select 我会写类似的东西

class Test(unittest.TestCase):    
    def test(self):
         self.assertIn('col1', create_test_select().columns)
         self.assertIn('col2', create_test_select().columns)

如何测试函数 add_test_col1_where_clause？我想知道它向 select 添加了正确的 where 子句。我最初的想法是检查 sqlachemy select object 中的 where 子句，但我无法让它工作。

Answer 1

SQLAlchemy 不直接公开选择的“where”子句部分；毕竟不是所有的 SELECT 语句都有。此外，该条款可能变得相当复杂。就个人而言，我只会在集成测试中测试表达式，并且只是为了确保返回正确的数据。

SQLAlchemy 确实为您提供了访问对象树的工具，尽管文档有些不足。您可以使用它从树中提取任何比较，所以 <left> <op> <right> left 的表达式和 right是列元素:

from sqlalchemy.sql import visitors, ColumnElement

def comparison_visitor(expr, callback):
    """Finds all binary operators, and calls callback(op, left, right)"""
    def visit_binary(op):
        callback(op, op.left, op.right)
    # visit each expr element, but for select clauses, ignore the column collection
    visitors.traverse(expr, {'column_collections': False}, {'binary': visit_binary})

visitors.traverse()遍历任何 SQLAlchemy 表达式(通过重复调用对象的 ColumnClause.get_children() method，将作为第二个参数给出的映射传递给 traverse())，并调用匹配 __visit_name__ 的函数每个对象的属性。 BinaryClause对象有 binary作为访问名称。

然后您可以使用它来测试是否存在特定条件:

from sqlalchemy.sql import operators
from sqlalchemy.sql.expressions import BindParameter
from sqlalchemy.sql.schema import Column

def test_where(self):
    where = add_test_col1_where_clause(create_test_select(), 'foo bar')

    def test_comparison(op, left, right):
        self.assertIs(op.operator, operators.eq)    # == test

        self.assertIsInstance(left, Column)         # between the col_1 column
        self.assertEq(left.name, 'col_1')

        self.assertIsInstance(right, BindParameter) # and a parameter
        self.assertEq(right.value, 'foo bar')       # with value 'foo bar'

    comparison_visitor(where, test_comparison)