HEX
Server: Apache/2.4.52 (Ubuntu)
System: Linux spn-python 5.15.0-89-generic #99-Ubuntu SMP Mon Oct 30 20:42:41 UTC 2023 x86_64
User: arjun (1000)
PHP: 8.1.2-1ubuntu2.20
Disabled: NONE
$ pwd: /usr/local/lib/python3.10/dist-packages/sqlalchemy/testing/__pycache__/
Upload Files
File: //usr/local/lib/python3.10/dist-packages/sqlalchemy/testing/__pycache__/asyncio.cpython-310.pyc
o

���g��@svddlmZddlmZddlZddlmZddlmZdZ	e�Z
d	d
�Zdd�Zd
d�Z
dd�Zdd�Zdd�ZdS)�)�annotations)�wrapsN�)�config�)�
_AsyncUtilTcCst��dS)zcalled when the test finishesN)�_async_util�close�r
r
�E/usr/local/lib/python3.10/dist-packages/sqlalchemy/testing/asyncio.py�	_shutdown#srcOstj|g|�Ri|��S�N)r�run��fn�args�kwargsr
r
r�_run_coroutine_function(srcOs*ts	||i|��Stj|g|�Ri|��S)aRun a function in an asyncio loop unconditionally.

    This function is used for provisioning features like
    testing a database connection for server info.

    Note that for blocking IO database drivers, this means they block the
    event loop.

    )�ENABLE_ASYNCIOr�run_in_greenletrr
r
r�
_assume_async,srcOs>ts	||i|��Stjrtj|g|�Ri|��S||i|��S)a�Run a function in an asyncio loop if any current drivers might need it.

    This function is used for provisioning features that take
    place outside of a specific database driver being selected, so if the
    current driver that happens to be used for the provisioning operation
    is an async driver, it will run in asyncio and not fail.

    Note that for blocking IO database drivers, this means they block the
    event loop.

    )rr�	any_asyncrrrr
r
r�_maybe_async_provisioning=s
rcOsDts	||i|��Stjj}|rtj|g|�Ri|��S||i|��S)z�Run a function in an asyncio loop if the current selected driver is
    async.

    This function is used for test setup/teardown and tests themselves
    where the current DB driver is known.


    )rr�_current�is_asyncrr)rrrrr
r
r�_maybe_asyncRs	rcsPt���rt���fdd��t�����fdd��}|St���fdd��}|S)z�Apply the _maybe_async function to an existing function and return
    as a wrapped callable, supporting generator functions as well.

    This is currently used for pytest fixtures that support generator use.

    cs"zt|�WSty�YSwr
)�next�
StopIteration)�gen)�_stopr
r�	call_nextqs

�z'_maybe_async_wrapper.<locals>.call_nextc?s0��|i|��}	t�|�}|�urdS|Vq	r
�r)rrr�value�rr rr
r�wrap_fixturexs�
�z*_maybe_async_wrapper.<locals>.wrap_fixturecst�g|�Ri|��Sr
r!)rr)rr
rr$�s)�inspect�isgeneratorfunction�objectr)rr$r
r#r�_maybe_async_wrapperfs
�r()�
__future__r�	functoolsrr%�r�util.concurrencyrrrrrrrrr(r
r
r
r�<module>s
@ Telegram