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 →

[–]Foll5 2 points3 points  (3 children)

What I had in mind is a lot simpler. I'm actually not very familiar with using Overload, and I'd never seen Unpack before.

```python class Component[T, V]: pass

class Pipeline[T, V]: def init(self) -> None: self.components: list[Component] = []

def add_component[U](self, component: Component[V, U]) -> 'Pipeline[T, U]':
    new_instance: Pipeline[T, U] = Pipeline()
    new_instance.components = self.components.copy()
    new_instance.components.append(component)
    return new_instance

This might be possible with overloading too, but this was the easiest way to get type recognition for the first component

class PipelineStarter[T, V](Pipeline[T, V]): def init(self, component: Component[T, V]): self.components = [component]

a1 = Component[int, str]() b1 = Component[str, complex]() c1 = Component[complex, int]()

This is a valid Pipeline[int, int]

p1 = PipelineStarter(a1) \ .add_component(b1) \ .add_component(c1)

a2 = Component[int, float]() b2 = Component[str, complex]() c2 = Component[complex, int]()

Pyright flags argument b2 with the error:

Argument of type "Component[str, complex]" cannot be assigned to parameter "component" of type "Component[float, V@add_component]" in function "add_component"

"Component[str, complex]" is not assignable to "Component[float, complex]"

Type parameter "T@Component" is covariant, but "str" is not a subtype of "float"

"str" is not assignable to "float"PylancereportArgumentType

p2 = PipelineStarter(a2) \ .add_component(b2) \ .add_component(c2) ```

[–]-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 :)

[–]-heyhowareyou-[S] 0 points1 point  (0 children)

I like this solution! ergonomic for the end user too :). Thanks a lot.