Add Middleware Support for OperationHandlers

[VegetarianOrc]

Description

This take on OperationHandler middleware leverages the fact that Handler uses an executor to invoke operation handler methods to simplify interceptor implementation. Using the executor to run sync handlers effectively narrows the return type at that point to Awaitable[StartOperationSync[OutputT] | StartOperationResultAsync]. Using that we can define an AwaitableOperationHandler and remove the complexity of sync vs async from interceptors.

Change Summary

• Updated OperationHandler.start to include the return type Awaitable[StartOperationResultSync[OutputT] | StartOperationResultAsync] to enable composing OperationHandlers.
• Add AwaitableOperationHandler to use when calling operation handler methods and in interceptors.
• Add OperationHandlerInterceptor
• Update Handler to apply any interceptors when looking up the operation handler by service & operation name.
• Add tests to validate both sync and async operation handlers can be intercepted and that interceptors are applied in the order they are provided.

closes #32

[cretz]

LGTM except for some naming things. Ideally we wouldn't add a layer of indirection and stack trace entry for every call just to support middleware, but not a big deal. Would also wait to merge this until the Temporal SDK PR is at least present to confirm the abstraction works as expected.

[cretz]

So there is a bit of a maybe-rare use case here that may struggle - accessing the underlying handler that the user themselves created. Some other SDKs' middleware is given the actual handler. I think we could consider having AwaitableOperationHandler have a def underlying(self) -> Optional[OperationHandler[Any, Any]], and we should make _EnsuredAwaitableOperationHandler return the underlying operation handler. Whether we 1) make the method optional to implement (I think we should consider making it required), and 2) return optional or non-optional (can be self in the rare case they wanted to hide the underlying) can both be subject to discussion.

But ok if we decide not to do anything here too.

[VegetarianOrc]

I'm not against adding a def underlying(self) method, but I think it's a bit confusing for an implementor and users. Is it intended to always return the root level OperationHandler, or just one level below it in the middleware stack? Would the caller of underlying() be expected to call underlying on each layer until it finally gets to the instance it expects?

Do you have a use case in mind that this would be useful for? My current feeling is that if the interceptor relies on the details of a specific operation handler then there is hopefully a better way to do it than that.

It would be nice to provide the actual handler in some ways, but it seems worth it to reduce the surface area of the sync/async handling. If we gave the operation handler directly we would need a way to provide the executor so that middleware implementations could submit the sync work to that and passing it as another arg to intercept felt awkward to me. I also didn't love the idea of every implementation needing to do the async callable check that happens in _EnsuredAwaitableOperationHandler.

[cretz]

Is it intended to always return the root level OperationHandler, or just one level below it in the middleware stack?

Just one below and if all do that, then it is the root one

Do you have a use case in mind that this would be useful for? My current feeling is that if the interceptor relies on the details of a specific operation handler then there is hopefully a better way to do it than that.

Not specifically at this time, but we learned with workflows that people want access to the instance of those in interceptors, so we added Workflow.Instance property.

I agree overall, we can punt on this and if there is value in having the non-middleware-defined OperationHandler instance, maybe we can expose it another way at that time (e.g. via contextvars or something).

[cretz]

This LGTM. Can you do me a favor before merging and confirm that latest main Temporal Python SDK completely passes tests with the Nexus dependency changed to point to this branch but no other changes? Just want to confirm there are no surprises.