Method test_lambda_concurrency

test/sql/test_lambdas.py:2195–2251

(self, testing_engine, mapping_fixture)

    @testing.requires.timing_intensive
    def test_lambda_concurrency(self, testing_engine, mapping_fixture):
        A = mapping_fixture
        engine = testing_engine(options={"pool_size": self.THREADS + 5})
        NUM_OF_LAMBDAS = 150

        code = """
from sqlalchemy import lambda_stmt, select


def generate_lambda_stmt(wanted):
    stmt = lambda_stmt(lambda: select(A.col1, A.col2, A.col3, A.col4))
"""

        for _ in range(NUM_OF_LAMBDAS):
            code += (
                "    stmt += lambda s: s.where((A.col1 == wanted) & "
                "(A.col2 == wanted) & (A.col3 == wanted) & "
                "(A.col4 == wanted))\n"
            )

        code += """
    return stmt
"""

        d = {"A": A, "__name__": "lambda_fake"}
        exec(code, d)
        generate_lambda_stmt = d["generate_lambda_stmt"]

        runs: List[Optional[int]] = [None for _ in range(self.THREADS)]
        conns = [engine.connect() for _ in range(self.THREADS)]

        def run(num):
            wanted = str(num)
            connection = conns[num]
            time.sleep(0.1)
            stmt = generate_lambda_stmt(wanted)
            time.sleep(0.1)
            row = connection.execute(stmt).first()
            if not row:
                runs[num] = False
            else:
                runs[num] = True

        threads = [
            threading.Thread(target=run, args=(num,))
            for num in range(self.THREADS)
        ]

        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(timeout=10)
        for conn in conns:
            conn.close()

        fails = len([r for r in runs if r is False])
        assert not fails, f"{fails} runs failed"