Class DialectSQL

lib/sqlalchemy/testing/assertsql.py:263–338

class DialectSQL(CompiledSQL):
    def _compile_dialect(self, execute_observed):
        return execute_observed.context.dialect

    def _compare_no_space(self, real_stmt, received_stmt):
        stmt = re.sub(r"[\n\t]", "", real_stmt)
        return received_stmt == stmt

    def _received_statement(self, execute_observed):
        received_stmt, received_params = super()._received_statement(
            execute_observed
        )

        # TODO: why do we need this part?
        for real_stmt in execute_observed.statements:
            if self._compare_no_space(
                real_stmt.context.statement, received_stmt
            ):
                break
        else:
            raise AssertionError(
                "Can't locate compiled statement %r in list of "
                "statements actually invoked" % received_stmt
            )

        return received_stmt, execute_observed.context.compiled_parameters

    def _dialect_adjusted_statement(self, dialect):
        paramstyle = dialect.paramstyle
        stmt = re.sub(r"[\n\t]", "", self.statement)

        # temporarily escape out PG double colons
        stmt = stmt.replace("::", "!!")

        if paramstyle == "pyformat":
            stmt = re.sub(r":([\w_]+)", r"%(\1)s", stmt)
        else:
            # positional params
            repl = None
            if paramstyle == "qmark":
                repl = "?"
            elif paramstyle == "format":
                repl = r"%s"
            elif paramstyle.startswith("numeric"):
                counter = itertools.count(1)

                num_identifier = "$" if paramstyle == "numeric_dollar" else ":"

                def repl(m):
                    return f"{num_identifier}{next(counter)}"

            stmt = re.sub(r":([\w_]+)", repl, stmt)

        # put them back
        stmt = stmt.replace("!!", "::")

        return stmt