o
    iU                     @  sJ  d Z ddlmZ ddlZddlZddlZddlmZ ddl	m
Z
mZmZmZmZmZmZmZ ddlZddlZddlmZ ddlmZmZmZ ddlZddlZddlZddlZddlZddl Zddl!Zddl"Zddl#Zddl$Zddl%m&Z&m'Z' dd	l$m(Z(m)Z) dd
l*m+Z+m,Z, ddl-m.Z. dZ/G dd dZ0eG dd dZ1dddZ2dS )zNexus worker    )annotationsN)	dataclass)AnyCallableMappingNoReturnOptionalSequenceTypeUnion)	LazyValue)CancelOperationContextHandlerStartOperationContext)ApplicationErrorWorkflowAlreadyStartedError)Infologger)RPCErrorRPCStatusCode   )Interceptorztemporal.api.failure.v1.Failurec                   @  sp   e Z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	d?d-d.Z
d@d2d3ZdAd7d8Zd9S )B_NexusWorkerbridge_worker-Callable[[], temporalio.bridge.worker.Worker]clienttemporalio.client.Client
task_queuestrservice_handlersSequence[Any]data_converter"temporalio.converter.DataConverterinterceptorsSequence[Interceptor]metric_metertemporalio.common.MetricMeterexecutor%Optional[concurrent.futures.Executor]returnNonec          	      C  sD   || _ || _|| _t||| _|| _|| _|| _i | _t	
 | _d S N)_bridge_worker_client_task_queuer   _handler_data_converter_interceptors_metric_meter_running_tasksasyncioQueue_fail_worker_exception_queue)	selfr   r   r   r   r!   r#   r%   r'    r8   ^/var/www/html/karishye-ai-python/venv/lib/python3.10/site-packages/temporalio/worker/_nexus.py__init__-   s   z_NexusWorker.__init__c              
     s  d fdd}t | }	 zt    }t j||gt jdI dH  | r3|  |I dH  |I dH }|dr|j	}|j
d	r\t  |j|j
jt|j
j j|j< nL|j
d
rxt  |j|j
jt|j
j j|j< n0td|j
 |dr j|jj }r|  ntd|jj   ntd| W n! tjjjy   |  Y dS  ty } ztd|d}~ww q)z:Continually poll for Nexus tasks and dispatch to handlers.r)   r   c                     s    j  I d H r+   )r6   getr8   r7   r8   r9   raise_from_exception_queueJ   s   z4_NexusWorker.run.<locals>.raise_from_exception_queueT)return_whenNtaskstart_operationcancel_operationzInvalid Nexus task request: cancel_taskz@Received cancel_task but no running task exists for task token: zInvalid Nexus task: zNexus worker failed)r)   r   )r4   create_taskr,   poll_nexus_taskwaitFIRST_COMPLETEDdonecancelHasFieldr?   request_handle_start_operation_task
task_tokenr@   dictheaderr3   _handle_cancel_operation_taskrA   NotImplementedErrorr;   rB   r   debugdecode
temporaliobridgeworkerPollShutdownError	ExceptionRuntimeError)r7   r=   exception_task	poll_task
nexus_taskr?   running_taskerrr8   r<   r9   runG   sl   










z_NexusWorker.runc                   sl   	 z%|    I d H }tjjjj|jjd}d|j	j
_|   |I d H  W n tjjjy4   Y d S w q)NT)rL   zWorker shutting down)r,   rD   rS   rT   protonexusNexusTaskCompletionr?   rL   errorfailuremessagecomplete_nexus_taskrU   rV   )r7   r?   
completionr8   r8   r9   drain_poll_queue   s   

z_NexusWorker.drain_poll_queuec                   s"   t j| j ddiI d H  d S )Nreturn_exceptionsT)r4   gatherr3   valuesr<   r8   r8   r9   wait_all_completed   s    z_NexusWorker.wait_all_completedrL   bytesrJ   .temporalio.api.nexus.v1.CancelOperationRequestheadersMapping[str, str]c                   sb  t |j|j|d}tjjj fdd| jd  zzYz j	
||jI dH  W n) tyS } ztd tjjjj| t|I dH d}W Y d}~nd}~ww tjjjj|tjjjjtjjj dd	}  |I dH  W n ty   td
 Y nw W z j|= W dS  ty   td Y dS w z j|= W w  ty   td Y w w )zHandle a cancel operation task.

        Attempt to execute the user cancel_operation method. Handle errors and send the
        task completion.
        )service	operationrn   c                        t  jdS N)r   r   r.   r8   r<   r8   r9   <lambda>       z<_NexusWorker._handle_cancel_operation_task.<locals>.<lambda>)infonexus_contextr   Nz/Failed to execute Nexus cancel operation methodrL   rb   )rA   rL   	completed$Failed to send Nexus task completionz:Failed to remove task for completed Nexus cancel operation)r   rp   rq   rS   r`   _operation_context_TemporalCancelOperationContextr-   setr/   rA   operation_tokenBaseExceptionr   warningrT   r_   ra   _handler_error_to_proto_exception_to_handler_errorapiv1ResponseCancelOperationResponser,   re   rW   	exceptionr3   KeyError)r7   rL   rJ   rn   ctxr]   rf   r8   r<   r9   rO      sd   



	

z*_NexusWorker._handle_cancel_operation_taskstart_request-temporalio.api.nexus.v1.StartOperationRequestc                   s:  zz_z|  ||I dH }W n6 tyD } z*td tjjjj|| 	t
|I dH d}t|tjjr:| j| W Y d}~nd}~ww tjjjj|tjjjj|dd}|  |I dH  W n tyo   td Y nw W z| j|= W dS  ty   td Y dS w z| j|= W w  ty   td Y w w )zHandle a start operation task.

        Attempt to execute the user start_operation method and invoke the data converter
        on the result. Handle errors and send the task completion.
        Nz.Failed to execute Nexus start operation methodry   )r@   rz   r|   z9Failed to remove task for completed Nexus start operation)_start_operationr   r   r   rS   rT   r_   r`   ra   r   r   
isinstance
concurrentfuturesBrokenExecutorr6   
put_nowaitr   r   r   r,   re   rW   r   r3   r   )r7   rL   r   rn   start_responser]   rf   r8   r8   r9   rK      sR   




z)_NexusWorker._handle_start_operation_task.temporalio.api.nexus.v1.StartOperationResponsec           	   
     sf  t |j|j||j|jdd |jD t|jd}tj	j
j| j fddd  tt j|jdi dd	}zY j||I dH }d
d |jD }t|tjjretjj	jjtjj	jjj|j|ddW S t|tjjr j|j gI dH \}tjj	jjtjj	jjj!||ddW S t"t#d tj$y } ztjj	jj %|I dH dW  Y d}~S d}~ww )a  Invoke the Nexus handler's start_operation method and construct the StartOperationResponse.

        OperationError is handled by this function, since it results in a StartOperationResponse.

        All other exceptions are handled by a caller of this function.
        c                 S  s   g | ]}t j|j|jd qS )urltype)nexusrpcLinkr   r   .0linkr8   r8   r9   
<listcomp>
  s    z1_NexusWorker._start_operation.<locals>.<listcomp>)rp   rq   rn   
request_idcallback_urlinbound_linkscallback_headersc                     rr   rs   rt   r8   r<   r8   r9   ru     rv   z/_NexusWorker._start_operation.<locals>.<lambda>)rx   r   rw   )r!   payloadN)
serializerrn   streamc                 S  s$   g | ]}t jjjj|j|jd qS r   )rS   r   r`   r   r   r   r   r   r8   r8   r9   r     s    )r   links)async_success)r   r   )sync_successzOperation start method must return either nexusrpc.handler.StartOperationResultSync or nexusrpc.handler.StartOperationResultAsync.)operation_error)&r   rp   rq   r   callbackr   rM   callback_headerrS   r`   r}   _TemporalStartOperationContextr-   r   r   _DummyPayloadSerializerr0   r   r/   r@   outbound_linksr   r   handlerStartOperationResultAsyncr   r   StartOperationResponseAsynctokenStartOperationResultSyncencodevalueSyncr   	TypeErrorOperationError_operation_error_to_proto)	r7   r   rn   r   inputresultr   r   r]   r8   r<   r9   r      sr   



z_NexusWorker._start_operationrb   5Union[nexusrpc.HandlerError, nexusrpc.OperationError]temporalio.api.nexus.v1.Failurec                   s   |j  }rIz4tjjj }| j||I dH  tj	j
|}tjjjj|dt|dtitj|ddddW S  tyH   td Y nw tjjjjt|i d	dS )
a  Serialize ``error`` as a Nexus Failure proto.

        The Nexus Failure represents the top-level error. If there is a cause chain
        attached to the exception, then serialize it as the ``details``.

        Notice that any stack trace attached to ``error`` itself is not included in the
        result.

        See https://github.com/nexus-rpc/api/blob/main/SPEC.md#failure
        Nrd   r   ),:)
separatorszutf-8)rd   metadatadetailsz2Failed to serialize cause chain of nexus exception    )	__cause__rS   r   rc   r   Failurer0   encode_failuregoogleprotobufjson_formatMessageToDictr`   popr   _TEMPORAL_FAILURE_PROTO_TYPEjsondumpsr   r   r   r   )r7   rb   causerc   failure_dictr8   r8   r9   #_nexus_error_to_nexus_failure_proto?  s.   


z0_NexusWorker._nexus_error_to_nexus_failure_protor]   nexusrpc.OperationError2temporalio.api.nexus.v1.UnsuccessfulOperationErrorc                   s&   t jjjj|jj| |I d H dS )N)operation_staterc   )rS   r   r`   r   UnsuccessfulOperationErrorstater   r   )r7   r]   r8   r8   r9   r   m  s
   
z&_NexusWorker._operation_error_to_protohandler_errornexusrpc.HandlerError$temporalio.api.nexus.v1.HandlerErrorc                   sf   |j du rtjjjjjn|j du rtjjjjjntjjjjj}tjj	jj
|jj| |I dH |dS )z:Serialize ``handler_error`` as a Nexus HandlerError proto.TFN)
error_typerc   retry_behavior)retryable_overriderS   r   enumsr   NexusHandlerErrorRetryBehavior,NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE0NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_UNSPECIFIEDr`   HandlerErrorr   r   r   )r7   r   r   r8   r8   r9   r   v  s   


z$_NexusWorker._handler_error_to_protoN)r   r   r   r   r   r   r   r    r!   r"   r#   r$   r%   r&   r'   r(   r)   r*   )r)   r*   )rL   rl   rJ   rm   rn   ro   r)   r*   )rL   rl   r   r   rn   ro   r)   r*   )r   r   rn   ro   r)   r   )rb   r   r)   r   )r]   r   r)   r   )r   r   r)   r   )__name__
__module____qualname__r:   r^   rg   rk   rO   rK   r   r   r   r   r8   r8   r8   r9   r   ,   s    


?


4
*
F
.	r   c                   @  s6   e Zd ZU ded< ded< dd	d
Z	ddddZdS )r   r"   r!   z temporalio.api.common.v1.Payloadr   r   r   r)   nexusrpc.Contentc                   s
   t d)Nz>The serialize method of the Serializer is not used by handlers)rP   )r7   r   r8   r8   r9   	serialize  s   z!_DummyPayloadSerializer.serializeNcontentas_typeOptional[Type[Any]]c              
     s^   z| j j| jg|r|gnd dI d H \}|W S  ty. } ztjdtjjdd|d }~ww )N)
type_hintsz5Data converter failed to decode Nexus operation inputFr   r   )r!   rR   r   rW   r   r   HandlerErrorTypeBAD_REQUEST)r7   r   r   r   r]   r8   r8   r9   deserialize  s"   z#_DummyPayloadSerializer.deserialize)r   r   r)   r   r+   )r   r   r   r   r)   r   )r   r   r   __annotations__r   r   r8   r8   r8   r9   r     s   
 
r   r]   r   r)   r   c                 C  s  t | tjr| }tt||j d} |j| _|j| _nt | tr.tj| jtj	j
| j d}nt | tr?tj| jtj	j
dd}nt | tr| jtjkrUtj| jtj	jd}n| jtjtjtjfv rltj| jtj	j
dd}n| jtjtjfv rtj| jtj	jd}ny| jtjtjtj
tjtjtjfv rtj| jtj	j
d}n]| jtjkrtj| jtj	jd}nL| jtjkrtj| jtj	jd}n;| jtjkrtj| jtj	jd}n*| jtj krtj| jtj	j!d}ntjd| j tj	j
d}ntjt| tj	j
d}| |_|S )N)rd   non_retryabler   F)r   zUnhandled RPC error status: )"r   r   r   r   r   	retryable__traceback__r   rd   r   INTERNALr   r   r   statusr   INVALID_ARGUMENTr   ALREADY_EXISTSFAILED_PRECONDITIONOUT_OF_RANGEABORTEDUNAVAILABLE	CANCELLED	DATA_LOSSUNKNOWNUNAUTHENTICATEDPERMISSION_DENIED	NOT_FOUNDRESOURCE_EXHAUSTEDUNIMPLEMENTEDNOT_IMPLEMENTEDDEADLINE_EXCEEDEDUPSTREAM_TIMEOUT)r]   handler_errr8   r8   r9   r     s   	






r   )r]   r   r)   r   )3__doc__
__future__r   r4   concurrent.futuresr   r   dataclassesr   typingr   r   r   r   r   r	   r
   r   google.protobuf.json_formatr   nexusrpc.handlerr   r   r   r   r   temporalio.api.common.v1rS   temporalio.api.enums.v1temporalio.api.failure.v1temporalio.api.nexus.v1temporalio.bridge.proto.nexustemporalio.bridge.workertemporalio.clienttemporalio.commontemporalio.convertertemporalio.nexustemporalio.exceptionsr   r   r   r   temporalio.servicer   r   _interceptorr   r   r   r   r   r8   r8   r8   r9   <module>   s@    (  ^