Я новичок в типе подсказок / проверок 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
?