o i0@sddlmZddlZddlmZmZddlmZmZm Z m Z m Z m Z m Z ddlmZmZmZmZddlmZmZddlmZmZmZmZdd lmZmZmZmZm Z m!Z!Gd d d ee eefZ"Gd d d e"eefZ#d!ddZ$d"ddZ%d#dd Z&dS)$) annotationsN)ABCabstractmethod)Any AwaitableCallableGenericOptionalTypeUnion)InputT OperationInfoOutputTServiceHandlerT) OperationServiceDefinition)get_operation_factoryis_async_callable is_callable is_subtype)CancelOperationContextFetchOperationInfoContextFetchOperationResultContextStartOperationContextStartOperationResultAsyncStartOperationResultSyncc@sHeZdZdZeddd ZedddZedddZedddZdS)OperationHandlera Base class for an operation handler in a Nexus service implementation. To define a Nexus operation handler, create a method on your service handler class that takes `self` and returns an instance of :py:class:`OperationHandler`, and apply the :py:func:`@nexusrpc.handler.operation_handler` decorator. To create an operation handler that is limited to returning synchronously, use :py:func:`@nexusrpc.handler.SyncOperationHandler` to create the instance of :py:class:`OperationHandler` from the start method. ctxrinputr returnUnion[StartOperationResultSync[OutputT], Awaitable[StartOperationResultSync[OutputT]], StartOperationResultAsync, Awaitable[StartOperationResultAsync]]cCdS)z Start the operation, completing either synchronously or asynchronously. Returns the result synchronously, or returns an operation token. Which path is taken may be decided at operation handling time. Nselfrrr#r#i/var/www/html/karishye-ai-python/venv/lib/python3.10/site-packages/nexusrpc/handler/_operation_handler.pystart/szOperationHandler.startrtokenstr.Union[OperationInfo, Awaitable[OperationInfo]]cCr")zO Return information about the current status of the operation. Nr#r%rr(r#r#r& fetch_info@zOperationHandler.fetch_infor"Union[OutputT, Awaitable[OutputT]]cCr")z5 Return the result of the operation. Nr#r+r#r#r& fetch_resultIr-zOperationHandler.fetch_resultrUnion[None, Awaitable[None]]cCr")z' Cancel the operation. Nr#r+r#r#r&cancelRr-zOperationHandler.cancelN)rrrr r r!)rrr(r)r r*)rrr(r)r r.)rrr(r)r r0) __name__ __module__ __qualname____doc__rr'r,r/r1r#r#r#r&r"s   rc@sBeZdZdZdddZdd d ZdddZd ddZd!ddZdS)"SyncOperationHandlerz An :py:class:`OperationHandler` that is limited to responding synchronously. This version of the class uses `async def` methods. For the syncio version, see :py:class:`nexusrpc.handler._syncio.SyncOperationHandler`. r'=Callable[[StartOperationContext, InputT], Awaitable[OutputT]]cCsHt|s t|d||_|jr t|jdd}r"|j|_dSdSdS)Nzd is not an `async def` method. SyncOperationHandler must be initialized with an `async def` method. __func__)r RuntimeError_startr5getattrr')r%r' start_funcr#r#r&__init__ds zSyncOperationHandler.__init__rrrr r !StartOperationResultSync[OutputT]cst|||IdHS)a Start the operation and return its final result synchronously. The name 'SyncOperationHandler' means that it responds synchronously in the sense that the start method delivers the final operation result as its return value, rather than returning an operation token representing an in-progress operation. This version of the class uses `async def` methods. For the syncio version, see :py:class:`nexusrpc.handler.syncio.SyncOperationHandler`. N)rr:r$r#r#r&r'qs zSyncOperationHandler.startrr(r)r c td)NzJCannot fetch operation info for an operation that responded synchronously.NotImplementedErrorr+r#r#r&r,zSyncOperationHandler.fetch_inforrcr?)NzECannot fetch the result of an operation that responded synchronously.r@r+r#r#r&r/rBz!SyncOperationHandler.fetch_resultrNonecr?)Nz>An operation that responded synchronously cannot be cancelled.r@r+r#r#r&r1szSyncOperationHandler.cancelN)r'r7)rrrr r r>)rrr(r)r r )rrr(r)r r)rrr(r)r rC) r2r3r4r5r=r'r,r/r1r#r#r#r&r6\s   r6user_service_clsType[ServiceHandlerT]serviceOptional[ServiceDefinition]r Bdict[str, Callable[[ServiceHandlerT], OperationHandler[Any, Any]]]c Csi}|rdd|jDnt}t}t|tD]f\}}t|\}}|rt|tr|j |vr=t d|j d|j d|rj|j |vrjd ddt|D} d |j d |d } | | r_| nd 7} | d 7} t| |j suJd|d|||j <||j q|S)zN Collect operation handler methods from a user service handler class. cSsh|] }|jdur|jqS)N) method_name).0opr#r#r& s  zEcollect_operation_handler_factories_by_method_name.. Operation '' in service 'z' is defined multiple times., css|] }d|dVqdS)'Nr#)rJsr#r#r& szEcollect_operation_handler_factories_by_method_name..zOperation method name 'z' in service handler zv does not match an operation method name in the service definition. Available method names in the service definition: z[none].z5' method name should not be None. This is an SDK bug.) operationsvaluessetinspect getmembersrr isinstancernamer9r2rIjoinsorted TypeErroradd) rDrF factoriesservice_method_namesseen_methodfactoryop_defn_namesmsgr#r#r&2collect_operation_handler_factories_by_method_names:      rhuser_methods_by_method_nameservice_definitionrrCc Cs|}|jD]}|jstd|d|d||jd}|s1td|d|jd|dt|\}}t|t sHtd |d |j d |j |j|j fvretd|jd |d |j d|j d |j dur|j durt |j |j fvr|j |j kst|j |j std|jd |d|j d|j d|j d |jdur|jdurt |j|jfvrt|j|jstd|jd |d|jd|jd|d q |rtd|d|ddt|ddS)aValidate operation handler methods against a service definition. For every operation in ``service_definition``: 1. There must be a method in ``user_methods`` whose method name matches the method name from the service definition. 2. The input and output types of the user method must be such that the user method is a subtype of the operation defined in the service definition, i.e. respecting input type contravariance and output type covariance. rMz' in service definition 'z' does not have a method name. Nz Service 'z4' does not implement an operation with method name 'z0'. But this operation is in service definition 'z'.zMethod 'z ' in class 'z' does not have a valid __nexus_operation__ attribute. Did you forget to decorate the operation method with an operation handler decorator such as :py:func:`@nexusrpc.handler.operation_handler`?rNz ' has name 'z.', but the name in the service definition is 'zZ'. Operation handlers may not override the name of an operation in the service definition.z' has input type 'z0', which is not compatible with the input type 'z' in interface 'z]'. The input type must be the same as or a superclass of the operation definition input type.z' has output type 'z1', which is not compatible with the output type 'z' in interface 'z]'. The output type must be the same as or a subclass of the operation definition output type.z1' implements more operations than the interface 'z'. Extra operations: rOrS)copyrTrUrI ValueErrorpopr]rrYrr2rZ input_typerr output_typer[r\keys)rDrirjrercmethod_op_defnr#r#r&"validate_operation_handler_methodss        rr service_namer) user_methodscCsXi}|D]\}}t|\}}t|ts td|d|d|||j<qt||dS)ah Create a service definition from operation handler factory methods. In general, users should have access to, or define, a service definition, and validate their service handler against it by passing the service definition to the :py:func:`@nexusrpc.handler.service_handler` decorator. This function is used when that is not the case. z In service 'zL', could not locate operation definition for user operation handler method 'z'. Did you forget to decorate the operation method with an operation handler decorator such as :py:func:`@nexusrpc.handler.operation_handler`?)rZrT)itemsrrYrrlrZr)rsrtop_defnsrZrcrbrer#r#r&1service_definition_from_operation_handler_methodss     rw)rDrErFrGr rH)rDrErirHrjrr rC)rsr)rtrHr r)' __future__rrWabcrrtypingrrrrr r r nexusrpc._commonr r rrnexusrpc._servicerrnexusrpc._utilrrrr_commonrrrrrrrr6rhrrrwr#r#r#r&s $   : 7 /V