Question: How do you use @effect.result with asyncio?

[ShravanSunder]

i'm unsure of how to use async functions and @effect.result. The below results in tons of typehints

  @effect.result[bool, LunaException]()
  async def create_collection(self, params: CreateCollectionParams):
    # Create a collection in Qdrant
    result = await self.connection.ok.create_collection(
      collection_name=params.collection_name, vectors_config=params.dense_vector_params, sparse_vectors_config=params.sparse_vector_params
    )
    return Ok(result)

Argument of type "(self: Self@VectordbClient, params: CreateCollectionParams) -> Coroutine[Any, Any, Result[bool, Any]]" cannot be assigned to parameter "fn" of type "(**_P@__call__) -> (Generator[bool | None, bool, bool | None] | Generator[bool | None, None, bool | None])" in function "__call__"
  Type "(self: Self@VectordbClient, params: CreateCollectionParams) -> Coroutine[Any, Any, Result[bool, Any]]" is incompatible with type "(**_P@__call__) -> (Generator[bool | None, bool, bool | None] | Generator[bool | None, None, bool | None])"
    Function return type "Coroutine[Any, Any, Result[bool, Any]]" is incompatible with type "Generator[bool | None, bool, bool | None] | Generator[bool | None, None, bool | None]"

[dbrattli]

For this to work we would need a separate effect.async_result.

[ShravanSunder]

@dbrattli is it entail just copying

Expression/expression/effect/result.py

Line 15 in 5b043db

class ResultBuilder(Builder[_TSource, Result[Any, _TError]]):

with async signatures?

[ShravanSunder]

i've started using my own decorator for results, here it is below for refrence. It wraps async or sync functions to return a Result[T, Exception]. It will ensure the return type is not double wrapped and it retuns the correct typing.

Unlike catch it will also properly type inputs of the wrapped fn

The LunaException and category is specific for my usecase and can just be replaced with any othter exception or TErr

import inspect
import typing as t
from functools import wraps

from expression import Error, Ok, Result
from shared.common.models.luna_exception import LunaException
from shared.common.models.luna_exception_category import LunaExceptionCategory

TInputs = t.ParamSpec("TInputs")
TOutputs = t.TypeVar("TOutputs")


def as_result(fallback: LunaExceptionCategory | LunaException = LunaExceptionCategory.wrapped_exception):
  """Decorator to wrap a function in a Result.
  Note: fallback_exception takes precedence over fallback_category.

  Returns:
    A function that returns a Result[TOutputs, LunaException]
  """

  def translate_result(output: TOutputs) -> Result[TOutputs, LunaException]:
    if isinstance(output, LunaException):
      return Error(output)
    elif isinstance(output, Result):
      return output
    elif isinstance(output, Exception):
      raise output
    else:
      return Ok(output)

  @t.overload
  def decorator(
    func: t.Callable[TInputs, Result[TOutputs, LunaException]],
  ) -> t.Callable[TInputs, Result[TOutputs, LunaException]]: ...

  @t.overload
  def decorator(  # type: ignore
    func: t.Callable[TInputs, t.Coroutine[t.Any, t.Any, Result[TOutputs, LunaException]]],
  ) -> t.Callable[TInputs, t.Coroutine[t.Any, t.Any, Result[TOutputs, LunaException]]]: ...

  @t.overload
  def decorator(
    func: t.Callable[TInputs, t.Coroutine[t.Any, t.Any, TOutputs]],
  ) -> t.Callable[TInputs, t.Coroutine[t.Any, t.Any, Result[TOutputs, LunaException]]]: ...

  @t.overload
  def decorator(func: t.Callable[TInputs, TOutputs]) -> t.Callable[TInputs, Result[TOutputs, LunaException]]: ...

  def decorator(
    func: t.Callable[TInputs, TOutputs]
    | t.Callable[TInputs, t.Coroutine[t.Any, t.Any, TOutputs]]
    | t.Callable[TInputs, Result[TOutputs, LunaException]]
    | t.Callable[TInputs, t.Coroutine[t.Any, t.Any, Result[TOutputs, LunaException]]],
  ) -> t.Callable[TInputs, Result[TOutputs, LunaException]] | t.Callable[TInputs, t.Coroutine[t.Any, t.Any, Result[TOutputs, LunaException]]]:
    if inspect.iscoroutinefunction(func):

      @wraps(func)
      async def async_wrapper(*args: TInputs.args, **kwargs: TInputs.kwargs) -> Result[TOutputs, LunaException]:
        try:
          result = t.cast(TOutputs, await func(*args, **kwargs))
          return translate_result(result)
        except LunaException as e:
          return Error(e)
        except Exception as e:
          if isinstance(fallback, LunaException):
            fallback.set_cause(e)
            return Error(fallback)
          return Error(LunaException(fallback, cause=e))

      return async_wrapper
    else:

      @wraps(func)
      def sync_wrapper(*args: TInputs.args, **kwargs: TInputs.kwargs) -> Result[TOutputs, LunaException]:
        try:
          result = t.cast(TOutputs, func(*args, **kwargs))
          return translate_result(result)
        except LunaException as e:
          return Error(e)
        except Exception as e:
          if isinstance(fallback, LunaException):
            fallback.set_cause(e)
            return Error(fallback)
          return Error(LunaException(fallback, cause=e))

      return sync_wrapper

  return decorator

[ShravanSunder]

@dbrattli if you'd like i can contribute a generic version of the above to the repo