U
    bhC,                     @  s   d Z ddlmZ ddlZddlmZ ddlmZ ddlm	Z	 ddl
mZ dd	lmZ dd
lmZ ddlmZ dZG dd de	Zdd Zdd ZdS )zAmazon SQS Connection.    )annotationsN)
Serializer)	transform)AsyncAWSQueryConnection)
AWSRequest   )boto3)AsyncMessage)
AsyncQueue)AsyncSQSConnectionc                      s   e Zd ZdZd6 fdd	Zdd Zdd	 Zd7d
dZd8ddZd9ddZ	dd Z
d:ddZd;ddZd<ddZd=ddZd>ddZd?dd Zd@d!d"ZdAd#d$ZdBd%d&ZdCd'd(ZdDd*d+ZdEd,d-ZeZd.d/ ZdFd0d1ZdGd2d3ZdHd4d5Z  ZS )Ir   zAsync SQS Connection.r   Nc                   sB   t d krtdt j|f||d| |d k	r6|ndg| _d S )Nzboto3 is not installed)Zregion_namedebugZApproximateReceiveCount)r   ImportErrorsuper__init__fetch_message_attributes)selfsqs_connectionr   Zregionr   kwargs	__class__ I/tmp/pip-unpacked-wheel-earovzxb/kombu/asynchronous/aws/sqs/connection.pyr      s     zAsyncSQSConnection.__init__c                 C  s^   |  }|r||d< d|i}i }| dkr4d|i}| dkrHd|d< tf |||d|S )	NActiondatagetparamspostz0application/x-www-form-urlencoded; charset=utf-8Content-Type)methodurlheaders)copylowerr   )r   	operationr   	queue_urlr   param_payloadr    r   r   r   _create_query_request$   s    z(AsyncSQSConnection._create_query_requestc                 C  s   |  }||d< | jjj}||}| jjj}i }|jd }d| }	|	|d< d|jd |j	}
|
|d< t
| |d}|jd	tj}tf ||d
|S )NQueueUrlZjsonVersionzapplication/x-amz-json-r   z{}.{}ZtargetPrefixzX-Amz-Target)r   r    r   )r   r   )r!   r   metaservice_modeloperation_modelZ	_endpointhostmetadataformatnamejsondumpsencodehttpr   r   ZDEFAULT_METHODr   )r   r#   r   r$   r)   r*   r   r    Zjson_versioncontent_typetargetr%   r   r   r   r   _create_json_request5   s0    




z'AsyncSQSConnection._create_json_requestc                 C  s   | j j}| j jj}|j}	|pi ||	i }
|	dkrF| ||
||}n(|	dkr^| ||
|}ntd|	 d|j	
 dkrdnd}|j|||d | }| j||d	S )
a  Override make_request to support different protocols.

        botocore has changed the default protocol of communicating
        with SQS backend from 'query' to 'json', so we need a special
        implementation of make_request for SQS. More information on this can
        be found in: https://github.com/celery/kombu/pull/1807.

        protocol_params: Optional[dict[str, dict]] of per-protocol additional parameters.
            Supported for the SQS query to json protocol transition.
        queryr/   zUnsupported protocol: .r   zpresign-urlstandard)signing_typecallback)r   Z_request_signerr(   r)   protocolr   r&   r5   	Exceptionr   r"   signprepareZ_mexe)r   Zoperation_namer   r$   verbr;   protocol_paramsZsignerr)   r<   
all_paramsrequestr9   prepared_requestr   r   r   make_requestW   s0    
     zAsyncSQSConnection.make_requestc                 C  s*   d|i}|rt |d|d< | jd||dS )N	QueueNamedZDefaultVisibilityTimeoutZCreateQueuer:   )r-   
get_object)r   
queue_namevisibility_timeoutr;   r   r   r   r   create_queuey   s     zAsyncSQSConnection.create_queueFc                 C  s   | j dd |j|dS )NZDeleteQueuer:   
get_statusid)r   queueZforce_deletionr;   r   r   r   delete_queue   s    zAsyncSQSConnection.delete_queuec                 C  s   | j j|d}|d S )N)rF   r'   )r   get_queue_url)r   rO   resr   r   r   rQ      s    z AsyncSQSConnection.get_queue_urlAllc                 C  s   | j dd|i|j|dS )NZGetQueueAttributesAttributeNamer:   )rH   rN   )r   rO   	attributer;   r   r   r   get_queue_attributes   s      z'AsyncSQSConnection.get_queue_attributesc              	   C  s*   | j di |j|d||ii||dddS )NZSetQueueAttributeZ
Attributes)zAttribute.NamezAttribute.Valuer/   r6   )r;   rA   rL   )r   rO   rU   valuer;   r   r   r   set_queue_attribute   s     
z&AsyncSQSConnection.set_queue_attributer   c              	   C  s   d|i}i i d}	|d k	r|n| j }
|r0||d< |
rd|	d dt|
i |	d tdt|
i |d k	rt||d< | jd	|d
tfg||||	dS )NZMaxNumberOfMessages)r6   r/   VisibilityTimeoutr/   ZAttributeNamesr6   rT   ZWaitTimeSecondsZReceiveMessageMessage)r;   parentrA   )r   updatelist_query_object_encodeget_listr	   )r   rO   r$   Znumber_messagesrJ   
attributesZwait_time_secondsr;   r   Zproto_paramsattrsr   r   r   receive_message   s&    
    z"AsyncSQSConnection.receive_messagec                 C  s   |  |||S N)delete_message_from_handler   rO   receipt_handler;   r   r   r   delete_message   s
      z!AsyncSQSConnection.delete_messagec                 C  sB   ddd |D it ddd |D id}| jdi |jd||d	S )
NEntriesc                 S  s   g | ]}|j |jd qS )IdReceiptHandlerN   rg   .0mr   r   r   
<listcomp>   s     z;AsyncSQSConnection.delete_message_batch.<locals>.<listcomp>ZDeleteMessageBatchRequestEntryc                 S  s   g | ]}|j |jd qS rj   rm   rn   r   r   r   rq      s   rW   ZDeleteMessageBatchPOSTr@   r;   rA   r_   rH   rN   )r   rO   messagesr;   p_paramsr   r   r   delete_message_batch   s$          z'AsyncSQSConnection.delete_message_batchc                 C  s   | j dd|i||dS )NZDeleteMessagerl   r:   )rM   rf   r   r   r   re      s      z-AsyncSQSConnection.delete_message_from_handlec                 C  s.   d|i}|rt ||d< | jd||jd|dS )NZMessageBodyZDelaySecondsZSendMessagerr   r@   r;   )intrH   rN   )r   rO   Zmessage_contentZdelay_secondsr;   r   r   r   r   send_message   s       zAsyncSQSConnection.send_messagec              
   C  sn   i }t |D ]J\}}d|d  }|| d|d | d|d | d|d i q| jd||jd	|d
S )NzSendMessageBatchRequestEntry.r   z.Idr   z.MessageBodyz.DelaySeconds   ZSendMessageBatchrr   rx   )	enumerater]   rH   rN   )r   rO   ru   r;   r   imsgprefixr   r   r   send_message_batch   s$          z%AsyncSQSConnection.send_message_batchc                 C  s   | j d||d|j|dS )NZChangeMessageVisibility)rl   rZ   r:   rL   )r   rO   rg   rJ   r;   r   r   r   change_message_visibility   s     z,AsyncSQSConnection.change_message_visibilityc                 C  s<   dd |D }d|it d|id}| jdi |jd||dS )	Nc                 S  s(   g | ] }|d  j |d  j|d dqS )r   r   )rk   rl   rZ   rm   )ro   tr   r   r   rq      s   zFAsyncSQSConnection.change_message_visibility_batch.<locals>.<listcomp>ri   Z(ChangeMessageVisibilityBatchRequestEntryrW   ZChangeMessageVisibilityBatchrr   rs   rt   )r   rO   ru   r;   entriesrv   r   r   r   change_message_visibility_batch   s    
   z2AsyncSQSConnection.change_message_visibility_batch c                 C  s(   i }|r||d< | j d|dtfg|dS )NZQueueNamePrefixZ
ListQueuesr'   r:   )r`   r
   )r   r   r;   r   r   r   r   get_all_queues  s      z!AsyncSQSConnection.get_all_queuesc                 C  s   |  |t| j||S rd   )r   r   _on_queue_ready)r   rI   r;   r   r   r   	get_queue  s    zAsyncSQSConnection.get_queuec                   s   t  fdd|D d S )Nc                 3  s   | ]}|j  r|V  qd S rd   )r   endswith)ro   qr.   r   r   	<genexpr>  s      z5AsyncSQSConnection._on_queue_ready.<locals>.<genexpr>)next)r   r.   Zqueuesr   r   r   r     s     z"AsyncSQSConnection._on_queue_readyc                 C  s   | j dd|jidtfg|dS )NZListDeadLetterSourceQueuesr'   r:   )r`   r   r
   )r   rO   r;   r   r   r   get_dead_letter_source_queues  s     z0AsyncSQSConnection.get_dead_letter_source_queuesc                 C  s   | j d|||d|j|dS )NZAddPermission)LabelZAWSAccountIdZ
ActionNamer:   rL   )r   rO   labelZaws_account_idaction_namer;   r   r   r   add_permission   s     z!AsyncSQSConnection.add_permissionc                 C  s   | j dd|i|j|dS )NZRemovePermissionr   r:   rL   )r   rO   r   r;   r   r   r   remove_permission*  s       z$AsyncSQSConnection.remove_permission)r   NN)NN)NN)FN)rS   N)N)r   NNNN)N)N)N)NN)N)N)N)r   N)N)N)N)N)__name__
__module____qualname____doc__r   r&   r5   rE   rK   rP   rQ   rV   rY   rc   rh   rw   re   rz   r   r   r   r   r   lookupr   r   r   r   __classcell__r   r   r   r   r      sJ   "
#   




       


 
   


 
	

	

 

r   c                 C  s"   i }t |d|  dd | D S )Nr   c                 S  s   i | ]\}}||qS r   r   )ro   kvr   r   r   
<dictcomp>3  s      z(_query_object_encode.<locals>.<dictcomp>)_query_object_encode_partitems)r   r   r   r   r   r_   0  s    r_   c                 C  s   |r| dn|}t |ttfrLt|D ] \}}t| | |d  | q(n>t |tr~| D ]\}}t| | | | q^nt|| |< d S )Nr7   r   )
isinstancer^   tupler|   r   dictr   str)r   r   partZdottedr}   itemkeyrX   r   r   r   r   6  s    
r   )r   
__future__r   r/   Zbotocore.serializer   Zviner   Z!kombu.asynchronous.aws.connectionr   Zkombu.asynchronous.aws.extr   extr   messager	   rO   r
   __all__r   r_   r   r   r   r   r   <module>   s     