File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/asgiref.tar
Back
timeout.py 0000644 00000007053 15030103312 0006575 0 ustar 00 # This code is originally sourced from the aio-libs project "async_timeout", # under the Apache 2.0 license. You may see the original project at # https://github.com/aio-libs/async-timeout # It is vendored here to reduce chain-dependencies on this library, and # modified slightly to remove some features we don't use. import asyncio import warnings from types import TracebackType from typing import Any # noqa from typing import Optional, Type class timeout: """timeout context manager. Useful in cases when you want to apply timeout logic around block of code or in cases when asyncio.wait_for is not suitable. For example: >>> with timeout(0.001): ... async with aiohttp.get('https://github.com') as r: ... await r.text() timeout - value in seconds or None to disable timeout logic loop - asyncio compatible event loop """ def __init__( self, timeout: Optional[float], *, loop: Optional[asyncio.AbstractEventLoop] = None, ) -> None: self._timeout = timeout if loop is None: loop = asyncio.get_running_loop() else: warnings.warn( """The loop argument to timeout() is deprecated.""", DeprecationWarning ) self._loop = loop self._task = None # type: Optional[asyncio.Task[Any]] self._cancelled = False self._cancel_handler = None # type: Optional[asyncio.Handle] self._cancel_at = None # type: Optional[float] def __enter__(self) -> "timeout": return self._do_enter() def __exit__( self, exc_type: Type[BaseException], exc_val: BaseException, exc_tb: TracebackType, ) -> Optional[bool]: self._do_exit(exc_type) return None async def __aenter__(self) -> "timeout": return self._do_enter() async def __aexit__( self, exc_type: Type[BaseException], exc_val: BaseException, exc_tb: TracebackType, ) -> None: self._do_exit(exc_type) @property def expired(self) -> bool: return self._cancelled @property def remaining(self) -> Optional[float]: if self._cancel_at is not None: return max(self._cancel_at - self._loop.time(), 0.0) else: return None def _do_enter(self) -> "timeout": # Support Tornado 5- without timeout # Details: https://github.com/python/asyncio/issues/392 if self._timeout is None: return self self._task = asyncio.current_task(self._loop) if self._task is None: raise RuntimeError( "Timeout context manager should be used " "inside a task" ) if self._timeout <= 0: self._loop.call_soon(self._cancel_task) return self self._cancel_at = self._loop.time() + self._timeout self._cancel_handler = self._loop.call_at(self._cancel_at, self._cancel_task) return self def _do_exit(self, exc_type: Type[BaseException]) -> None: if exc_type is asyncio.CancelledError and self._cancelled: self._cancel_handler = None self._task = None raise asyncio.TimeoutError if self._timeout is not None and self._cancel_handler is not None: self._cancel_handler.cancel() self._cancel_handler = None self._task = None return None def _cancel_task(self) -> None: if self._task is not None: self._task.cancel() self._cancelled = True __pycache__/testing.cpython-310.pyc 0000644 00000006140 15030103312 0013117 0 ustar 00 o �h� � @ sB d dl Z d dlZd dlZddlmZ ddlmZ G dd� d�ZdS )� N� )�guarantee_single_callable)�timeoutc @ sP e Zd ZdZdd� Zddd�Zddd �Zd d � Zd d � Zddd�Z ddd�Z dS )�ApplicationCommunicatorz} Runs an ASGI application in a test mode, allowing sending of messages to it and retrieval of messages it sends. c C sN t |�| _|| _t�� | _t�� | _t�� � tj | �|| jj | jj ��| _ d S �N)r � application�scope�asyncio�Queue� input_queue� output_queue� contextvars�Context�run� create_task�get�put�future)�selfr r � r �B/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/testing.py�__init__ s �z ApplicationCommunicator.__init__r c � s� �zXt |�4 I dH �# z | jI dH | j�� W n tjy" Y nw W d �I dH n 1 I dH s3w Y W | j�� sX| j�� z | jI dH W dS tjyW Y dS w dS | j�� sw| j�� z| jI dH W w tjyv Y w w w )zV Waits for the application to stop itself and returns any exceptions. N)� async_timeoutr �resultr �CancelledError�done�cancel)r r r r r �wait s6 � ��(�� �� ��z ApplicationCommunicator.waitTc C s. | j �� s | j �� d S |r| j �� d S d S r )r r r r )r � exceptionsr r r �stop0 s �z ApplicationCommunicator.stopc C s( z | j dd� W d S ty Y d S w )NF)r )r � RuntimeError)r r r r �__del__7 s �z ApplicationCommunicator.__del__c � s �| j �|�I dH dS )z; Sends a single message to the application N)r r )r �messager r r � send_input? s �z"ApplicationCommunicator.send_inputc � s� �| j �� r | j �� z*t|�4 I dH � | j�� I dH W d �I dH W S 1 I dH s.w Y W dS tjyh } z&| j �� rJ| j �� |�| j �� z | j I dH W |� tj yc Y |�w d}~ww )zX Receives a single message from the application, with optional timeout. N) r r r r r r r � TimeoutErrorr r )r r �er r r �receive_outputF s* � 4� �����z&ApplicationCommunicator.receive_output皙�����?�{�G�z�?c � sR �t �� }t �� | |k r$| j�� sdS t�|�I dH t �� | |k s | j�� S )zO Checks that there is no message to receive in the given time. FN)�time� monotonicr �emptyr �sleep)r r �interval�startr r r �receive_nothing] s � � z'ApplicationCommunicator.receive_nothingN)r )T)r' r( ) �__name__� __module__� __qualname__�__doc__r r r r! r# r& r/ r r r r r s r )r r r) � compatibilityr r r r r r r r �<module> s __pycache__/server.cpython-310.pyc 0000644 00000013153 15030103312 0012752 0 ustar 00 o �hu � @ sH d dl Z d dlZd dlZd dlZddlmZ e�e�ZG dd� d�Z dS )� N� )�guarantee_single_callablec @ s^ e Zd ZdZdZddd�Zdd� Zdd � Zd d � Zd d � Z dd� Z dd� Z dd� Z dd� Z dS )�StatelessServerax Base server class that handles basic concepts like application instance creation/pooling, exception handling, and similar, for stateless protocols (i.e. ones without actual incoming connections to the process) Your code should override the handle() method, doing whatever it needs to, and calling get_or_create_application_instance with a unique `scope_id` and `scope` for the scope it wants to get. If an application instance is found with the same `scope_id`, you are given its input queue, otherwise one is made for you with the scope provided and you are given that fresh new input queue. Either way, you should do something like: input_queue = self.get_or_create_application_instance( "user-123456", {"type": "testprotocol", "user_id": "123456", "username": "andrew"}, ) input_queue.put_nowait(message) If you try and create an application instance and there are already `max_application` instances, the oldest/least recently used one will be reclaimed and shut down to make space. Application coroutines that error will be found periodically (every 100ms by default) and have their exceptions printed to the console. Override application_exception() if you want to do more when this happens. If you override run(), make sure you handle things like launching the application checker. g�������?�� c C s || _ || _i | _d S �N)� application�max_applications�application_instances)�selfr r � r �A/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/server.py�__init__. s zStatelessServer.__init__c C sJ t �� }t �| �� � z |�| �� � W dS ty$ t�d� Y dS w )zD Runs the asyncio event loop with our handler loop. z Exiting due to Ctrl-C/interruptN) �asyncio�get_event_loop� ensure_future�application_checker�run_until_complete�handle�KeyboardInterrupt�logger�info)r � event_loopr r r �run7 s �zStatelessServer.runc � � �t d��)NzYou must implement handle()��NotImplementedError)r r r r r B s �zStatelessServer.handlec � r )zM Receives outbound sends from applications and handles them. z%You must implement application_send()r )r �scope�messager r r �application_sendE s �z StatelessServer.application_sendc s� |�j v rt�� �j | d< �j | d S t�j ��jkr)��� t�j ��jks t�� }t�j�}t� |� |j � �fdd�d��}||� t�� d��j |< |S )zH Creates an application instance and returns its queue. � last_used� input_queuec s �� � | �S r )r )r �r r r r �<lambda>_ s zDStatelessServer.get_or_create_application_instance.<locals>.<lambda>)r �receive�send)r �futurer r ) r �time�lenr �"delete_oldest_application_instancer �Queuer r r �get)r �scope_idr r �application_instancer% r r! r �"get_or_create_application_instanceM s* � �� �z2StatelessServer.get_or_create_application_instancec C sL t dd� | j�� D ��}| j�� D ]\}}|d |kr#| �|� dS qdS )zC Finds and deletes the oldest application instance c s s � | ]}|d V qdS )r Nr )�.0�detailsr r r � <genexpr>n s � �zEStatelessServer.delete_oldest_application_instance.<locals>.<genexpr>r N)�minr �values�items�delete_application_instance)r � oldest_timer+ r/ r r r r( j s � ��z2StatelessServer.delete_oldest_application_instancec C s2 | j | }| j |= |d �� s|d �� dS dS )z� Removes an application instance (makes sure its task is stopped, then removes it from the current set) r% N)r �done�cancel)r r+ r/ r r r r4 x s �z+StatelessServer.delete_application_instancec � s~ � t �| j�I dH t| j�� �D ]+\}}|d �� r=|d �� }|r-| �||�I dH z| j|= W q t y< Y qw qq)z� Goes through the set of current application instance Futures and cleans up any that are done/prints exceptions for any that errored. TNr% ) r �sleep� application_checker_interval�listr r3 r6 � exception�application_exception�KeyError)r r+ r/ r; r r r r � s � ���z#StatelessServer.application_checkerc � s* �t �d|d�t�|j��d|� �� dS )zL Called whenever an application coroutine has an exception. z%Exception inside application: %s %s%s� z N)�logging�error�join� traceback� format_tb� __traceback__)r r; �application_detailsr r r r< � s ��z%StatelessServer.application_exceptionN)r )�__name__� __module__� __qualname__�__doc__r9 r r r r r- r( r4 r r< r r r r r s r ) r r? r& rB � compatibilityr � getLoggerrF r r r r r r �<module> s __pycache__/__init__.cpython-310.pyc 0000644 00000000265 15030103312 0013203 0 ustar 00 o �h � @ s d Z dS )z3.8.1N)� __version__� r r �C/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/__init__.py�<module> s __pycache__/typing.cpython-310.pyc 0000644 00000016626 15030103312 0012766 0 ustar 00 o �hx � @ s� d dl Z d dlmZmZmZmZmZmZmZm Z m Z m Z m Z m Z e jdkr,d dlmZ nd dlmZ dZG dd� de �ZG dd � d e �ZG d d � d e �ZG d d � d e �Ze eef Ze eeef ZG dd� de �ZG dd� de �ZG dd� de �ZG dd� de �ZG dd� de �Z G dd� de �Z G dd� de �Z G d d � d e �Z G d d � d e �Z G d d!� d!e �Z!G d"d#� d#e �Z"G d$d%� d%e �Z#G d&d'� d'e �Z$G d(d)� d)e �Z%G d*d+� d+e �Z&G d,d-� d-e �Z'G d.d/� d/e �Z(G d0d1� d1e �Z)G d2d3� d3e �Z*G d4d5� d5e �Z+G d6d7� d7e �Z,G d8d9� d9e �Z-e ee e e"e&e(e)f Z.e eee e e e!e#e$e%e'e*e+e,e-f Z/eg ee. f Z0ee/ged f Z1G d:d;� d;e �Z2e e2 Z3eee0e1ged f Z4e e3e4f Z5dS )<� N) �Any� Awaitable�Callable�Dict�Iterable�Literal�Optional�Protocol�Tuple�Type� TypedDict�Union)� � )� NotRequired)#� ASGIVersions� HTTPScope�WebSocketScope� LifespanScope�WWWScope�Scope�HTTPRequestEvent�HTTPResponseStartEvent�HTTPResponseBodyEvent�HTTPResponseTrailersEvent�HTTPResponsePathsendEvent�HTTPServerPushEvent�HTTPDisconnectEvent�WebSocketConnectEvent�WebSocketAcceptEvent�WebSocketReceiveEvent�WebSocketSendEvent�WebSocketResponseStartEvent�WebSocketResponseBodyEvent�WebSocketDisconnectEvent�WebSocketCloseEvent�LifespanStartupEvent�LifespanShutdownEvent� LifespanStartupCompleteEvent�LifespanStartupFailedEvent� LifespanShutdownCompleteEvent�LifespanShutdownFailedEvent�ASGIReceiveEvent� ASGISendEvent�ASGIReceiveCallable�ASGISendCallable� ASGI2Protocol�ASGI2Application�ASGI3Application�ASGIApplicationc @ s. e Zd ZU eed< eed ed f ed< dS )r � spec_versionz2.0z3.0�versionN)�__name__� __module__� __qualname__�str�__annotations__r r � r; r; �A/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/typing.pyr = s r c @ s� e Zd ZU ed ed< eed< eed< eed< eed< eed< eed< eed < eed < ee eef ed < e e ee f ed < e e ee e f ed < e e eef ed< e e ee eef f ed< dS )r �http�type�asgi� http_version�method�scheme�path�raw_path� query_string� root_path�headers�client�server�state� extensionsN�r6 r7 r8 r r: r r9 �bytesr r r �intr r r �objectr; r; r; r< r B s r c @ s� e Zd ZU ed ed< eed< eed< eed< eed< eed< eed< eed < ee eef ed < e e ee f ed < e e ee e f ed < ee ed < e e eef ed< e e ee eef f ed< dS )r � websocketr> r? r@ rB rC rD rE rF rG rH rI � subprotocolsrJ rK NrL r; r; r; r< r S s r c @ s6 e Zd ZU ed ed< eed< eeee f ed< dS )r �lifespanr> r? rJ N) r6 r7 r8 r r: r r r r9 r r; r; r; r< r d � r c @ �* e Zd ZU ed ed< eed< eed< dS )r z http.requestr> �body� more_bodyN�r6 r7 r8 r r: rM �boolr; r; r; r< r n � r c @ s* e Zd ZU ed ed< eeef ed< dS )�HTTPResponseDebugEventzhttp.response.debugr> �infoN)r6 r7 r8 r r: r r9 rO r; r; r; r< rZ t s rZ c @ s> e Zd ZU ed ed< eed< eeeef ed< e ed< dS )r zhttp.response.startr> �statusrG �trailersN) r6 r7 r8 r r: rN r r rM rX r; r; r; r< r y s r c @ rT )r zhttp.response.bodyr> rU rV NrW r; r; r; r< r � rY r c @ s6 e Zd ZU ed ed< eeeef ed< eed< dS )r zhttp.response.trailersr> rG � more_trailersN) r6 r7 r8 r r: r r rM rX r; r; r; r< r � s r c @ �" e Zd ZU ed ed< eed< dS )r zhttp.response.pathsendr> rC N�r6 r7 r8 r r: r9 r; r; r; r< r � � r c @ �6 e Zd ZU ed ed< eed< eeeef ed< dS )r zhttp.response.pushr> rC rG N) r6 r7 r8 r r: r9 r r rM r; r; r; r< r � rS r c @ � e Zd ZU ed ed< dS )r zhttp.disconnectr> N�r6 r7 r8 r r: r; r; r; r< r � � r c @ rc )r zwebsocket.connectr> Nrd r; r; r; r< r � re r c @ s: e Zd ZU ed ed< ee ed< eee e f ed< dS )r zwebsocket.acceptr> � subprotocolrG N) r6 r7 r8 r r: r r9 r r rM r; r; r; r< r � s r c @ �2 e Zd ZU ed ed< ee ed< ee ed< dS )r zwebsocket.receiver> rM �textN�r6 r7 r8 r r: r rM r9 r; r; r; r< r � � r c @ rg )r! zwebsocket.sendr> rM rh Nri r; r; r; r< r! � rj r! c @ rb )r" z websocket.http.response.startr> r\ rG N) r6 r7 r8 r r: rN r r rM r; r; r; r< r" � rS r" c @ rT )r# z websocket.http.response.bodyr> rU rV NrW r; r; r; r< r# � rY r# c @ r_ )r$ zwebsocket.disconnectr> �codeN)r6 r7 r8 r r: rN r; r; r; r< r$ � ra r$ c @ s. e Zd ZU ed ed< eed< ee ed< dS )r% zwebsocket.closer> rk �reasonN)r6 r7 r8 r r: rN r r9 r; r; r; r< r% � s r% c @ rc )r& zlifespan.startupr> Nrd r; r; r; r< r& � re r& c @ rc )r' zlifespan.shutdownr> Nrd r; r; r; r< r' � re r' c @ rc )r( zlifespan.startup.completer> Nrd r; r; r; r< r( � re r( c @ r_ )r) zlifespan.startup.failedr> �messageNr` r; r; r; r< r) � ra r) c @ rc )r* zlifespan.shutdown.completer> Nrd r; r; r; r< r* � re r* c @ r_ )r+ zlifespan.shutdown.failedr> rm Nr` r; r; r; r< r+ � ra r+ c @ s4 e Zd Zdeddfdd�Zdededdfdd �ZdS ) r0 �scope�returnNc C s d S �Nr; )�selfrn r; r; r< �__init__ s zASGI2Protocol.__init__�receive�sendc � s �d S rp r; )rq rs rt r; r; r< �__call__ s �zASGI2Protocol.__call__)r6 r7 r8 r rr r. r/ ru r; r; r; r< r0 s ���r0 )6�sys�typingr r r r r r r r r r r r � version_infor �typing_extensions�__all__r r r r r r r rZ r r r r r r r r r r! r" r# r$ r% r& r' r( r) r* r+ r, r- r. r/ r0 r1 r2 r3 r; r; r; r< �<module> s� 8 ' �� �� ��� __pycache__/sync.cpython-310.pyc 0000644 00000032627 15030103312 0012427 0 ustar 00 o �h�S � @ s� U d dl Z d dlZ d dlZd dlZd dlZd dlZd dlZd dlZd dlZd dl Z d dl m Z m Z d dl mZmZmZmZmZmZmZmZmZmZmZmZ ddlmZ ddl m Z ej dkred dl m Z nd dl m Z ersd d l!m"Z" ed ed ef d �Z#e d �Z$ed�Z%dej&ddfdd�Z'e(ed�r�ej)Z)ej*Z*ee#ge#f e+d< n e j)Z)de#de#fdd�Z*G dd� d�Z,G dd� dee$e%f �Z-G dd� dee$e%f �Z.ed d �d e/deeee$eeee%f f ee$ee% f f gee$e%f f fd d ��Z0ed d �d!eee$eeee%f f ee$ee% f f d e/dee$e%f fd"d ��Z0 d-d d �d!eeee$eeee%f f ee$ee% f f d e/deeeee$eeee%f f ee$ee% f f gee$e%f f ee$e%f f fd#d �Z0ed$dd%�d&e/d'ed( deee$e%f gee$eeee%f f f fd)d*��Z1ed$dd%�dee$e%f d&e/d'ed( dee$eeee%f f fd+d*��Z1 d-d$dd%�deee$e%f d&e/d'ed( deeee$e%f gee$eeee%f f f ee$eeee%f f f fd,d*�Z1dS ).� N)�Future�ThreadPoolExecutor) � TYPE_CHECKING�Any� Awaitable�Callable� Coroutine�Dict�Generic�List�Optional�TypeVar�Union�overload� )�CurrentThreadExecutor)�Local)� � )� ParamSpec)� OptExcInfo�_F.)�bound�_P�_R�context�returnc C sN | D ]"}| � |�}z |� � |kr|�|� W q ty$ |�|� Y qw d S �N)�get�set� LookupError)r �cvar�cvalue� r# �?/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/sync.py�_restore_context, s � ��r% �markcoroutinefunction�funcc C s t jj| _| S r )�asyncio� coroutines� _is_coroutine)r' r# r# r$ r& C s c @ s( e Zd ZdZdd� Zdd� Zdd� ZdS ) �ThreadSensitiveContextaY Async context manager to manage context for thread sensitive mode This context manager controls which thread pool executor is used when in thread sensitive mode. By default, a single thread pool executor is shared within a process. The ThreadSensitiveContext() context manager may be used to specify a thread pool per context. This context manager is re-entrant, so only the outer-most call to ThreadSensitiveContext will set the context. Usage: >>> import time >>> async with ThreadSensitiveContext(): ... await sync_to_async(time.sleep, 1)() c C s d | _ d S r )�token��selfr# r# r$ �__init__\ s z ThreadSensitiveContext.__init__c � s6 �zt j�� W | S ty t j�| �| _Y | S w r )� SyncToAsync�thread_sensitive_contextr r r r, r- r# r# r$ � __aenter___ s � ��z!ThreadSensitiveContext.__aenter__c � s8 �| j sd S tj�| d �}|r|�� tj�| j � d S r )r, r0 �context_to_thread_executor�pop�shutdownr1 �reset)r. �exc�value�tb�executorr# r# r$ � __aexit__g s �z ThreadSensitiveContext.__aexit__N)�__name__� __module__� __qualname__�__doc__r/ r2 r; r# r# r# r$ r+ H s r+ c @ s� e Zd ZU dZe� Zded< i Zded< d dee e e e e e f f e e ee f f defd d �Zd e jd e jd e fdd�Zdd� Zde de d e e e f fdd�Zddddddd eej d e jd e jd d fd d �Zd S )!� AsyncToSynca� Utility class which turns an awaitable that only works on the thread with the event loop into a synchronous callable that works in a subthread. If the call stack contains an async loop, the code runs there. Otherwise, the code runs in a new loop in a new thread. Either way, this thread then pauses and waits to run any thread_sensitive code called from further down the call stack using SyncToAsync, before finally exiting once the async task returns. r � executorsz6Dict[asyncio.AbstractEventLoop, CurrentThreadExecutor]�loop_thread_executorsF� awaitable�force_new_loopc C s� t |�rt|�stt|d|��stjddd� || _z| jj| _W n ty* Y nw || _d | _ zt � � | _ W d S t yC Y d S w )N�__call__z4async_to_sync was passed a non-async-marked callable� )� stacklevel) �callable�iscoroutinefunction�getattr�warnings�warnrC �__self__�AttributeErrorrD �main_event_loopr( �get_running_loop� RuntimeError)r. rC rD r# r# r$ r/ � s* ��� � �zAsyncToSync.__init__�args�kwargsr c O s� d}| j s | js ttjdd �}|r |t�� kr ttjdd �| _zt�� }W n t y. Y n w |� � r7t d��t � }t| j dd �}t � }|| j _t�� g} ttjdd �} d } zb| j|t�� | | g|�R i |��} | jrp| j� � s�t�� } || j| < tdd�} | �| j| | �}|r�|�|� |�� n| j�| jj | � |r�|�|� W | d ur�| j| = t | d � || j _|�� S | d ur�| j| = t | d � || j _w ) NT�main_event_loop_pidrO znYou cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.�current� task_contextr �� max_workersr ) rD rO rJ r0 � threadlocal�os�getpidr( rP rQ � is_runningr rA r rU � contextvars� copy_context� main_wrap�sys�exc_info�new_event_looprB r �submit�_run_event_loop�run_until_future�result�call_soon_threadsafe� create_taskr% )r. rR rS �__traceback_hide__rT � event_loop� call_result� old_executor�current_executorr rV �looprC � loop_executor� loop_futurer# r# r$ rE � sz �� �� ��� � � � � zAsyncToSync.__call__c s� t �|� zm|�|� W z[t �|�� � D ]}|�� q� fdd�}|�|� � � D ]}|�� r0q)|�� durA|�d|�� |d�� q)t|d�r[|�|� � � W |� � t �| j � dS W |� � t �| j � dS |� � t �| j � w zYt �|�� � D ]}|�� q{� fdd�}|�|� � � D ]}|�� r�q�|�� dur�|�d|�� |d�� q�t|d�r�|�|� � � W |� � t �| j � w W |� � t �| j � w |� � t �| j � w )zP Runs the given event loop (designed to be called in a thread). c � s �t j� ddi�I d H d S )N�return_exceptionsT)r( �gatherr# ��tasksr# r$ rr s �z+AsyncToSync._run_event_loop.<locals>.gatherNz(unhandled exception during loop shutdown)�message� exception�task�shutdown_asyncgens) r( �set_event_loop�run_until_complete� all_tasks�cancel� cancelledrv �call_exception_handler�hasattrrx �closerO )r. rn �cororw rr r# rs r$ rd sn ��� ��� ��� ��zAsyncToSync._run_event_loop�parent�objtypec C � t �| j|�}t �|| j�S �z* Include self for methods )� functools�partialrE �update_wrapperrC �r. r� r� r' r# r# r$ �__get__$ s zAsyncToSync.__get__rk z Future[_R]ra r rV z!Optional[List[asyncio.Task[Any]]]r Nc � s �d}|dur t |d � t�� }|dur |dur |�|� z]z*|d r=z|d � ty< | j|i |��I dH } Y n w | j|i |��I dH } W n ty` } z |�| � W Y d} ~ n d} ~ ww |�| � W |durt|durt|�|� t � � |d< dS |dur�|dur�|�|� t � � |d< w )zs Wraps the awaitable with something that puts the result into the result/exception future. TNr r ) r% r( � current_task�append� BaseExceptionrC � set_exception� set_result�remover] r^ ) r. rk ra rV r rR rS ri r� rf �er# r# r$ r_ + s8 � ���� � � zAsyncToSync.main_wrap)F)r< r= r> r? r rA �__annotations__rB r r r r r r r �boolr/ rR rS rE rd r� r r] �Contextr_ r# r# r# r$ r@ q s@ ���� � \ $�������r@ c @ s� e Zd ZU dZe�� Zedd�Ze � d�Z de d< e � d�Z de d< e�� Zde d < d d eeef d eded dd fdd�Zdejdejdefdd�Zdededeeeeeef f fdd�Z dd � Z d S ) r0 a Utility class which turns a synchronous callable into an awaitable that runs in a threadpool. It also sets a threadlocal inside the thread so calls to AsyncToSync can escape it. If thread_sensitive is passed, the code will run in the same thread as any outer code. This is needed for underlying Python code that is not threadsafe (for example, code which handles SQLite database connections). If the outermost program is async (i.e. SyncToAsync is outermost), then this will be a dedicated single sub-thread that all sync code runs in, one after the other. If the outermost program is sync (i.e. AsyncToSync is outermost), this will just be the main thread. This is achieved by idling with a CurrentThreadExecutor while AsyncToSync is blocking its sync parent, rather than just blocking. If executor is passed in, that will be used instead of the loop's default executor. In order to pass in an executor, thread_sensitive must be set to False, otherwise a TypeError will be raised. r rW r1 z.contextvars.ContextVar[ThreadSensitiveContext]�deadlock_contextz contextvars.ContextVar[bool]zEweakref.WeakKeyDictionary[ThreadSensitiveContext, ThreadPoolExecutor]r3 TNr' �thread_sensitiver: r r c C s� t |�rt|�stt|d|��rtd��|| _t�| |� || _t| � |r.|d ur.td��|| _ z|j | _ W d S t yB Y d S w )NrE z4sync_to_async can only be applied to sync functions.z6executor must not be set when thread_sensitive is True) rH rI rJ � TypeErrorr' r� r� �_thread_sensitiver& � _executorrM rN )r. r' r� r: r# r# r$ r/ � s& ��� �zSyncToAsync.__init__rR rS c � s� �d}t �� }| jrVttjdd �}|r|}nC| j�d �r7| j�� }|| jv r,| j| }n-t dd�}|| j|< n"|tj v rBtj | }n| j �d�rLt d��| j }| j �d� n| j}t�� }tj| jg|�R i |��} |j} g } |�|t�| j|t�� | | | ��} ziz t �| �I d H } W nD t jy� d}z | d }|� � z |I d H d}W n t jy� Y nw W n t y� Y nw | � � r |r�| � � | I d H } Y nw W t |� | j �d� | S W t |� | j �d� | S t |� | j �d� w )NTrU r rW Fz9Single thread executor already being used, would deadlockr ) r( rP r� rJ r@ rA r1 r r3 r rB r� rQ �single_thread_executorr r� r] r^ r� r� r' �run�run_in_executor�thread_handlerr` ra �shield�CancelledErrorr| � IndexError�doner% )r. rR rS ri rn �current_thread_executorr: r1 r �childr' rV � exec_coro�ret� cancel_parentrw r# r# r$ rE � s� � ��� �� ��� � �zSyncToAsync.__call__r� r� c C r� r� )r� r� rE r� r' r� r# r# r$ r� � s zSyncToAsync.__get__c O s` d}|| j _t�� | j _|| j _|d r)z|d � ty( ||i |�� Y S w ||i |��S )zE Wraps the sync application with exception handling. Tr )rY rO rZ r[ rT rV r� )r. rn ra rV r' rR rS ri r# r# r$ r� � s �zSyncToAsync.thread_handler)TN) r< r= r> r? � threading�localrY r r� r] � ContextVarr1 r� r� �weakref�WeakKeyDictionaryr3 r r r r� r r/ rR rS rE r r r� r� r# r# r# r$ r0 V s> � � �� ��� �P�� � r0 F�rD rD c C � d S r r# r� r# r# r$ � async_to_sync s r� rC c C r� r r# �rC rD r# r# r$ r� s c s | d u r � fdd�S t | � d�S )Nc s t | � d�S )Nr� �r@ ��fr� r# r$ �<lambda>5 s �z async_to_sync.<locals>.<lambda>r� r� r� r# r� r$ r� $ s �T�r� r: r� r: r c C r� r r# r� r# r# r$ � sync_to_async? s r� c C r� r r# �r' r� r: r# r# r$ r� H s c s$ | d u r � �fdd�S t | �� d�S )Nc s t | �� d�S )Nr� �r0 r� �r: r� r# r$ r� \ s �z sync_to_async.<locals>.<lambda>r� r� r� r# r� r$ r� R s �r )2r( �asyncio.coroutinesr] r� �inspectrZ r` r� rK r� �concurrent.futuresr r �typingr r r r r r r r r r r r r� r r� r � version_infor �typing_extensions� _typeshedr r r r r� r% r rI r& r� r+ r@ r0 r� r� r� r# r# r# r$ �<module> s� 8 ) f 8��* �� � ����� �������� * �� �� ����& �� ��� � �����&��� __pycache__/local.cpython-310.pyc 0000644 00000010153 15030103312 0012533 0 ustar 00 o �h� � @ sT d dl Z d dlZd dlZd dlZd dlmZmZmZ G dd� d�ZG dd� d�Z dS )� N)�Any�Dict�Unionc sR e Zd ZdZddd�Zdd� Zded eddf� fd d � Zdeddfd d �Z � Z S )�_CVarzStorage utility for Local.�returnNc C s t �d�| _d S )Nz asgiref.local)� contextvars� ContextVar�_data��self� r �@/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/local.py�__init__ s �z_CVar.__init__c C s8 | j �i �}z|| W S ty t| �d|����w �Nz object has no attribute )r �get�KeyError�AttributeError�r �key�storage_objectr r r � __getattr__ s �z_CVar.__getattr__r �valuec s: |dkr t � �||�S | j�i �}|||< | j�|� d S )Nr )�super� __setattr__r r �set)r r r r �� __class__r r r s z_CVar.__setattr__c C s< | j �i �}||v r||= | j �|� d S t| �d|����r )r r r r r r r r � __delattr__ s z_CVar.__delattr__)r N) �__name__� __module__� __qualname__�__doc__r r �strr r r � __classcell__r r r r r s r c sR e Zd ZdZddeddfdd�Zejdd � �Zd d � Z � fd d �Z dd� Z � Z S )�Locala Local storage for async tasks. This is a namespace object (similar to `threading.local`) where data is also local to the current async task (if there is one). In async threads, local means in the same sense as the `contextvars` module - i.e. a value set in an async frame will be visible: - to other async code `await`-ed from this frame. - to tasks spawned using `asyncio` utilities (`create_task`, `wait_for`, `gather` and probably others). - to code scheduled in a sync thread using `sync_to_async` In "sync" threads (a thread with no async event loop running), the data is thread-local, but additionally shared with async code executed via the `async_to_sync` utility, which schedules async code in a new thread and copies context across to that thread. If `thread_critical` is True, then the local will only be visible per-thread, behaving exactly like `threading.local` if the thread is sync, and as `contextvars` if the thread is async. This allows genuinely thread-sensitive code (such as DB handles) to be kept stricly to their initial thread and disable the sharing across `sync_to_async` and `async_to_sync` wrapped calls. Unlike plain `contextvars` objects, this utility is threadsafe. F�thread_criticalr Nc C s2 || _ t�� | _| |rt�� | _d S t� | _d S �N)�_thread_critical� threading�RLock� _thread_lock�local�_storager )r r% r r r r D s zLocal.__init__c c s� � | j r+zt�� W n ty | jV Y d S w t| jd�s$t� | j_| jjV d S | j� | jV W d � d S 1 s>w Y d S )N�cvar) r' �asyncio�get_running_loop� RuntimeErrorr, �hasattrr r- r* r r r r � _lock_storageQ s � � "�zLocal._lock_storagec C s6 | � � � }t||�W d � S 1 sw Y d S r& )r2 �getattr�r r �storager r r r t s $�zLocal.__getattr__c sP |dv r t � �||�S | �� �}t|||� W d � d S 1 s!w Y d S )N)�_localr, r' r* )r r r2 �setattr)r r r r5 r r r r x s "�zLocal.__setattr__c C s8 | � � �}t||� W d � d S 1 sw Y d S r& )r2 �delattrr4 r r r r ~ s "�zLocal.__delattr__)F) r r r r! �boolr � contextlib�contextmanagerr2 r r r r# r r r r r$ ( s " r$ ) r. r: r r( �typingr r r r r$ r r r r �<module> s __pycache__/current_thread_executor.cpython-310.pyc 0000644 00000006652 15030103312 0016401 0 ustar 00 o �h� � @ s� d dl Z d dlZd dlZd dlmZmZ d dlmZmZm Z m Z m Z ej dkr.d dlm Z nd dlm Z e d�Ze d�Ze d�ZG d d � d �ZG d d � d e�ZdS ) � N)�Executor�Future)� TYPE_CHECKING�Any�Callable�TypeVar�Union)� � )� ParamSpec�_T�_P�_Rc @ s@ e Zd ZdZdddeeef dejdejfdd�Z d d d �Z d S )� _WorkItemz� Represents an item needing to be run in the executor. Copied from ThreadPoolExecutor (but it's private, so we're not going to rely on importing it) �future� Future[_R]�fn�args�kwargsc O s || _ || _|| _|| _d S �N)r r r r )�selfr r r r � r �R/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/current_thread_executor.py�__init__ s z_WorkItem.__init__�returnNc C sr d}| j �� s d S z | j| ji | j��}W n ty0 } z| j �|� d } W Y d }~d S d }~ww | j �|� d S )NT)r � set_running_or_notify_cancelr r r � BaseException� set_exception� set_result)r �__traceback_hide__�result�excr r r �run# s ��z _WorkItem.run�r N) �__name__� __module__� __qualname__�__doc__r r r r r r r" r r r r r s � �� � r c @ sZ e Zd ZdZddd�Zddd �Zd eeef d ej d ej dd fdd�Z e s+dd� Z dS dS )�CurrentThreadExecutorz� An Executor that actually runs code in the thread it is instantiated in. Passed to other threads running async code, so they can run sync code in the thread they came from. r Nc C s t �� | _t�� | _d| _d S )NF)� threading�current_thread� _work_thread�queue�Queue� _work_queue�_broken)r r r r r 8 s z CurrentThreadExecutor.__init__r � Future[Any]c C sh t �� | jkr td��|�| jj� z | j�� }||u r#W d| _dS t |t �s*J �|� � ~qd| _w )z� Runs the code in the work queue until a result is available from the future. Should be run from the thread the executor is initialised in. z<You cannot run CurrentThreadExecutor from a different threadTN) r) r* r+ � RuntimeError�add_done_callbackr. �put�getr/ � isinstancer r" )r r � work_itemr r r �run_until_future= s � �� z&CurrentThreadExecutor.run_until_futurer r r r c O sR t �� | jkr td��| jrtd��t� }t||g|�R i |��}| j�|� |S )Nz@You cannot submit onto CurrentThreadExecutor from its own threadz/CurrentThreadExecutor already quit or is broken) r) r* r+ r1 r/ r r r. r3 )r r r r �fr6 r r r �_submitV s � z CurrentThreadExecutor._submitc O s | j |g|�R i |��S r )r9 )r r r r r r r �submitr s z CurrentThreadExecutor.submitr# )r r0 r N)r$ r% r& r' r r7 r r r r r r9 r r: r r r r r( 1 s ��� � �r( )r, �sysr) �concurrent.futuresr r �typingr r r r r � version_infor �typing_extensionsr r r r r( r r r r �<module> s __pycache__/wsgi.cpython-310.pyc 0000644 00000011312 15030103312 0012410 0 ustar 00 o �ha � @ sH d dl mZ d dlmZ d dlmZmZ G dd� d�ZG dd� d�ZdS ) � )�BytesIO)�SpooledTemporaryFile)� AsyncToSync� sync_to_asyncc @ s e Zd ZdZdd� Zdd� ZdS )� WsgiToAsgizG Wraps a WSGI application to make it into an ASGI application. c C s || _ d S �N)�wsgi_application��selfr � r �?/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/wsgi.py�__init__ s zWsgiToAsgi.__init__c � s �t | j�|||�I dH dS )z� ASGI application instantiation point. We return a new WsgiToAsgiInstance here with the WSGI app and the scope, ready to respond when it is __call__ed. N)�WsgiToAsgiInstancer )r �scope�receive�sendr r r �__call__ s � zWsgiToAsgi.__call__N)�__name__� __module__� __qualname__�__doc__r r r r r r r s r c @ s> e Zd ZdZdd� Zdd� Zdd� Zd d d �Zed d � �Z dS )r z; Per-socket instance of a wrapped WSGI application c C s || _ d| _d | _d S )NF)r �response_started�response_content_lengthr r r r r s zWsgiToAsgiInstance.__init__c � s� �|d dkr t d��|| _tdd��<} |� I d H }|d dkr%t d��|�|�d d �� |�d �s4nq|�d � t|�| _| �|�I d H W d � d S 1 sRw Y d S ) N�type�httpz&WSGI wrapper received a non-HTTP scopei )�max_sizeTz http.requestz0WSGI wrapper received a non-HTTP-request message�body� � more_bodyr ) � ValueErrorr r �write�get�seekr � sync_send� run_wsgi_app)r r r r r �messager r r r " s" � � "�zWsgiToAsgiInstance.__call__c C s` |� dd��d��d�}|d �d��d�}|�|�r#|t|�d� }|d |||d �d �d |d d |� d d�|t� dddd� }d|v rY|d d |d< t|d d �|d< nd|d< d|d< |� d�durp|d d |d< | j� dg �D ]6\}}|�d�}|d kr�d }n|d kr�d }n d |�� � d!d"� }|�d�}||v r�|| d# | }|||< qw|S )$zM Builds a scope and request body into a WSGI environ object. � root_path� �utf8�latin1�pathN�method� query_string�asciizHTTP/%s� http_version)� r �schemer TF) �REQUEST_METHOD� SCRIPT_NAME� PATH_INFO� QUERY_STRING�SERVER_PROTOCOLz wsgi.versionzwsgi.url_schemez wsgi.inputz wsgi.errorszwsgi.multithreadzwsgi.multiprocessz wsgi.run_once�serverr � SERVER_NAMEr/ � SERVER_PORT� localhost�80�client� REMOTE_ADDR�headers�content-length�CONTENT_LENGTHz content-type� CONTENT_TYPEzHTTP_%s�-�_�,) r! �encode�decode� startswith�lenr �strr �upper�replace) r r r � script_name� path_info�environ�name�value�corrected_namer r r � build_environ5 sH � z WsgiToAsgiInstance.build_environNc C s� | j r |d �|d ��t| d�r|du rtd��|�dd�\}}t|�}dd� |D �}d| _|D ]\}}|�� d kr@t|�| _q1d ||d �| _dS ) z/ WSGI start_response callable. r/ � �response_startNz=You cannot call start_response a second time without exc_info� c S s( g | ]\}}|� � �d �|�d �f�qS )r- )�lowerrD )�.0rN rO r r r � <listcomp>v s ��z5WsgiToAsgiInstance.start_response.<locals>.<listcomp>r> zhttp.response.start)r �statusr= ) r �with_traceback�hasattrr �split�intr rU rS ) r rX �response_headers�exc_info� status_coderB r= rN rO r r r �start_responsef s( �� � �z!WsgiToAsgiInstance.start_responsec C s� | � | j|�}d}| �|| j�D ]:}| js d| _| �| j� | jdur4| j| }t|�|kr4|d|� }| �d|dd�� |t|�7 }|| jkrJ nq| jsWd| _| �| j� | �ddi� dS )z� Called in a subthread to run the WSGI app. We encapsulate like this so that the start_response callable is called in the same thread. r TNzhttp.response.body)r r r r ) rQ r r r` r r# rS r rG )r r rM � bytes_sent�output� bytes_allowedr r r r$ � s* � � z WsgiToAsgiInstance.run_wsgi_appr ) r r r r r r rQ r` r r$ r r r r r s 1 r N) �ior �tempfiler � asgiref.syncr r r r r r r r �<module> s __pycache__/timeout.cpython-310.pyc 0000644 00000006330 15030103312 0013131 0 ustar 00 o �h+ � @ sJ d dl Z d dlZd dlmZ d dlmZ d dlmZmZ G dd� d�ZdS )� N)� TracebackType)�Any)�Optional�Typec @ s� e Zd ZdZdd�d ee deej ddfdd�Zd dd �Z d e e d e d e dee fd d�Zd dd�Zd e e d e d e ddfdd�Zede fdd��Zedee fdd��Zd dd�Zd e e ddfdd�Zd dd �ZdS ) �timeouta� timeout context manager. Useful in cases when you want to apply timeout logic around block of code or in cases when asyncio.wait_for is not suitable. For example: >>> with timeout(0.001): ... async with aiohttp.get('https://github.com') as r: ... await r.text() timeout - value in seconds or None to disable timeout logic loop - asyncio compatible event loop N)�loopr �returnc C sF || _ |d u r t�� }nt�dt� || _d | _d| _d | _ d | _ d S )Nz-The loop argument to timeout() is deprecated.F) �_timeout�asyncio�get_running_loop�warnings�warn�DeprecationWarning�_loop�_task� _cancelled�_cancel_handler� _cancel_at)�selfr r � r �B/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/timeout.py�__init__ s � ztimeout.__init__c C s | � � S �N�� _do_enter�r r r r � __enter__2 s ztimeout.__enter__�exc_type�exc_val�exc_tbc C s | � |� d S r ��_do_exit�r r r r r r r �__exit__5 s ztimeout.__exit__c � s �| � � S r r r r r r � __aenter__> s �ztimeout.__aenter__c � s �| � |� d S r r r" r r r � __aexit__A s �ztimeout.__aexit__c C s | j S r )r r r r r �expiredI s ztimeout.expiredc C s$ | j d urt| j | j�� d�S d S )Ng )r �maxr �timer r r r � remainingM s ztimeout.remainingc C st | j d u r| S t�| j�| _| jd u rtd��| j dkr%| j�| j� | S | j�� | j | _ | j� | j | j�| _ | S )Nz4Timeout context manager should be used inside a taskr ) r r � current_taskr r � RuntimeError� call_soon� _cancel_taskr( r �call_atr r r r r r T s � ztimeout._do_enterc C sP |t ju r| jrd | _d | _t j�| jd ur#| jd ur#| j�� d | _d | _d S r )r �CancelledErrorr r r � TimeoutErrorr �cancel)r r r r r r! h s ztimeout._do_exitc C s" | j d ur| j �� d| _d S d S )NT)r r1 r r r r r r- s s �ztimeout._cancel_task)r r )r N)�__name__� __module__� __qualname__�__doc__r �floatr �AbstractEventLoopr r r � BaseExceptionr �boolr# r$ r% �propertyr& r) r r! r- r r r r r sJ ��� � ��� � ��� � r ) r r �typesr �typingr r r r r r r r �<module> s __pycache__/compatibility.cpython-310.pyc 0000644 00000002625 15030103312 0014317 0 ustar 00 o �hF � @ s0 d dl Z ddlmZ dd� Zdd� Zdd � ZdS ) � N� )�iscoroutinefunctionc C sP t | dd�rdS t | dd�rdS t�| �rdS t| d�r#t| j�r#dS t| � S )zY Tests to see if an application is a legacy-style (double-callable) application. �_asgi_single_callableF�_asgi_double_callableT�__call__)�getattr�inspect�isclass�hasattrr r �� application� r �H/usr/local/CyberCP/lib/python3.10/site-packages/asgiref/compatibility.py�is_double_callable s r c s � fdd�}|S )zS Transforms a double-callable ASGI application into a single-callable one. c � s �� | �}|||�I d H S )Nr )�scope�receive�send�instancer r r �new_application! s �z2double_to_single_callable.<locals>.new_applicationr )r r r r r �double_to_single_callable s r c C s t | �rt| �} | S )z� Takes either a single- or double-callable application and always returns it in single-callable style. Use this to add backwards compatibility for ASGI 2.0 applications to your server/test harness/etc. )r r r r r r �guarantee_single_callable( s r )r �syncr r r r r r r r �<module> s local.py 0000644 00000011310 15030103312 0006170 0 ustar 00 import asyncio import contextlib import contextvars import threading from typing import Any, Dict, Union class _CVar: """Storage utility for Local.""" def __init__(self) -> None: self._data: "contextvars.ContextVar[Dict[str, Any]]" = contextvars.ContextVar( "asgiref.local" ) def __getattr__(self, key): storage_object = self._data.get({}) try: return storage_object[key] except KeyError: raise AttributeError(f"{self!r} object has no attribute {key!r}") def __setattr__(self, key: str, value: Any) -> None: if key == "_data": return super().__setattr__(key, value) storage_object = self._data.get({}) storage_object[key] = value self._data.set(storage_object) def __delattr__(self, key: str) -> None: storage_object = self._data.get({}) if key in storage_object: del storage_object[key] self._data.set(storage_object) else: raise AttributeError(f"{self!r} object has no attribute {key!r}") class Local: """Local storage for async tasks. This is a namespace object (similar to `threading.local`) where data is also local to the current async task (if there is one). In async threads, local means in the same sense as the `contextvars` module - i.e. a value set in an async frame will be visible: - to other async code `await`-ed from this frame. - to tasks spawned using `asyncio` utilities (`create_task`, `wait_for`, `gather` and probably others). - to code scheduled in a sync thread using `sync_to_async` In "sync" threads (a thread with no async event loop running), the data is thread-local, but additionally shared with async code executed via the `async_to_sync` utility, which schedules async code in a new thread and copies context across to that thread. If `thread_critical` is True, then the local will only be visible per-thread, behaving exactly like `threading.local` if the thread is sync, and as `contextvars` if the thread is async. This allows genuinely thread-sensitive code (such as DB handles) to be kept stricly to their initial thread and disable the sharing across `sync_to_async` and `async_to_sync` wrapped calls. Unlike plain `contextvars` objects, this utility is threadsafe. """ def __init__(self, thread_critical: bool = False) -> None: self._thread_critical = thread_critical self._thread_lock = threading.RLock() self._storage: "Union[threading.local, _CVar]" if thread_critical: # Thread-local storage self._storage = threading.local() else: # Contextvar storage self._storage = _CVar() @contextlib.contextmanager def _lock_storage(self): # Thread safe access to storage if self._thread_critical: try: # this is a test for are we in a async or sync # thread - will raise RuntimeError if there is # no current loop asyncio.get_running_loop() except RuntimeError: # We are in a sync thread, the storage is # just the plain thread local (i.e, "global within # this thread" - it doesn't matter where you are # in a call stack you see the same storage) yield self._storage else: # We are in an async thread - storage is still # local to this thread, but additionally should # behave like a context var (is only visible with # the same async call stack) # Ensure context exists in the current thread if not hasattr(self._storage, "cvar"): self._storage.cvar = _CVar() # self._storage is a thread local, so the members # can't be accessed in another thread (we don't # need any locks) yield self._storage.cvar else: # Lock for thread_critical=False as other threads # can access the exact same storage object with self._thread_lock: yield self._storage def __getattr__(self, key): with self._lock_storage() as storage: return getattr(storage, key) def __setattr__(self, key, value): if key in ("_local", "_storage", "_thread_critical", "_thread_lock"): return super().__setattr__(key, value) with self._lock_storage() as storage: setattr(storage, key, value) def __delattr__(self, key): with self._lock_storage() as storage: delattr(storage, key) typing.py 0000644 00000014170 15030103312 0006417 0 ustar 00 import sys from typing import ( Any, Awaitable, Callable, Dict, Iterable, Literal, Optional, Protocol, Tuple, Type, TypedDict, Union, ) if sys.version_info >= (3, 11): from typing import NotRequired else: from typing_extensions import NotRequired __all__ = ( "ASGIVersions", "HTTPScope", "WebSocketScope", "LifespanScope", "WWWScope", "Scope", "HTTPRequestEvent", "HTTPResponseStartEvent", "HTTPResponseBodyEvent", "HTTPResponseTrailersEvent", "HTTPResponsePathsendEvent", "HTTPServerPushEvent", "HTTPDisconnectEvent", "WebSocketConnectEvent", "WebSocketAcceptEvent", "WebSocketReceiveEvent", "WebSocketSendEvent", "WebSocketResponseStartEvent", "WebSocketResponseBodyEvent", "WebSocketDisconnectEvent", "WebSocketCloseEvent", "LifespanStartupEvent", "LifespanShutdownEvent", "LifespanStartupCompleteEvent", "LifespanStartupFailedEvent", "LifespanShutdownCompleteEvent", "LifespanShutdownFailedEvent", "ASGIReceiveEvent", "ASGISendEvent", "ASGIReceiveCallable", "ASGISendCallable", "ASGI2Protocol", "ASGI2Application", "ASGI3Application", "ASGIApplication", ) class ASGIVersions(TypedDict): spec_version: str version: Union[Literal["2.0"], Literal["3.0"]] class HTTPScope(TypedDict): type: Literal["http"] asgi: ASGIVersions http_version: str method: str scheme: str path: str raw_path: bytes query_string: bytes root_path: str headers: Iterable[Tuple[bytes, bytes]] client: Optional[Tuple[str, int]] server: Optional[Tuple[str, Optional[int]]] state: NotRequired[Dict[str, Any]] extensions: Optional[Dict[str, Dict[object, object]]] class WebSocketScope(TypedDict): type: Literal["websocket"] asgi: ASGIVersions http_version: str scheme: str path: str raw_path: bytes query_string: bytes root_path: str headers: Iterable[Tuple[bytes, bytes]] client: Optional[Tuple[str, int]] server: Optional[Tuple[str, Optional[int]]] subprotocols: Iterable[str] state: NotRequired[Dict[str, Any]] extensions: Optional[Dict[str, Dict[object, object]]] class LifespanScope(TypedDict): type: Literal["lifespan"] asgi: ASGIVersions state: NotRequired[Dict[str, Any]] WWWScope = Union[HTTPScope, WebSocketScope] Scope = Union[HTTPScope, WebSocketScope, LifespanScope] class HTTPRequestEvent(TypedDict): type: Literal["http.request"] body: bytes more_body: bool class HTTPResponseDebugEvent(TypedDict): type: Literal["http.response.debug"] info: Dict[str, object] class HTTPResponseStartEvent(TypedDict): type: Literal["http.response.start"] status: int headers: Iterable[Tuple[bytes, bytes]] trailers: bool class HTTPResponseBodyEvent(TypedDict): type: Literal["http.response.body"] body: bytes more_body: bool class HTTPResponseTrailersEvent(TypedDict): type: Literal["http.response.trailers"] headers: Iterable[Tuple[bytes, bytes]] more_trailers: bool class HTTPResponsePathsendEvent(TypedDict): type: Literal["http.response.pathsend"] path: str class HTTPServerPushEvent(TypedDict): type: Literal["http.response.push"] path: str headers: Iterable[Tuple[bytes, bytes]] class HTTPDisconnectEvent(TypedDict): type: Literal["http.disconnect"] class WebSocketConnectEvent(TypedDict): type: Literal["websocket.connect"] class WebSocketAcceptEvent(TypedDict): type: Literal["websocket.accept"] subprotocol: Optional[str] headers: Iterable[Tuple[bytes, bytes]] class WebSocketReceiveEvent(TypedDict): type: Literal["websocket.receive"] bytes: Optional[bytes] text: Optional[str] class WebSocketSendEvent(TypedDict): type: Literal["websocket.send"] bytes: Optional[bytes] text: Optional[str] class WebSocketResponseStartEvent(TypedDict): type: Literal["websocket.http.response.start"] status: int headers: Iterable[Tuple[bytes, bytes]] class WebSocketResponseBodyEvent(TypedDict): type: Literal["websocket.http.response.body"] body: bytes more_body: bool class WebSocketDisconnectEvent(TypedDict): type: Literal["websocket.disconnect"] code: int class WebSocketCloseEvent(TypedDict): type: Literal["websocket.close"] code: int reason: Optional[str] class LifespanStartupEvent(TypedDict): type: Literal["lifespan.startup"] class LifespanShutdownEvent(TypedDict): type: Literal["lifespan.shutdown"] class LifespanStartupCompleteEvent(TypedDict): type: Literal["lifespan.startup.complete"] class LifespanStartupFailedEvent(TypedDict): type: Literal["lifespan.startup.failed"] message: str class LifespanShutdownCompleteEvent(TypedDict): type: Literal["lifespan.shutdown.complete"] class LifespanShutdownFailedEvent(TypedDict): type: Literal["lifespan.shutdown.failed"] message: str ASGIReceiveEvent = Union[ HTTPRequestEvent, HTTPDisconnectEvent, WebSocketConnectEvent, WebSocketReceiveEvent, WebSocketDisconnectEvent, LifespanStartupEvent, LifespanShutdownEvent, ] ASGISendEvent = Union[ HTTPResponseStartEvent, HTTPResponseBodyEvent, HTTPResponseTrailersEvent, HTTPServerPushEvent, HTTPDisconnectEvent, WebSocketAcceptEvent, WebSocketSendEvent, WebSocketResponseStartEvent, WebSocketResponseBodyEvent, WebSocketCloseEvent, LifespanStartupCompleteEvent, LifespanStartupFailedEvent, LifespanShutdownCompleteEvent, LifespanShutdownFailedEvent, ] ASGIReceiveCallable = Callable[[], Awaitable[ASGIReceiveEvent]] ASGISendCallable = Callable[[ASGISendEvent], Awaitable[None]] class ASGI2Protocol(Protocol): def __init__(self, scope: Scope) -> None: ... async def __call__( self, receive: ASGIReceiveCallable, send: ASGISendCallable ) -> None: ... ASGI2Application = Type[ASGI2Protocol] ASGI3Application = Callable[ [ Scope, ASGIReceiveCallable, ASGISendCallable, ], Awaitable[None], ] ASGIApplication = Union[ASGI2Application, ASGI3Application] py.typed 0000644 00000000000 15030103312 0006215 0 ustar 00 testing.py 0000644 00000006631 15030103312 0006565 0 ustar 00 import asyncio import contextvars import time from .compatibility import guarantee_single_callable from .timeout import timeout as async_timeout class ApplicationCommunicator: """ Runs an ASGI application in a test mode, allowing sending of messages to it and retrieval of messages it sends. """ def __init__(self, application, scope): self.application = guarantee_single_callable(application) self.scope = scope self.input_queue = asyncio.Queue() self.output_queue = asyncio.Queue() # Clear context - this ensures that context vars set in the testing scope # are not "leaked" into the application which would normally begin with # an empty context. In Python >= 3.11 this could also be written as: # asyncio.create_task(..., context=contextvars.Context()) self.future = contextvars.Context().run( asyncio.create_task, self.application(scope, self.input_queue.get, self.output_queue.put), ) async def wait(self, timeout=1): """ Waits for the application to stop itself and returns any exceptions. """ try: async with async_timeout(timeout): try: await self.future self.future.result() except asyncio.CancelledError: pass finally: if not self.future.done(): self.future.cancel() try: await self.future except asyncio.CancelledError: pass def stop(self, exceptions=True): if not self.future.done(): self.future.cancel() elif exceptions: # Give a chance to raise any exceptions self.future.result() def __del__(self): # Clean up on deletion try: self.stop(exceptions=False) except RuntimeError: # Event loop already stopped pass async def send_input(self, message): """ Sends a single message to the application """ # Give it the message await self.input_queue.put(message) async def receive_output(self, timeout=1): """ Receives a single message from the application, with optional timeout. """ # Make sure there's not an exception to raise from the task if self.future.done(): self.future.result() # Wait and receive the message try: async with async_timeout(timeout): return await self.output_queue.get() except asyncio.TimeoutError as e: # See if we have another error to raise inside if self.future.done(): self.future.result() else: self.future.cancel() try: await self.future except asyncio.CancelledError: pass raise e async def receive_nothing(self, timeout=0.1, interval=0.01): """ Checks that there is no message to receive in the given time. """ # `interval` has precedence over `timeout` start = time.monotonic() while time.monotonic() - start < timeout: if not self.output_queue.empty(): return False await asyncio.sleep(interval) return self.output_queue.empty() sync.py 0000644 00000051704 15030103312 0006065 0 ustar 00 import asyncio import asyncio.coroutines import contextvars import functools import inspect import os import sys import threading import warnings import weakref from concurrent.futures import Future, ThreadPoolExecutor from typing import ( TYPE_CHECKING, Any, Awaitable, Callable, Coroutine, Dict, Generic, List, Optional, TypeVar, Union, overload, ) from .current_thread_executor import CurrentThreadExecutor from .local import Local if sys.version_info >= (3, 10): from typing import ParamSpec else: from typing_extensions import ParamSpec if TYPE_CHECKING: # This is not available to import at runtime from _typeshed import OptExcInfo _F = TypeVar("_F", bound=Callable[..., Any]) _P = ParamSpec("_P") _R = TypeVar("_R") def _restore_context(context: contextvars.Context) -> None: # Check for changes in contextvars, and set them to the current # context for downstream consumers for cvar in context: cvalue = context.get(cvar) try: if cvar.get() != cvalue: cvar.set(cvalue) except LookupError: cvar.set(cvalue) # Python 3.12 deprecates asyncio.iscoroutinefunction() as an alias for # inspect.iscoroutinefunction(), whilst also removing the _is_coroutine marker. # The latter is replaced with the inspect.markcoroutinefunction decorator. # Until 3.12 is the minimum supported Python version, provide a shim. if hasattr(inspect, "markcoroutinefunction"): iscoroutinefunction = inspect.iscoroutinefunction markcoroutinefunction: Callable[[_F], _F] = inspect.markcoroutinefunction else: iscoroutinefunction = asyncio.iscoroutinefunction # type: ignore[assignment] def markcoroutinefunction(func: _F) -> _F: func._is_coroutine = asyncio.coroutines._is_coroutine # type: ignore return func class ThreadSensitiveContext: """Async context manager to manage context for thread sensitive mode This context manager controls which thread pool executor is used when in thread sensitive mode. By default, a single thread pool executor is shared within a process. The ThreadSensitiveContext() context manager may be used to specify a thread pool per context. This context manager is re-entrant, so only the outer-most call to ThreadSensitiveContext will set the context. Usage: >>> import time >>> async with ThreadSensitiveContext(): ... await sync_to_async(time.sleep, 1)() """ def __init__(self): self.token = None async def __aenter__(self): try: SyncToAsync.thread_sensitive_context.get() except LookupError: self.token = SyncToAsync.thread_sensitive_context.set(self) return self async def __aexit__(self, exc, value, tb): if not self.token: return executor = SyncToAsync.context_to_thread_executor.pop(self, None) if executor: executor.shutdown() SyncToAsync.thread_sensitive_context.reset(self.token) class AsyncToSync(Generic[_P, _R]): """ Utility class which turns an awaitable that only works on the thread with the event loop into a synchronous callable that works in a subthread. If the call stack contains an async loop, the code runs there. Otherwise, the code runs in a new loop in a new thread. Either way, this thread then pauses and waits to run any thread_sensitive code called from further down the call stack using SyncToAsync, before finally exiting once the async task returns. """ # Keeps a reference to the CurrentThreadExecutor in local context, so that # any sync_to_async inside the wrapped code can find it. executors: "Local" = Local() # When we can't find a CurrentThreadExecutor from the context, such as # inside create_task, we'll look it up here from the running event loop. loop_thread_executors: "Dict[asyncio.AbstractEventLoop, CurrentThreadExecutor]" = {} def __init__( self, awaitable: Union[ Callable[_P, Coroutine[Any, Any, _R]], Callable[_P, Awaitable[_R]], ], force_new_loop: bool = False, ): if not callable(awaitable) or ( not iscoroutinefunction(awaitable) and not iscoroutinefunction(getattr(awaitable, "__call__", awaitable)) ): # Python does not have very reliable detection of async functions # (lots of false negatives) so this is just a warning. warnings.warn( "async_to_sync was passed a non-async-marked callable", stacklevel=2 ) self.awaitable = awaitable try: self.__self__ = self.awaitable.__self__ # type: ignore[union-attr] except AttributeError: pass self.force_new_loop = force_new_loop self.main_event_loop = None try: self.main_event_loop = asyncio.get_running_loop() except RuntimeError: # There's no event loop in this thread. pass def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: __traceback_hide__ = True # noqa: F841 if not self.force_new_loop and not self.main_event_loop: # There's no event loop in this thread. Look for the threadlocal if # we're inside SyncToAsync main_event_loop_pid = getattr( SyncToAsync.threadlocal, "main_event_loop_pid", None ) # We make sure the parent loop is from the same process - if # they've forked, this is not going to be valid any more (#194) if main_event_loop_pid and main_event_loop_pid == os.getpid(): self.main_event_loop = getattr( SyncToAsync.threadlocal, "main_event_loop", None ) # You can't call AsyncToSync from a thread with a running event loop try: event_loop = asyncio.get_running_loop() except RuntimeError: pass else: if event_loop.is_running(): raise RuntimeError( "You cannot use AsyncToSync in the same thread as an async event loop - " "just await the async function directly." ) # Make a future for the return information call_result: "Future[_R]" = Future() # Make a CurrentThreadExecutor we'll use to idle in this thread - we # need one for every sync frame, even if there's one above us in the # same thread. old_executor = getattr(self.executors, "current", None) current_executor = CurrentThreadExecutor() self.executors.current = current_executor # Wrapping context in list so it can be reassigned from within # `main_wrap`. context = [contextvars.copy_context()] # Get task context so that parent task knows which task to propagate # an asyncio.CancelledError to. task_context = getattr(SyncToAsync.threadlocal, "task_context", None) loop = None # Use call_soon_threadsafe to schedule a synchronous callback on the # main event loop's thread if it's there, otherwise make a new loop # in this thread. try: awaitable = self.main_wrap( call_result, sys.exc_info(), task_context, context, *args, **kwargs, ) if not (self.main_event_loop and self.main_event_loop.is_running()): # Make our own event loop - in a new thread - and run inside that. loop = asyncio.new_event_loop() self.loop_thread_executors[loop] = current_executor loop_executor = ThreadPoolExecutor(max_workers=1) loop_future = loop_executor.submit( self._run_event_loop, loop, awaitable ) if current_executor: # Run the CurrentThreadExecutor until the future is done current_executor.run_until_future(loop_future) # Wait for future and/or allow for exception propagation loop_future.result() else: # Call it inside the existing loop self.main_event_loop.call_soon_threadsafe( self.main_event_loop.create_task, awaitable ) if current_executor: # Run the CurrentThreadExecutor until the future is done current_executor.run_until_future(call_result) finally: # Clean up any executor we were running if loop is not None: del self.loop_thread_executors[loop] _restore_context(context[0]) # Restore old current thread executor state self.executors.current = old_executor # Wait for results from the future. return call_result.result() def _run_event_loop(self, loop, coro): """ Runs the given event loop (designed to be called in a thread). """ asyncio.set_event_loop(loop) try: loop.run_until_complete(coro) finally: try: # mimic asyncio.run() behavior # cancel unexhausted async generators tasks = asyncio.all_tasks(loop) for task in tasks: task.cancel() async def gather(): await asyncio.gather(*tasks, return_exceptions=True) loop.run_until_complete(gather()) for task in tasks: if task.cancelled(): continue if task.exception() is not None: loop.call_exception_handler( { "message": "unhandled exception during loop shutdown", "exception": task.exception(), "task": task, } ) if hasattr(loop, "shutdown_asyncgens"): loop.run_until_complete(loop.shutdown_asyncgens()) finally: loop.close() asyncio.set_event_loop(self.main_event_loop) def __get__(self, parent: Any, objtype: Any) -> Callable[_P, _R]: """ Include self for methods """ func = functools.partial(self.__call__, parent) return functools.update_wrapper(func, self.awaitable) async def main_wrap( self, call_result: "Future[_R]", exc_info: "OptExcInfo", task_context: "Optional[List[asyncio.Task[Any]]]", context: List[contextvars.Context], *args: _P.args, **kwargs: _P.kwargs, ) -> None: """ Wraps the awaitable with something that puts the result into the result/exception future. """ __traceback_hide__ = True # noqa: F841 if context is not None: _restore_context(context[0]) current_task = asyncio.current_task() if current_task is not None and task_context is not None: task_context.append(current_task) try: # If we have an exception, run the function inside the except block # after raising it so exc_info is correctly populated. if exc_info[1]: try: raise exc_info[1] except BaseException: result = await self.awaitable(*args, **kwargs) else: result = await self.awaitable(*args, **kwargs) except BaseException as e: call_result.set_exception(e) else: call_result.set_result(result) finally: if current_task is not None and task_context is not None: task_context.remove(current_task) context[0] = contextvars.copy_context() class SyncToAsync(Generic[_P, _R]): """ Utility class which turns a synchronous callable into an awaitable that runs in a threadpool. It also sets a threadlocal inside the thread so calls to AsyncToSync can escape it. If thread_sensitive is passed, the code will run in the same thread as any outer code. This is needed for underlying Python code that is not threadsafe (for example, code which handles SQLite database connections). If the outermost program is async (i.e. SyncToAsync is outermost), then this will be a dedicated single sub-thread that all sync code runs in, one after the other. If the outermost program is sync (i.e. AsyncToSync is outermost), this will just be the main thread. This is achieved by idling with a CurrentThreadExecutor while AsyncToSync is blocking its sync parent, rather than just blocking. If executor is passed in, that will be used instead of the loop's default executor. In order to pass in an executor, thread_sensitive must be set to False, otherwise a TypeError will be raised. """ # Storage for main event loop references threadlocal = threading.local() # Single-thread executor for thread-sensitive code single_thread_executor = ThreadPoolExecutor(max_workers=1) # Maintain a contextvar for the current execution context. Optionally used # for thread sensitive mode. thread_sensitive_context: "contextvars.ContextVar[ThreadSensitiveContext]" = ( contextvars.ContextVar("thread_sensitive_context") ) # Contextvar that is used to detect if the single thread executor # would be awaited on while already being used in the same context deadlock_context: "contextvars.ContextVar[bool]" = contextvars.ContextVar( "deadlock_context" ) # Maintaining a weak reference to the context ensures that thread pools are # erased once the context goes out of scope. This terminates the thread pool. context_to_thread_executor: "weakref.WeakKeyDictionary[ThreadSensitiveContext, ThreadPoolExecutor]" = ( weakref.WeakKeyDictionary() ) def __init__( self, func: Callable[_P, _R], thread_sensitive: bool = True, executor: Optional["ThreadPoolExecutor"] = None, ) -> None: if ( not callable(func) or iscoroutinefunction(func) or iscoroutinefunction(getattr(func, "__call__", func)) ): raise TypeError("sync_to_async can only be applied to sync functions.") self.func = func functools.update_wrapper(self, func) self._thread_sensitive = thread_sensitive markcoroutinefunction(self) if thread_sensitive and executor is not None: raise TypeError("executor must not be set when thread_sensitive is True") self._executor = executor try: self.__self__ = func.__self__ # type: ignore except AttributeError: pass async def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R: __traceback_hide__ = True # noqa: F841 loop = asyncio.get_running_loop() # Work out what thread to run the code in if self._thread_sensitive: current_thread_executor = getattr(AsyncToSync.executors, "current", None) if current_thread_executor: # If we have a parent sync thread above somewhere, use that executor = current_thread_executor elif self.thread_sensitive_context.get(None): # If we have a way of retrieving the current context, attempt # to use a per-context thread pool executor thread_sensitive_context = self.thread_sensitive_context.get() if thread_sensitive_context in self.context_to_thread_executor: # Re-use thread executor in current context executor = self.context_to_thread_executor[thread_sensitive_context] else: # Create new thread executor in current context executor = ThreadPoolExecutor(max_workers=1) self.context_to_thread_executor[thread_sensitive_context] = executor elif loop in AsyncToSync.loop_thread_executors: # Re-use thread executor for running loop executor = AsyncToSync.loop_thread_executors[loop] elif self.deadlock_context.get(False): raise RuntimeError( "Single thread executor already being used, would deadlock" ) else: # Otherwise, we run it in a fixed single thread executor = self.single_thread_executor self.deadlock_context.set(True) else: # Use the passed in executor, or the loop's default if it is None executor = self._executor context = contextvars.copy_context() child = functools.partial(self.func, *args, **kwargs) func = context.run task_context: List[asyncio.Task[Any]] = [] # Run the code in the right thread exec_coro = loop.run_in_executor( executor, functools.partial( self.thread_handler, loop, sys.exc_info(), task_context, func, child, ), ) ret: _R try: ret = await asyncio.shield(exec_coro) except asyncio.CancelledError: cancel_parent = True try: task = task_context[0] task.cancel() try: await task cancel_parent = False except asyncio.CancelledError: pass except IndexError: pass if exec_coro.done(): raise if cancel_parent: exec_coro.cancel() ret = await exec_coro finally: _restore_context(context) self.deadlock_context.set(False) return ret def __get__( self, parent: Any, objtype: Any ) -> Callable[_P, Coroutine[Any, Any, _R]]: """ Include self for methods """ func = functools.partial(self.__call__, parent) return functools.update_wrapper(func, self.func) def thread_handler(self, loop, exc_info, task_context, func, *args, **kwargs): """ Wraps the sync application with exception handling. """ __traceback_hide__ = True # noqa: F841 # Set the threadlocal for AsyncToSync self.threadlocal.main_event_loop = loop self.threadlocal.main_event_loop_pid = os.getpid() self.threadlocal.task_context = task_context # Run the function # If we have an exception, run the function inside the except block # after raising it so exc_info is correctly populated. if exc_info[1]: try: raise exc_info[1] except BaseException: return func(*args, **kwargs) else: return func(*args, **kwargs) @overload def async_to_sync( *, force_new_loop: bool = False, ) -> Callable[ [Union[Callable[_P, Coroutine[Any, Any, _R]], Callable[_P, Awaitable[_R]]]], Callable[_P, _R], ]: ... @overload def async_to_sync( awaitable: Union[ Callable[_P, Coroutine[Any, Any, _R]], Callable[_P, Awaitable[_R]], ], *, force_new_loop: bool = False, ) -> Callable[_P, _R]: ... def async_to_sync( awaitable: Optional[ Union[ Callable[_P, Coroutine[Any, Any, _R]], Callable[_P, Awaitable[_R]], ] ] = None, *, force_new_loop: bool = False, ) -> Union[ Callable[ [Union[Callable[_P, Coroutine[Any, Any, _R]], Callable[_P, Awaitable[_R]]]], Callable[_P, _R], ], Callable[_P, _R], ]: if awaitable is None: return lambda f: AsyncToSync( f, force_new_loop=force_new_loop, ) return AsyncToSync( awaitable, force_new_loop=force_new_loop, ) @overload def sync_to_async( *, thread_sensitive: bool = True, executor: Optional["ThreadPoolExecutor"] = None, ) -> Callable[[Callable[_P, _R]], Callable[_P, Coroutine[Any, Any, _R]]]: ... @overload def sync_to_async( func: Callable[_P, _R], *, thread_sensitive: bool = True, executor: Optional["ThreadPoolExecutor"] = None, ) -> Callable[_P, Coroutine[Any, Any, _R]]: ... def sync_to_async( func: Optional[Callable[_P, _R]] = None, *, thread_sensitive: bool = True, executor: Optional["ThreadPoolExecutor"] = None, ) -> Union[ Callable[[Callable[_P, _R]], Callable[_P, Coroutine[Any, Any, _R]]], Callable[_P, Coroutine[Any, Any, _R]], ]: if func is None: return lambda f: SyncToAsync( f, thread_sensitive=thread_sensitive, executor=executor, ) return SyncToAsync( func, thread_sensitive=thread_sensitive, executor=executor, ) compatibility.py 0000644 00000003106 15030103312 0007753 0 ustar 00 import inspect from .sync import iscoroutinefunction def is_double_callable(application): """ Tests to see if an application is a legacy-style (double-callable) application. """ # Look for a hint on the object first if getattr(application, "_asgi_single_callable", False): return False if getattr(application, "_asgi_double_callable", False): return True # Uninstanted classes are double-callable if inspect.isclass(application): return True # Instanted classes depend on their __call__ if hasattr(application, "__call__"): # We only check to see if its __call__ is a coroutine function - # if it's not, it still might be a coroutine function itself. if iscoroutinefunction(application.__call__): return False # Non-classes we just check directly return not iscoroutinefunction(application) def double_to_single_callable(application): """ Transforms a double-callable ASGI application into a single-callable one. """ async def new_application(scope, receive, send): instance = application(scope) return await instance(receive, send) return new_application def guarantee_single_callable(application): """ Takes either a single- or double-callable application and always returns it in single-callable style. Use this to add backwards compatibility for ASGI 2.0 applications to your server/test harness/etc. """ if is_double_callable(application): application = double_to_single_callable(application) return application wsgi.py 0000644 00000015141 15030103312 0006055 0 ustar 00 from io import BytesIO from tempfile import SpooledTemporaryFile from asgiref.sync import AsyncToSync, sync_to_async class WsgiToAsgi: """ Wraps a WSGI application to make it into an ASGI application. """ def __init__(self, wsgi_application): self.wsgi_application = wsgi_application async def __call__(self, scope, receive, send): """ ASGI application instantiation point. We return a new WsgiToAsgiInstance here with the WSGI app and the scope, ready to respond when it is __call__ed. """ await WsgiToAsgiInstance(self.wsgi_application)(scope, receive, send) class WsgiToAsgiInstance: """ Per-socket instance of a wrapped WSGI application """ def __init__(self, wsgi_application): self.wsgi_application = wsgi_application self.response_started = False self.response_content_length = None async def __call__(self, scope, receive, send): if scope["type"] != "http": raise ValueError("WSGI wrapper received a non-HTTP scope") self.scope = scope with SpooledTemporaryFile(max_size=65536) as body: # Alright, wait for the http.request messages while True: message = await receive() if message["type"] != "http.request": raise ValueError("WSGI wrapper received a non-HTTP-request message") body.write(message.get("body", b"")) if not message.get("more_body"): break body.seek(0) # Wrap send so it can be called from the subthread self.sync_send = AsyncToSync(send) # Call the WSGI app await self.run_wsgi_app(body) def build_environ(self, scope, body): """ Builds a scope and request body into a WSGI environ object. """ script_name = scope.get("root_path", "").encode("utf8").decode("latin1") path_info = scope["path"].encode("utf8").decode("latin1") if path_info.startswith(script_name): path_info = path_info[len(script_name) :] environ = { "REQUEST_METHOD": scope["method"], "SCRIPT_NAME": script_name, "PATH_INFO": path_info, "QUERY_STRING": scope["query_string"].decode("ascii"), "SERVER_PROTOCOL": "HTTP/%s" % scope["http_version"], "wsgi.version": (1, 0), "wsgi.url_scheme": scope.get("scheme", "http"), "wsgi.input": body, "wsgi.errors": BytesIO(), "wsgi.multithread": True, "wsgi.multiprocess": True, "wsgi.run_once": False, } # Get server name and port - required in WSGI, not in ASGI if "server" in scope: environ["SERVER_NAME"] = scope["server"][0] environ["SERVER_PORT"] = str(scope["server"][1]) else: environ["SERVER_NAME"] = "localhost" environ["SERVER_PORT"] = "80" if scope.get("client") is not None: environ["REMOTE_ADDR"] = scope["client"][0] # Go through headers and make them into environ entries for name, value in self.scope.get("headers", []): name = name.decode("latin1") if name == "content-length": corrected_name = "CONTENT_LENGTH" elif name == "content-type": corrected_name = "CONTENT_TYPE" else: corrected_name = "HTTP_%s" % name.upper().replace("-", "_") # HTTPbis say only ASCII chars are allowed in headers, but we latin1 just in case value = value.decode("latin1") if corrected_name in environ: value = environ[corrected_name] + "," + value environ[corrected_name] = value return environ def start_response(self, status, response_headers, exc_info=None): """ WSGI start_response callable. """ # Don't allow re-calling once response has begun if self.response_started: raise exc_info[1].with_traceback(exc_info[2]) # Don't allow re-calling without exc_info if hasattr(self, "response_start") and exc_info is None: raise ValueError( "You cannot call start_response a second time without exc_info" ) # Extract status code status_code, _ = status.split(" ", 1) status_code = int(status_code) # Extract headers headers = [ (name.lower().encode("ascii"), value.encode("ascii")) for name, value in response_headers ] # Extract content-length self.response_content_length = None for name, value in response_headers: if name.lower() == "content-length": self.response_content_length = int(value) # Build and send response start message. self.response_start = { "type": "http.response.start", "status": status_code, "headers": headers, } @sync_to_async def run_wsgi_app(self, body): """ Called in a subthread to run the WSGI app. We encapsulate like this so that the start_response callable is called in the same thread. """ # Translate the scope and incoming request body into a WSGI environ environ = self.build_environ(self.scope, body) # Run the WSGI app bytes_sent = 0 for output in self.wsgi_application(environ, self.start_response): # If this is the first response, include the response headers if not self.response_started: self.response_started = True self.sync_send(self.response_start) # If the application supplies a Content-Length header if self.response_content_length is not None: # The server should not transmit more bytes to the client than the header allows bytes_allowed = self.response_content_length - bytes_sent if len(output) > bytes_allowed: output = output[:bytes_allowed] self.sync_send( {"type": "http.response.body", "body": output, "more_body": True} ) bytes_sent += len(output) # The server should stop iterating over the response when enough data has been sent if bytes_sent == self.response_content_length: break # Close connection if not self.response_started: self.response_started = True self.sync_send(self.response_start) self.sync_send({"type": "http.response.body"}) server.py 0000644 00000013565 15030103312 0006422 0 ustar 00 import asyncio import logging import time import traceback from .compatibility import guarantee_single_callable logger = logging.getLogger(__name__) class StatelessServer: """ Base server class that handles basic concepts like application instance creation/pooling, exception handling, and similar, for stateless protocols (i.e. ones without actual incoming connections to the process) Your code should override the handle() method, doing whatever it needs to, and calling get_or_create_application_instance with a unique `scope_id` and `scope` for the scope it wants to get. If an application instance is found with the same `scope_id`, you are given its input queue, otherwise one is made for you with the scope provided and you are given that fresh new input queue. Either way, you should do something like: input_queue = self.get_or_create_application_instance( "user-123456", {"type": "testprotocol", "user_id": "123456", "username": "andrew"}, ) input_queue.put_nowait(message) If you try and create an application instance and there are already `max_application` instances, the oldest/least recently used one will be reclaimed and shut down to make space. Application coroutines that error will be found periodically (every 100ms by default) and have their exceptions printed to the console. Override application_exception() if you want to do more when this happens. If you override run(), make sure you handle things like launching the application checker. """ application_checker_interval = 0.1 def __init__(self, application, max_applications=1000): # Parameters self.application = application self.max_applications = max_applications # Initialisation self.application_instances = {} ### Mainloop and handling def run(self): """ Runs the asyncio event loop with our handler loop. """ event_loop = asyncio.get_event_loop() asyncio.ensure_future(self.application_checker()) try: event_loop.run_until_complete(self.handle()) except KeyboardInterrupt: logger.info("Exiting due to Ctrl-C/interrupt") async def handle(self): raise NotImplementedError("You must implement handle()") async def application_send(self, scope, message): """ Receives outbound sends from applications and handles them. """ raise NotImplementedError("You must implement application_send()") ### Application instance management def get_or_create_application_instance(self, scope_id, scope): """ Creates an application instance and returns its queue. """ if scope_id in self.application_instances: self.application_instances[scope_id]["last_used"] = time.time() return self.application_instances[scope_id]["input_queue"] # See if we need to delete an old one while len(self.application_instances) > self.max_applications: self.delete_oldest_application_instance() # Make an instance of the application input_queue = asyncio.Queue() application_instance = guarantee_single_callable(self.application) # Run it, and stash the future for later checking future = asyncio.ensure_future( application_instance( scope=scope, receive=input_queue.get, send=lambda message: self.application_send(scope, message), ), ) self.application_instances[scope_id] = { "input_queue": input_queue, "future": future, "scope": scope, "last_used": time.time(), } return input_queue def delete_oldest_application_instance(self): """ Finds and deletes the oldest application instance """ oldest_time = min( details["last_used"] for details in self.application_instances.values() ) for scope_id, details in self.application_instances.items(): if details["last_used"] == oldest_time: self.delete_application_instance(scope_id) # Return to make sure we only delete one in case two have # the same oldest time return def delete_application_instance(self, scope_id): """ Removes an application instance (makes sure its task is stopped, then removes it from the current set) """ details = self.application_instances[scope_id] del self.application_instances[scope_id] if not details["future"].done(): details["future"].cancel() async def application_checker(self): """ Goes through the set of current application instance Futures and cleans up any that are done/prints exceptions for any that errored. """ while True: await asyncio.sleep(self.application_checker_interval) for scope_id, details in list(self.application_instances.items()): if details["future"].done(): exception = details["future"].exception() if exception: await self.application_exception(exception, details) try: del self.application_instances[scope_id] except KeyError: # Exception handling might have already got here before us. That's fine. pass async def application_exception(self, exception, application_details): """ Called whenever an application coroutine has an exception. """ logging.error( "Exception inside application: %s\n%s%s", exception, "".join(traceback.format_tb(exception.__traceback__)), f" {exception}", ) __init__.py 0000644 00000000026 15030103312 0006637 0 ustar 00 __version__ = "3.8.1" current_thread_executor.py 0000644 00000007611 15030103312 0012036 0 ustar 00 import queue import sys import threading from concurrent.futures import Executor, Future from typing import TYPE_CHECKING, Any, Callable, TypeVar, Union if sys.version_info >= (3, 10): from typing import ParamSpec else: from typing_extensions import ParamSpec _T = TypeVar("_T") _P = ParamSpec("_P") _R = TypeVar("_R") class _WorkItem: """ Represents an item needing to be run in the executor. Copied from ThreadPoolExecutor (but it's private, so we're not going to rely on importing it) """ def __init__( self, future: "Future[_R]", fn: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs, ): self.future = future self.fn = fn self.args = args self.kwargs = kwargs def run(self) -> None: __traceback_hide__ = True # noqa: F841 if not self.future.set_running_or_notify_cancel(): return try: result = self.fn(*self.args, **self.kwargs) except BaseException as exc: self.future.set_exception(exc) # Break a reference cycle with the exception 'exc' self = None # type: ignore[assignment] else: self.future.set_result(result) class CurrentThreadExecutor(Executor): """ An Executor that actually runs code in the thread it is instantiated in. Passed to other threads running async code, so they can run sync code in the thread they came from. """ def __init__(self) -> None: self._work_thread = threading.current_thread() self._work_queue: queue.Queue[Union[_WorkItem, "Future[Any]"]] = queue.Queue() self._broken = False def run_until_future(self, future: "Future[Any]") -> None: """ Runs the code in the work queue until a result is available from the future. Should be run from the thread the executor is initialised in. """ # Check we're in the right thread if threading.current_thread() != self._work_thread: raise RuntimeError( "You cannot run CurrentThreadExecutor from a different thread" ) future.add_done_callback(self._work_queue.put) # Keep getting and running work items until we get the future we're waiting for # back via the future's done callback. try: while True: # Get a work item and run it work_item = self._work_queue.get() if work_item is future: return assert isinstance(work_item, _WorkItem) work_item.run() del work_item finally: self._broken = True def _submit( self, fn: Callable[_P, _R], *args: _P.args, **kwargs: _P.kwargs, ) -> "Future[_R]": # Check they're not submitting from the same thread if threading.current_thread() == self._work_thread: raise RuntimeError( "You cannot submit onto CurrentThreadExecutor from its own thread" ) # Check they're not too late or the executor errored if self._broken: raise RuntimeError("CurrentThreadExecutor already quit or is broken") # Add to work queue f: "Future[_R]" = Future() work_item = _WorkItem(f, fn, *args, **kwargs) self._work_queue.put(work_item) # Return the future return f # Python 3.9+ has a new signature for submit with a "/" after `fn`, to enforce # it to be a positional argument. If we ignore[override] mypy on 3.9+ will be # happy but 3.8 will say that the ignore comment is unused, even when # defining them differently based on sys.version_info. # We should be able to remove this when we drop support for 3.8. if not TYPE_CHECKING: def submit(self, fn, *args, **kwargs): return self._submit(fn, *args, **kwargs)
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings