Iterating Files Recursively with RxPy

rxpy_iterating_and_filtering_files_recursively.py

from rx import Observable
import os


def recursive_files_in_directory(folder):

    def emit_files_recursively(observer):
        for root, directories, filenames in os.walk(folder):
            for directory in directories:
                observer.on_next(os.path.join(root, directory))
            for filename in filenames:
                observer.on_next(os.path.join(root, filename))

        observer.on_completed()

    return Observable.create(emit_files_recursively)

recursive_files_in_directory('C:\\MyFolder') \
    .filter(lambda f: f.endswith('.csv')) \
    .flat_map(lambda f: Observable.from_(open(f))) \ # read lines
    .subscribe(lambda l: print(l)) # print lines