File: //usr/local/lib/python3.9/site-packages/celery/backends/__pycache__/dynamodb.cpython-39.pyc
a
X>h|L � @ s� d Z ddlmZ ddlmZ ddlmZmZ ddlmZm Z ddl
m Z ddl
mZ ddlmZ d d
lmZ zdd lZdd lmZ W n ey� d ZZY n0 d
Zedd�Zee�ZG dd� de�Zd S )z"AWS DynamoDB result store backend.� )�
namedtuple)�
ip_address)�sleep�time)�Any�Dict)�
_parse_url)�ImproperlyConfigured)�
get_logger� )�KeyValueStoreBackendN)� ClientError)�DynamoDBBackend�DynamoDBAttribute��name� data_typec sj e Zd ZdZdZdZdZdZdZdZ dZ
e ddd�Z e d d
d�Z
e d d d�Ze d
d d�Ze dd d�ZdZdZd?� fdd� Zedd� �Zd@dd�Zdd� Zdd� Zdd� Zdd � Zd d � Zd d � Z d!d"� Z dAd$d%�Z d&d'� Z d(d)� Z e!e"e!e#f d*�d+d,�Z$e!e"e!e#f d*�d-d.�Z%d/d0� Z&e'd1d2� �Z(d3d4� Z)d5d6� Z*d7d8� Z+d9d:� Z,e-e.d*�d;d<�Z/� fd=d>�Z0� Z1S )Br z�AWS DynamoDB result backend.
Raises:
celery.exceptions.ImproperlyConfigured:
if module :pypi:`boto3` is not available.
Zceleryr NT�id�Sr �result�BZ chord_count�N� timestamp�ttlc
s� t � j|i |�� || _|p | j| _ts0td��d}d }d }|d u�r�t|�\}} }
} } }
}| }| }|d u}|d u}||kr�td��|}| dks�t�| �r�d| � d|
� �| _ d| _
t � d�
| j �� n| | _
| jjj}|d �}|r�|| _ t|�d
| j��| _t|�d | j��| _|�d | j�}|�rtzt|�| _W n@ t�yr } z&t jd
|� d�|d� |�W Y d }~n
d }~0 0 |
�p~| j| _| j| j| jf| _d | _|�r�| j ||d� d S )NzBYou need to install the boto3 library to use the DynamoDB backend.Fz6You need to specify both the Access Key ID and Secret.� localhostzhttp://�:z us-east-1z*Using local-only DynamoDB endpoint URL: {}Zdynamodb_endpoint_url�read�writeZ ttl_secondszTTL must be a number; got "�")�exc_info)�
access_key_id�secret_access_key) �super�__init__�url�
table_name�boto3r � parse_urlr � _is_valid_ip� endpoint_url�
aws_region�logger�warning�formatZapp�conf�get�int�read_capacity_units�write_capacity_units�time_to_live_seconds�
ValueError�error�
_key_field� _value_field�_timestamp_field�_available_fields�_client� _get_client)�selfr$ r% �args�kwargsZaws_credentials_given�aws_access_key_id�aws_secret_access_key�schemeZregion�port�username�password�table�queryZaccess_key_givenZsecret_key_given�_getZconfig_endpoint_urlr �e�� __class__� �B/usr/local/lib/python3.9/site-packages/celery/backends/dynamodb.pyr# D s� �
����
����
���zDynamoDBBackend.__init__c C s( zt | � W dS ty" Y dS 0 d S )NTF)r r4 )�iprK rK rL r( � s
z DynamoDBBackend._is_valid_ipc C s| | j du rvd| ji}|dur,|�||d�� | jdur@| j|d< tjdi |��| _ | �� | �� durv| �� | � � | j S )zGet client connection.NZ region_name)r? r@ r) �dynamodb)rN )
r: r* �updater) r& �client�_get_or_create_table�_has_ttl�_validate_ttl_methods�_set_table_ttl)r< r r! Zclient_parametersrK rK rL r; � s(
��
�� zDynamoDBBackend._get_clientc C s6 | j j| j jd�g| j| j jdd�g| j| jd�d�S )z=Get the boto3 structure describing the DynamoDB table schema.)�
AttributeNameZ
AttributeType�HASH)rU ZKeyType)ZReadCapacityUnitsZWriteCapacityUnits)ZAttributeDefinitions� TableNameZ KeySchemaZProvisionedThroughput)r6 r r r% r1 r2 �r< rK rK rL �_get_table_schema� s ������z!DynamoDBBackend._get_table_schemac
C s� | � � }z| jj| jd�W S ty� } zz|jd �dd�}|dkr�| jjf i |��}t� d�
| j�� | � d� t� d�
| j�� |W Y d }~S |�W Y d }~n
d }~0 0 d S )
z=Create table if not exists, otherwise return the description.�rW �Error�Code�UnknownZResourceNotFoundExceptionz*DynamoDB Table {} did not exist, creating.�ACTIVEz#DynamoDB Table {} is now available.N) rY r: �describe_tabler% r
�responser/ Z create_tabler+ �infor- �_wait_for_table_status)r< Z table_schemarH �
error_code�table_descriptionrK rK rL rQ � s( ��
��z$DynamoDBBackend._get_or_create_tablec C s | j du rdS | j dkS )z�Return the desired Time to Live config.
- True: Enable TTL on the table; use expiry.
- False: Disable TTL on the table; don't use expiry.
- None: Ignore TTL on the table; don't use expiry.
Nr )r3 rX rK rK rL rR � s �zDynamoDBBackend._has_ttlc C sb d}g }t |�D ]}t| j|�s|�|� q|r^t�djd�|�d�� tdjd�|�d���dS )z:Verify boto support for the DynamoDB Time to Live methods.)�update_time_to_live�describe_time_to_livezdboto3 method(s) {methods} not found; ensure that boto3>=1.9.178 and botocore>=1.12.178 are installed�,)�methodsz#boto3 method(s) {methods} not foundN) �list�hasattrr: �appendr+ r5 r- �join�AttributeError)r< Zrequired_methodsZmissing_methods�methodrK rK rL rS � s" �����z%DynamoDBBackend._validate_ttl_methodsc C s | j | �� |d�d�S )zBGet the boto3 structure describing the DynamoDB TTL specification.)ZEnabledrU )rW ZTimeToLiveSpecification)r% rR )r< �
ttl_attr_namerK rK rL �_get_ttl_specification s
��z&DynamoDBBackend._get_ttl_specificationc
C s~ z| j j| jd�}W nd tyx } zL|jd �dd�}|jd �dd�}t�dj| j||d�� |�W Y d }~n
d }~0 0 |S )NrZ r[ r\ r] �MessagezJError describing Time to Live on DynamoDB table {table}: {code}: {message})rE �code�message) r: rf r% r
r` r/ r+ r5 r- )r< � descriptionrH rc �
error_messagerK rK rL �_get_table_ttl_description! s �
��z*DynamoDBBackend._get_table_ttl_descriptionc C s� | � � }|d d }|dv rd|d d }| �� r�|| jjkr�t�dj|dkrPdnd| jd �� |S nN|d
v r�| �� s�t�dj|d kr�d nd
| jd �� |S nt�dj|| jd�� |dkr�|n| jj}z@| j j
f i | j |d���}t� dj| j| �� | jjd�� |W S t
�y| } z\|jd �dd�}|jd �dd�}t�dj| �� �rRdnd| j||d�� |�W Y d}~n
d}~0 0 dS ) z,Enable or disable Time to Live on the table.ZTimeToLiveDescriptionZTimeToLiveStatus)�ENABLEDZENABLINGrU z5DynamoDB Time to Live is {situation} on table {table}rw zalready enabledzcurrently being enabled)Z situationrE )�DISABLEDZ DISABLINGrx zalready disabledzcurrently being disabledzWUnknown DynamoDB Time to Live status {status} on table {table}. Attempting to continue.)�statusrE )ro zUDynamoDB table Time to Live updated: table={table} enabled={enabled} attribute={attr})rE �enabled�attrr[ r\ r] rq zHError {action} Time to Live on DynamoDB table {table}: {code}: {message}ZenablingZ disabling)�actionrE rr rs N)rv rR �
_ttl_fieldr r+ �debugr- r% r, r: re rp ra r
r` r/ r5 ) r< rt ry Z
cur_attr_name� attr_name�
specificationrH rc ru rK rK rL rT 6 s|
� ��� ��� ��&�
�����
�� z DynamoDBBackend._set_table_ttlr^ c C sN d}|sJ| j j| jd�}t�d�| j|�� |d d }||k}td� qdS )z#Poll for the expected table status.FrZ z+Waiting for DynamoDB table {} to become {}.ZTableZ TableStatusr N)rP r_ r% r+ r~ r- r )r< �expectedZachieved_staterd Zcurrent_statusrK rK rL rb � s ��� z&DynamoDBBackend._wait_for_table_statusc C s | j | jj| jj|iid�S )z0Construct the item retrieval request parameters.)rW �Key)r% r6 r r �r< �keyrK rK rL �_prepare_get_request� s ���z$DynamoDBBackend._prepare_get_requestc C s~ t � }| j| jj| jj|i| jj| jj|i| jj| jjt|�iid�}| �� rz|d � | j
j| j
jtt || j ��ii� |S )z/Construct the item creation request parameters.�rW �Itemr� )
r r% r6 r r r7 r8 �strrR rO r} r0 r3 )r< r� �valuer Z put_requestrK rK rL �_prepare_put_request� s* �� �����z$DynamoDBBackend._prepare_put_request)r� �returnc C sF t � }| j| jj| jj|i| jj| jjdi| jj| jjt|�iid�S )z7Construct the counter initialization request parameters�0r� )r r% r6 r r � _count_filedr8 r� )r< r� r rK rK rL �_prepare_init_count_request� s �� ���z+DynamoDBBackend._prepare_init_count_requestc C s@ | j | jj| jj|iid| jj� d| jj� d�dddiidd�S ) z2Construct the counter increment request parameterszset z = z + :numz:numr �1Z UPDATED_NEW)rW r� ZUpdateExpressionZExpressionAttributeValuesZ ReturnValues)r% r6 r r r� r� rK rK rL �_prepare_inc_count_request� s ����z*DynamoDBBackend._prepare_inc_count_requestc s d� vr i S � fdd�| j D �S )z1Convert get_item() response to field-value pairs.r� c s$ i | ] }|j � d |j |j �qS )r� r )�.0�field�� raw_responserK rL �
<dictcomp> s �z1DynamoDBBackend._item_to_dict.<locals>.<dictcomp>)r9 )r< r� rK r� rL �
_item_to_dict� s
�z DynamoDBBackend._item_to_dictc C s | � � S �N)r; rX rK rK rL rP s zDynamoDBBackend.clientc C s< t |�}| �|�}| jjf i |��}| �|�}|�| jj�S r� )r� r� rP Zget_itemr� r/ r7 r )r< r� �request_parameters�
item_response�itemrK rK rL r/ s
zDynamoDBBackend.getc C s* t |�}| �||�}| jjf i |�� d S r� )r� r� rP �put_item)r< r� r� r� rK rK rL �set s zDynamoDBBackend.setc s � fdd�|D �S )Nc s g | ]}� � |��qS rK )r/ )r� r� rX rK rL �
<listcomp> � z(DynamoDBBackend.mget.<locals>.<listcomp>rK )r< �keysrK rX rL �mget s zDynamoDBBackend.mgetc C s( t |�}| �|�}| jjf i |�� d S r� )r� r� rP Z delete_item)r< r� r� rK rK rL �delete s
zDynamoDBBackend.deletec C sD t |�}| �|�}| jjf i |��}|d | jj | jj }t|�S )z<Atomically increase the chord_count and return the new countZ
Attributes)r� r� rP Z update_itemr� r r r0 )r< r� r� r� Z new_countrK rK rL �incr s
zDynamoDBBackend.incrc sD | � |d �}| �t|��}| jjf i |�� t� j||fi |��S )Nr )Zget_key_for_chordr� r� rP r� r" �_apply_chord_incr)r< Zheader_result_args�bodyr> Z chord_keyZinit_count_requestrI rK rL r� ' s ��z!DynamoDBBackend._apply_chord_incr)NN)NN)r^ )2�__name__�
__module__� __qualname__�__doc__r% r1 r2 r* r) r3 Zsupports_autoexpirer r6 r7 r� r8 r} r9 Zimplements_incrr# � staticmethodr( r; rY rQ rR rS rp rv rT rb r� r� r� r r r� r� r� �propertyrP r/ r� r� r� �bytesr0 r� r� �
__classcell__rK rK rI rL r sP Y
p
r ) r� � collectionsr � ipaddressr r r �typingr r Zkombu.utils.urlr r' Zcelery.exceptionsr Zcelery.utils.logr
�baser r& Zbotocore.exceptionsr
� ImportError�__all__r r� r+ r rK rK rK rL �<module> s"