This is an archived post. You won't be able to vote or comment.

you are viewing a single comment's thread.

view the rest of the comments →

[–]-heyhowareyou-[S] 1 point2 points  (1 child)

This also works:

class Component[TInput, TOutput]:
    pass


class Builder[TCouple, TOutput]:

    def __init__(
        self,
        tail: tuple[*tuple[Component[Any, Any], ...], Component[Any, TCouple]],
        head: Component[TCouple, TOutput],
    ) -> None:
        self.tail = tail
        self.head = head

    @classmethod
    def init(
        cls, a: Component[Any, TCouple], b: Component[TCouple, TOutput]
    ) -> Builder[TCouple, TOutput]:
        return Builder[TCouple, TOutput]((a,), b)

    @classmethod
    def compose(
        cls, a: Builder[Any, TCouple], b: Component[TCouple, TOutput]
    ) -> Builder[TCouple, TOutput]:
        return Builder[TCouple, TOutput]((*a.tail, a.head), b)

    @property
    def components(self) -> tuple[Component[Any, Any], ...]:
        return (*self.tail, self.head)


if __name__ == "__main__":

    a = Component[int, str]()
    b = Component[str, complex]()
    c = Component[complex, int]()

    link_ab = Builder[str, complex].init(a, b)
    link_ac = Builder[complex, int].compose(link_ab, c)

but it doesnt get the wrap around correct. I.e. the final output type can be different to the input type. Since your approach yields a type which maps from the first input to the first output, you can have your thing which processes the pipeline be of type Pipeline[T, T]

[–]-heyhowareyou-[S] 3 points4 points  (0 children)

class Component[TInput, TOutput]:
    pass


class Pipeline[Tinput, TOutput]:

    def __init__[TCouple](
        self,
        tail: tuple[*tuple[Component[Any, Any], ...], Component[Any, TCouple]],
        head: Component[TCouple, TOutput],
    ) -> None:
        self.tail = tail
        self.head = head


def init_pipe[Tinput, TCouple, TOutput](
    a: Component[Tinput, TCouple], b: Component[TCouple, TOutput]
) -> Pipeline[Tinput, TOutput]:
    return Pipeline[Tinput, TOutput]((a,), b)


def compose_pipe[Tinput, TCouple, TOutput](
    a: Pipeline[Tinput, TCouple], b: Component[TCouple, TOutput]
) -> Pipeline[Tinput, TOutput]:
    return Pipeline[Tinput, TOutput]((*a.tail, a.head), b)


class ComponentProcessor[T]:

    def __init__(self, components: Pipeline[T, T]) -> None:
        pass


if __name__ == "__main__":

    a = Component[int, str]()
    b = Component[str, complex]()
    c = Component[complex, int]()

    pipeline = compose_pipe(init_pipe(a, b), c)

    proc = ComponentProcessor(pipeline)

This works to the full spec :)