SQLAlchemy asyncio threading fixture

conftest.py

@pytest.fixture(scope='session', autouse=True)
def sqlalchemy_thread_check(connection):
    """Raise an exception is SQL is executed in the event loop thread in the application code."""
    def check_thread(*args):
        if asyncio._get_running_loop():
            frames = [f for f in extract_stack()[:-1] if 'ebserver' in f.filename]
            if frames:
                raise RuntimeError(f'SQL executed in the event loop thread '
                                   f'(from {frames[-1].filename}:{frames[-1].lineno})')

    for event in 'before_execute', 'before_cursor_execute':
        listen(connection, event, check_thread)