Source code for

from typing import List, Union

from sqlalchemy import text

from import isMssqlDialect, isPostGreSQLDialect

def _createMssqlSqlText(values: List[Union[int, str]]) -> str:
    if not values:
        name = "peekCsvVarcharToTable"  # Either will do

    elif isinstance(values[0], str):
        name = "peekCsvVarcharToTable"

    elif isinstance(values[0], int):
        name = "peekCsvIntToTable"
        values = [str(v) for v in values]

        raise NotImplementedError("The value supplies isn't a str or int, %s", values[0])

    return text("SELECT * FROM [dbo].[%s]('%s')" % (name, ','.join(values)))

[docs]def makeOrmValuesSubqueryCondition(ormSession, column, values: List[Union[int, str]]): """ Make Orm Values Subquery :param ormSession: The orm session instance :param column: The column from the Declarative table, eg TableItem.colName :param values: A list of string or int values """ if isPostGreSQLDialect(ormSession.bind): return column.in_(values) if not isMssqlDialect(ormSession.bind): raise NotImplementedError() sql = _createMssqlSqlText(values) sub_qry = ormSession.query(column) # Any column, it just assigns a name sub_qry = sub_qry.from_statement(sql) return column.in_(sub_qry)
[docs]def makeCoreValuesSubqueryCondition(engine, column, values: List[Union[int, str]]): """ Make Core Values Subquery :param engine: The database engine, used to determine the dialect :param column: The column, eg TableItem.__table__.c.colName :param values: A list of string or int values """ if isPostGreSQLDialect(engine): return column.in_(values) if not isMssqlDialect(engine): raise NotImplementedError() sql = _createMssqlSqlText(values) return column.in_(sql)