Python проверка типа во время выполнения в конвейере с генератором - PullRequest
0 голосов
/ 11 февраля 2020

Я новичок в типе подсказок / проверок Python. У меня есть конвейер, который читает данные, очищает их, расширяет их новыми полями, а затем выводит их в файл. По сути это код, не вдаваясь в подробности.

from functools import reduce

pipe = [
        MyReader(base_dir)(), 
        MyCleaner(), 
        MyExtender(), 
        MyDumper()
       ]
res = reduce(lambda f, g: g(f), pipe)

Пример моего MyReader класса:

class MyReader():
    def __init__(self, base_dir: str) -> None:
        self.base_dir = base_dir

    def data_generator(self) -> Generator[Union[DataTypeA, DataTypeB], None, None]:
        for root, _, files in itertools.chain.from_iterable(os.walk(str(path)) for path in self.path_dir):
            for filename in files:
                content = read_file(os.path.join(root, filename))
                yield content

    def __call__(self):
        return self.data_generator()

и пример MyExtender класса:

class MyExtender():
    def __init__(self, type: str):
        if type == 'TypeA':
            self.pipe = [calculate_one_thing, calculate_another_thing]
        else:
            self.pipe = [calculate_a_different_thing]

    def __call__(self, gen):
        func_pipe = lambda x: reduce(lambda f,g: g(f), self.pipe, x)
        return (func_pipe(item) for item in gen)

Прежде всего, Я хочу доказать, что генератор в MyReader выдает элементы, которые могут принадлежать TypeA или TypeB, где каждый из типов имеет обязательные и дополнительные поля. Если класс получает элемент, который не содержит обязательного поля ни одного из типов, или содержит поле, которое не указано ни как обязательное, ни как необязательное для какого-либо класса, проверка типа должна выдать исключение / ошибку.

То же самое относится к классу MyExtender (и аналогично для других шагов в конвейере).

Я немного изучил pydanctic, но не смог понять, как использовать его в моем конвейере. Есть ли простой способ включить проверку типа / подсказку уже в моем генераторе внутри MyReader?

...