Closed
Description
I'm having some errors with typing, if it's not a returns
problem feel free to close this issue
Mypy errors:
get_stream_details.py:78: error: Argument 1 to "apply" of "Result" has incompatible type "Result[overloaded function, Any]"; expected "Result[Callable[[Stream], StreamDetails], FailureDetails]"
get_stream_details.py:78: error: Argument 1 has incompatible type "Result[StreamDetails, FailureDetails]"; expected "Maybe[Any]"
get_stream_details.py:78: error: Incompatible return value type (got "Maybe[Any]", expected "Result[StreamDetails, FailureDetails]")
get_stream_details.py:79: error: Argument 1 to "_flow" has incompatible type "Result[overloaded function, Any]"; expected "Result[overloaded function, Any]"
get_stream_details.py:79: error: Argument 1 to "from_value" of "Result" has incompatible type overloaded function; expected overloaded function
get_stream_details.py:81: error: Cannot infer type argument 1 of "_bind"
get_stream_details.py:81: error: Argument 1 to "_bind" has incompatible type "Callable[[Any], Result[Any, FailureDetails]]"; expected "Callable[[Any], Maybe[Any]]"
get_stream_details.py:81: error: Incompatible return value type (got "Result[Any, FailureDetails]", expected "Maybe[object]")
get_stream_details.py:81: error: Incompatible return value type (got "Result[Any, FailureDetails]", expected "Maybe[Any]")
get_stream_details.py:81: error: Cannot infer type argument 1 of "Success"
get_stream_details.py:82: error: Cannot infer type argument 1 of "_bind"
get_stream_details.py:82: error: Argument 1 to "_bind" has incompatible type "Callable[[Any], Result[Any, FailureDetails]]"; expected "Callable[[Any], Maybe[Any]]"
get_stream_details.py:82: error: Incompatible return value type (got "Result[Any, FailureDetails]", expected "Maybe[object]")
get_stream_details.py:82: error: Incompatible return value type (got "Result[Any, FailureDetails]", expected "Maybe[Any]")
get_stream_details.py:82: error: Cannot infer type argument 1 of "Success"
All the errors above are from my return statement in GetStreamDetailsUsecase.__call__!
Code:
@dataclass_json
@dataclass
class StreamDetails:
stream: Stream
projects: List[Project]
ksql_stream_detailed: KSQLStreamDetailed
@classmethod
@curry
def build(
cls,
stream: Stream,
projects: List[Project],
ksql_stream_detailed: KSQLStreamDetailed,
) -> "StreamDetails":
return cls(stream, projects, ksql_stream_detailed)
class FindStreamByStreamId(ABC):
@abstractmethod
def __call__(self, stream_id: UUID) -> Result[Maybe[Stream], FailureDetails]:
pass
class FindProjectsByStream(ABC):
@abstractmethod
def __call__(self, stream: Stream) -> Result[List[Project], FailureDetails]:
pass
# "GetStreamByNameUsecase" definition
# class GetStreamByNameUsecase:
# def __call__(
# self, stream_name: str
# ) -> Result[KSQLStreamDetailed, BusinessFailureDetails]:
# pass
class GetStreamDetailsUsecase:
def __init__(
self,
find_stream_by_stream_id: FindStreamByStreamId,
find_projects_by_stream: FindProjectsByStream,
get_stream_by_name: GetStreamByNameUsecase,
):
self.__find_stream_by_stream_id = find_stream_by_stream_id
self.__find_projects_by_stream = find_projects_by_stream
self.__get_stream_by_name = get_stream_by_name
def __call__(self, stream_id: UUID) -> Result[StreamDetails, FailureDetails]:
stream = self.__find_stream_by_stream_id(stream_id).bind(
self.__verify_if_stream_exist
)
partial_projects = partial(stream.bind, self.__find_projects_by_stream)
partial_ksql = partial(
stream.bind, lambda stream_: self.__get_stream_by_name(stream_.name)
)
return flow(
Result.from_value(StreamDetails.build),
stream.apply,
bind(lambda to_apply: partial_projects().apply(Success(to_apply))),
bind(lambda to_apply: partial_ksql().apply(Success(to_apply))),
)
def __verify_if_stream_exist(
self, stream: Maybe[Stream]
) -> Result[Stream, BusinessFailureDetails]:
return maybe_to_result(stream).alt(
lambda __: BusinessFailureDetails(
reason="NOT_FOUND", failure_message="Stream not found"
)
)
But if I change the flow to handmade flow just one error appeared:
Mypy error:
error: Argument 1 to "from_value" of "Result" has incompatible type overloaded function; expected overloaded function
Code:
class GetStreamDetailsUsecase:
# CODE
def __call__(self, stream_id: UUID) -> Result[StreamDetails, FailureDetails]:
stream = self.__find_stream_by_stream_id(stream_id).bind(
self.__verify_if_stream_exist
)
partial_projects = partial(stream.bind, self.__find_projects_by_stream)
partial_ksql = partial(
stream.bind, lambda stream_: self.__get_stream_by_name(stream_.name)
)
details = Result.from_value(StreamDetails.build)
details_added_stream = stream.apply(details)
details_added_projects = details_added_stream.bind(lambda to_apply: partial_projects().apply(Success(to_apply)))
details_added_ksql = details_added_projects.bind(lambda to_apply: partial_ksql().apply(Success(to_apply)))
return details_added_ksql
# CODE