Coverage for collectra/tasks/text.py: 88.89%

1"""Tasks for combining and transforming text artefacts.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from collectra.tasks.base import Task 

8from collectra.types.texts import Text 

9 

10__all__ = ["ConcatenateText"] 

11 

12 

13class ConcatenateText(Task[Text]): 

14 """Join every input Text artefact into one Text artefact.""" 

15 

16 collective = True 

17 

18 def __init__(self, name: str, separator: str = "\n", **kwargs: Any) -> None: 

19 super().__init__(name, separator=separator, **kwargs) 

20 

21 def input_type(self) -> type: 

22 return Text 

23 

24 def output_type(self) -> type: 

25 return Text 

26 

27 def run(self, *texts: Text) -> Text: 

28 for text in texts: 

29 if not isinstance(text, Text): 

30 raise TypeError(f"text must be Text, got {type(text)}") 

31 return Text( 

32 name=self.get_output_name(), 

33 data=self.separator.join(str(text()) for text in texts), 

34 )