File: //usr/local/lib/python3.10/dist-packages/anyio/__pycache__/lowlevel.cpython-310.pyc
o
���gI � @ s� U d dl mZ d dlZd dlmZ d dlmZmZmZm Z m
Z
d dl m Z ddl
mZ e d�Ze d �Zd#d d
�Zd#dd�Zd#dd�Zd$dd�Ze � Zded< i Zded< edd�G dd � d ��ZG d d � d ej�ZG d d � d ee �ZG d!d"� d"ee �Z dS )%� )� annotationsN)� dataclass)�Any�Generic�Literal�TypeVar�overload)�WeakKeyDictionary� )�get_async_backend�T�D�return�Nonec � � �t � �� I dH dS )z�
Check for cancellation and allow the scheduler to switch to another task.
Equivalent to (but more efficient than)::
await checkpoint_if_cancelled()
await cancel_shielded_checkpoint()
.. versionadded:: 3.0
N)r �
checkpoint� r r �9/usr/local/lib/python3.10/dist-packages/anyio/lowlevel.pyr � �
r c � r )z�
Enter a checkpoint if the enclosing cancel scope has been cancelled.
This does not allow the scheduler to switch to a different task.
.. versionadded:: 3.0
N)r �checkpoint_if_cancelledr r r r r s � r c � r )z�
Allow the scheduler to switch to another task but without checking for cancellation.
Equivalent to (but potentially more efficient than)::
with CancelScope(shield=True):
await checkpoint()
.. versionadded:: 3.0
N)r �cancel_shielded_checkpointr r r r r * r r �objectc C s
t � �� S )ze
Return a backend specific token object that can be used to get back to the event
loop.
)r �
current_tokenr r r r r : s
r z&WeakKeyDictionary[Any, dict[str, Any]]� _run_varszdict[Any, _TokenWrapper]�_token_wrappersT)�frozenc @ s e Zd ZU dZded< dS )�
_TokenWrapper)�_token� __weakref__r r N)�__name__�
__module__� __qualname__� __slots__�__annotations__r r r r r G s
r c @ s e Zd Ze�� ZdS )� _NoValueSetN)r r r! �enum�auto� NO_VALUE_SETr r r r r$ M s r$ c @ s e Zd ZdZd dd�ZdS )
� RunvarToken��_var�_value� _redeemed�var� RunVar[T]�value�%T | Literal[_NoValueSet.NO_VALUE_SET]c C s || _ || _d| _d S )NFr) )�selfr- r/ r r r �__init__T s
zRunvarToken.__init__N)r- r. r/ r0 )r r r! r" r2 r r r r r( Q s r( c @ s� e Zd ZU dZdZejZded< e� Z ded< efd$d d �Z
e d%dd��Z e
d&dd��Ze
d'dd��Zefd(dd�Zd)dd �Zd*d d �Zd+d!d"�Zd#S ),�RunVarz[
Like a :class:`~contextvars.ContextVar`, except scoped to the running event loop.
��_name�_defaultz!Literal[_NoValueSet.NO_VALUE_SET]r' zset[_TokenWrapper]r �name�str�defaultr0 c C s || _ || _d S �Nr4 )r1 r7 r9 r r r r2 e s
zRunVar.__init__r � dict[str, T]c C s4 t � }zt| W S ty i }t|< | Y S w r: )r r �KeyError)r1 �token�run_varsr r r �
_current_varsk s
�zRunVar._current_varsr
�T | Dc C � d S r: r �r1 r9 r r r �gett � z
RunVar.getr c C rA r: r �r1 r r r rC w rD �%D | Literal[_NoValueSet.NO_VALUE_SET]c C s\ z| j | j W S ty$ |tjur| Y S | jtjur"| j Y S Y nw td| j� d���)NzRun variable "z!" has no value and no default set)r? r5 r< r3 r' r6 � LookupErrorrB r r r rC z s
�� �r/ �RunvarToken[T]c C s* | j }t| |�| jtj��}||| j< |S r: )r? r( rC r5 r3 r' )r1 r/ � current_varsr= r r r �set� s
z
RunVar.setr= r c C sf |j | ur td��|jrtd��|jtju r'z| j| j= W n ty& Y nw |j| j| j< d|_d S )Nz)This token does not belong to this RunVarz This token has already been usedT) r* �
ValueErrorr, r+ r$ r' r? r5 r< )r1 r= r r r �reset� s
�
z RunVar.resetc C s d| j �d�S )Nz
<RunVar name=�>)r5 rE r r r �__repr__� s zRunVar.__repr__N)r7 r8 r9 r0 )r r; )r9 r
r r@ )r r )r9 rF r r@ )r/ r r rH )r= rH r r )r r8 )r r r! �__doc__r" r$ r' r# rJ r r2 �propertyr? r rC rL rN r r r r r3 Z s$
� �
r3 )r r )r r ) �
__future__r r% � dataclassesr �typingr r r r r �weakrefr �_core._eventloopr r r
r r r r r r# r r �Enumr$ r( r3 r r r r �<module> s&