data_algebra/test_util.py
def check_transform_on_handles(
    *,
    ops,
    data,
    expect,
    db_handles,
    float_tol: float = 1e-8,
    check_column_order: bool = False,
    cols_case_sensitive: bool = False,
    check_row_order: bool = False,
    check_parse: bool = True,
    local_data_model=None,
    empty_produces_empty: bool = True,
) -> None:
    """
    Test an operator dag produces the expected result, and parses correctly.
    Asserts if there are issues

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param db_handles:  list of database handles to use in testing
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param local_data_model: optional alternate evaluation model
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: None, assert if there is an issue
    """

    # convert single table to dictionary
    if not isinstance(data, dict):
        cols_used = ops.columns_used()
        table_name = [k for k in cols_used.keys()][0]
        data = {table_name: data}
    assert isinstance(data, dict)
    n_tables = len(data)
    assert n_tables > 0
    if local_data_model is None:
        local_data_model = data_algebra.default_data_model
    assert isinstance(ops, ViewRepresentation)
    if not local_data_model.is_appropriate_data_instance(expect):
        raise TypeError("expected expect to be a local_data_model.pd.DataFrame")
    cols_used = ops.columns_used()
    if len(cols_used) < 1:
        raise ValueError("no tables used")
    # check all needed tables are present
    for k in cols_used.keys():
        v = data[k]
        assert local_data_model.is_appropriate_data_instance(v)
    if check_parse:
        if not formats_to_self(ops):
            raise ValueError("ops did not round-trip format")
    res = ops.eval(data_map=data)
    # try pandas path
    if not local_data_model.is_appropriate_data_instance(res):
        raise ValueError(
            "expected res to be local_data_model.pd.DataFrame, got: " + str(type(res))
        )
    if not equivalent_frames(
        res,
        expect,
        float_tol=float_tol,
        check_column_order=check_column_order,
        cols_case_sensitive=cols_case_sensitive,
        check_row_order=check_row_order,
    ):
        raise ValueError("Pandas eval result did not match expect")
    if len(data) == 1:
        res_t = ops.transform(list(data.values())[0])
        if not equivalent_frames(
            res_t,
            expect,
            float_tol=float_tol,
            check_column_order=check_column_order,
            cols_case_sensitive=cols_case_sensitive,
            check_row_order=check_row_order,
        ):
            raise ValueError("Pandas transform result did not match expect")
    # try on empty inputs
    empty_map = {k: v.iloc[range(0), :].reset_index(drop=True, inplace=False) for k, v in data.items()}
    empty_res = ops.eval(empty_map)
    assert local_data_model.is_appropriate_data_instance(empty_res)
    assert set(empty_res.columns) == set(res.columns)
    if empty_produces_empty:
        assert empty_res.shape[0] == 0
    else:
        assert empty_res.shape[0] > 0
    # try on combinations of empty and original
    if n_tables == 2:
        keys = [k for k in data.keys()]
        partial_maps = [
            {
                keys[0]: data[keys[0]].reset_index(drop=True, inplace=False),
                keys[1]: data[keys[1]],
            },
            {
                keys[0]: data[keys[0]],
                keys[1]: data[keys[1]].reset_index(drop=True, inplace=False),
            },
        ]
        for pm in partial_maps:
            empty_res_i = ops.eval(pm)
            assert set(empty_res_i.columns) == set(res.columns)
    # try any db paths
    if db_handles is not None:
        for db_handle in db_handles:
            to_del = set()
            sql_statements = []
            for initial_commas in [True, False]:
                for use_with in [True, False]:
                    sql_format_options = data_algebra.db_model.SQLFormatOptions(
                        use_with=use_with,
                        annotate=True,
                        sql_indent=" ",
                        initial_commas=initial_commas,
                    )
                    sql = db_handle.to_sql(ops, sql_format_options=sql_format_options,)
                    assert isinstance(sql, str)
                    sql_statements.append(sql)
                    # print(sql)
            if db_handle.conn is not None:
                for (k, v) in data.items():
                    db_handle.insert_table(v, table_name=k, allow_overwrite=True)
                    to_del.add(k)
                caught = None
                res_db_sql = []
                res_db_ops = None
                try:
                    for sql in sql_statements:
                        res_db_sql_i = db_handle.read_query(sql)
                        res_db_sql.append(res_db_sql_i)
                    res_db_ops = db_handle.read_query(ops)
                except Exception as e:
                    caught = e
                for k in to_del:
                    db_handle.drop_table(k)
                if caught is not None:
                    raise caught
                for res in res_db_sql:
                    if not equivalent_frames(
                        res,
                        expect,
                        float_tol=float_tol,
                        check_column_order=check_column_order,
                        cols_case_sensitive=cols_case_sensitive,
                        check_row_order=check_row_order,
                    ):
                        raise ValueError(f"{db_handle} SQL result did not match expect")
                if not equivalent_frames(
                    res_db_ops,
                    expect,
                    float_tol=float_tol,
                    check_column_order=check_column_order,
                    cols_case_sensitive=cols_case_sensitive,
                    check_row_order=check_row_order,
                ):
                    raise ValueError(f"{db_handle} ops result did not match expect")

    Test an operator dag produces the expected result, and parses correctly.
    Asserts if there are issues

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param db_handles:  list of database handles to use in testing
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param local_data_model: optional alternate evaluation model
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: None, assert if there is an issue
    
Test an operator dag produces the expected result, and parses correctly.
Asserts if there are issues
Test an operator dag produces the expected result and parses correctly . Asserts if there are issues
def check_transform_on_handles(
    *,
    ops,
    data,
    expect,
    db_handles,
    float_tol: float = 1e-8,
    check_column_order: bool = False,
    cols_case_sensitive: bool = False,
    check_row_order: bool = False,
    check_parse: bool = True,
    local_data_model=None,
    empty_produces_empty: bool = True,
) -> None:
    if not isinstance(data, dict):
        cols_used = ops.columns_used()
        table_name = [k for k in cols_used.keys()][0]
        data = {table_name: data}
    assert isinstance(data, dict)
    n_tables = len(data)
    assert n_tables > 0
    if local_data_model is None:
        local_data_model = data_algebra.default_data_model
    assert isinstance(ops, ViewRepresentation)
    if not local_data_model.is_appropriate_data_instance(expect):
        raise TypeError("expected expect to be a local_data_model.pd.DataFrame")
    cols_used = ops.columns_used()
    if len(cols_used) < 1:
        raise ValueError("no tables used")
    for k in cols_used.keys():
        v = data[k]
        assert local_data_model.is_appropriate_data_instance(v)
    if check_parse:
        if not formats_to_self(ops):
            raise ValueError("ops did not round-trip format")
    res = ops.eval(data_map=data)
    if not local_data_model.is_appropriate_data_instance(res):
        raise ValueError(
            "expected res to be local_data_model.pd.DataFrame, got: " + str(type(res))
        )
    if not equivalent_frames(
        res,
        expect,
        float_tol=float_tol,
        check_column_order=check_column_order,
        cols_case_sensitive=cols_case_sensitive,
        check_row_order=check_row_order,
    ):
        raise ValueError("Pandas eval result did not match expect")
    if len(data) == 1:
        res_t = ops.transform(list(data.values())[0])
        if not equivalent_frames(
            res_t,
            expect,
            float_tol=float_tol,
            check_column_order=check_column_order,
            cols_case_sensitive=cols_case_sensitive,
            check_row_order=check_row_order,
        ):
            raise ValueError("Pandas transform result did not match expect")
    empty_map = {k: v.iloc[range(0), :].reset_index(drop=True, inplace=False) for k, v in data.items()}
    empty_res = ops.eval(empty_map)
    assert local_data_model.is_appropriate_data_instance(empty_res)
    assert set(empty_res.columns) == set(res.columns)
    if empty_produces_empty:
        assert empty_res.shape[0] == 0
    else:
        assert empty_res.shape[0] > 0
    if n_tables == 2:
        keys = [k for k in data.keys()]
        partial_maps = [
            {
                keys[0]: data[keys[0]].reset_index(drop=True, inplace=False),
                keys[1]: data[keys[1]],
            },
            {
                keys[0]: data[keys[0]],
                keys[1]: data[keys[1]].reset_index(drop=True, inplace=False),
            },
        ]
        for pm in partial_maps:
            empty_res_i = ops.eval(pm)
            assert set(empty_res_i.columns) == set(res.columns)
    if db_handles is not None:
        for db_handle in db_handles:
            to_del = set()
            sql_statements = []
            for initial_commas in [True, False]:
                for use_with in [True, False]:
                    sql_format_options = data_algebra.db_model.SQLFormatOptions(
                        use_with=use_with,
                        annotate=True,
                        sql_indent=" ",
                        initial_commas=initial_commas,
                    )
                    sql = db_handle.to_sql(ops, sql_format_options=sql_format_options,)
                    assert isinstance(sql, str)
                    sql_statements.append(sql)
            if db_handle.conn is not None:
                for (k, v) in data.items():
                    db_handle.insert_table(v, table_name=k, allow_overwrite=True)
                    to_del.add(k)
                caught = None
                res_db_sql = []
                res_db_ops = None
                try:
                    for sql in sql_statements:
                        res_db_sql_i = db_handle.read_query(sql)
                        res_db_sql.append(res_db_sql_i)
                    res_db_ops = db_handle.read_query(ops)
                except Exception as e:
                    caught = e
                for k in to_del:
                    db_handle.drop_table(k)
                if caught is not None:
                    raise caught
                for res in res_db_sql:
                    if not equivalent_frames(
                        res,
                        expect,
                        float_tol=float_tol,
                        check_column_order=check_column_order,
                        cols_case_sensitive=cols_case_sensitive,
                        check_row_order=check_row_order,
                    ):
                        raise ValueError(f"{db_handle} SQL result did not match expect")
                if not equivalent_frames(
                    res_db_ops,
                    expect,
                    float_tol=float_tol,
                    check_column_order=check_column_order,
                    cols_case_sensitive=cols_case_sensitive,
                    check_row_order=check_row_order,
                ):
                    raise ValueError(f"{db_handle} ops result did not match expect")
def check_transform_on_handles ( * , ops , data , expect , db_handles , float_tol : float = 1e-8 , check_column_order : bool = False , cols_case_sensitive : bool = False , check_row_order : bool = False , check_parse : bool = True , local_data_model = None , empty_produces_empty : bool = True , ) -> None : if not isinstance ( data , dict ) : cols_used = ops . columns_used ( ) table_name = [ k for k in cols_used . keys ( ) ] [ 0 ] data = { table_name : data } assert isinstance ( data , dict ) n_tables = len ( data ) assert n_tables > 0 if local_data_model is None : local_data_model = data_algebra . default_data_model assert isinstance ( ops , ViewRepresentation ) if not local_data_model . is_appropriate_data_instance ( expect ) : raise TypeError ( "expected expect to be a local_data_model.pd.DataFrame" ) cols_used = ops . columns_used ( ) if len ( cols_used ) < 1 : raise ValueError ( "no tables used" ) for k in cols_used . keys ( ) : v = data [ k ] assert local_data_model . is_appropriate_data_instance ( v ) if check_parse : if not formats_to_self ( ops ) : raise ValueError ( "ops did not round-trip format" ) res = ops . eval ( data_map = data ) if not local_data_model . is_appropriate_data_instance ( res ) : raise ValueError ( "expected res to be local_data_model.pd.DataFrame, got: " + str ( type ( res ) ) ) if not equivalent_frames ( res , expect , float_tol = float_tol , check_column_order = check_column_order , cols_case_sensitive = cols_case_sensitive , check_row_order = check_row_order , ) : raise ValueError ( "Pandas eval result did not match expect" ) if len ( data ) == 1 : res_t = ops . transform ( list ( data . values ( ) ) [ 0 ] ) if not equivalent_frames ( res_t , expect , float_tol = float_tol , check_column_order = check_column_order , cols_case_sensitive = cols_case_sensitive , check_row_order = check_row_order , ) : raise ValueError ( "Pandas transform result did not match expect" ) empty_map = { k : v . iloc [ range ( 0 ) , : ] . reset_index ( drop = True , inplace = False ) for k , v in data . items ( ) } empty_res = ops . eval ( empty_map ) assert local_data_model . is_appropriate_data_instance ( empty_res ) assert set ( empty_res . columns ) == set ( res . columns ) if empty_produces_empty : assert empty_res . shape [ 0 ] == 0 else : assert empty_res . shape [ 0 ] > 0 if n_tables == 2 : keys = [ k for k in data . keys ( ) ] partial_maps = [ { keys [ 0 ] : data [ keys [ 0 ] ] . reset_index ( drop = True , inplace = False ) , keys [ 1 ] : data [ keys [ 1 ] ] , } , { keys [ 0 ] : data [ keys [ 0 ] ] , keys [ 1 ] : data [ keys [ 1 ] ] . reset_index ( drop = True , inplace = False ) , } , ] for pm in partial_maps : empty_res_i = ops . eval ( pm ) assert set ( empty_res_i . columns ) == set ( res . columns ) if db_handles is not None : for db_handle in db_handles : to_del = set ( ) sql_statements = [ ] for initial_commas in [ True , False ] : for use_with in [ True , False ] : sql_format_options = data_algebra . db_model . SQLFormatOptions ( use_with = use_with , annotate = True , sql_indent = " " , initial_commas = initial_commas , ) sql = db_handle . to_sql ( ops , sql_format_options = sql_format_options , ) assert isinstance ( sql , str ) sql_statements . append ( sql ) if db_handle . conn is not None : for ( k , v ) in data . items ( ) : db_handle . insert_table ( v , table_name = k , allow_overwrite = True ) to_del . add ( k ) caught = None res_db_sql = [ ] res_db_ops = None try : for sql in sql_statements : res_db_sql_i = db_handle . read_query ( sql ) res_db_sql . append ( res_db_sql_i ) res_db_ops = db_handle . read_query ( ops ) except Exception as e : caught = e for k in to_del : db_handle . drop_table ( k ) if caught is not None : raise caught for res in res_db_sql : if not equivalent_frames ( res , expect , float_tol = float_tol , check_column_order = check_column_order , cols_case_sensitive = cols_case_sensitive , check_row_order = check_row_order , ) : raise ValueError ( f"{db_handle} SQL result did not match expect" ) if not equivalent_frames ( res_db_ops , expect , float_tol = float_tol , check_column_order = check_column_order , cols_case_sensitive = cols_case_sensitive , check_row_order = check_row_order , ) : raise ValueError ( f"{db_handle} ops result did not match expect" )
Test an operator dag produces the expected result, and parses correctly.
Test an operator dag produces the expected result and parses correctly .
"""
    Test an operator dag produces the expected result, and parses correctly.
    Asserts if there are issues

    :param ops: data_algebra.data_ops.ViewRepresentation
    :param data: pd.DataFrame or map of strings to pd.DataFrame
    :param expect: pd.DataFrame
    :param db_handles:  list of database handles to use in testing
    :param float_tol: passed to equivalent_frames()
    :param check_column_order: passed to equivalent_frames()
    :param cols_case_sensitive: passed to equivalent_frames()
    :param check_row_order: passed to equivalent_frames()
    :param check_parse: if True check expression parses/formats to self
    :param local_data_model: optional alternate evaluation model
    :param empty_produces_empty: logical, if true assume emtpy inputs should produce empty output
    :return: None, assert if there is an issue
    """ # convert single table to dictionary # check all needed tables are present # try pandas path # try on empty inputs # try on combinations of empty and original # try any db paths # print(sql)
None, assert if there is an issue
None assert if there is an issue
ops
None
data
pd.DataFrame or map of strings to pd.DataFrame
pd . DataFrame or map of strings to pd . DataFrame
expect
None
db_handles
list of database handles to use in testing
list of database handles to use in testing
float_tol
float
passed to equivalent_frames()
passed to equivalent_frames ()
check_column_order
bool
passed to equivalent_frames()
passed to equivalent_frames ()
cols_case_sensitive
bool
passed to equivalent_frames()
passed to equivalent_frames ()
check_row_order
bool
passed to equivalent_frames()
passed to equivalent_frames ()
check_parse
bool
if True check expression parses/formats to self
if True check expression parses / formats to self
local_data_model
optional alternate evaluation model
optional alternate evaluation model
empty_produces_empty
bool
logical, if true assume emtpy inputs should produce empty output
logical if true assume emtpy inputs should produce empty output