Is there a way to decouple a csv generator in my code that doesn't close the file?

Question

I wanted to write the following to decouple a data feed from the rest of my program, but it doesn't work throwing error - "ValueError: I/O operation on closed file."

class Feed:
    def __init__(self, data_loc=None):
        self.data_loc = data_loc
        with open(self.data_loc, newline='') as f:
            self.reader = csv.reader(f)

    def get_next_tick(self):
        return next(self.reader)


raw_data_loc = '{path to file}'

feed = Feed(data_loc=raw_data_loc)

print(feed.get_next_item())

I understand that after init is run the file closes and hence the error, but is there a way of achieving the separation of this csv generator in my code?

Answer 1

Expanding on my comment, you can implement Feed as a context manager like so:

class Feed:
    def __init__(self, *args, **kwargs):
        self.file_obj = open(*args, **kwargs)
        self.reader = csv.reader(self.file_obj)
    def __enter__(self):
        return self
    def __exit__(self, type, value, traceback):
        self.close()
    def get_next_tick(self):
        return next(self.reader)
    def close(self):
        self.file_obj.close()

To print the first line of the feed we can do the following:

with Feed('feed.csv') as feed:
    print(feed.get_next_tick())

Because we used a context manager, the __exit__ method is called even if the feed is empty and raises a StopIteration exception.

If you want to be able to iterate over the feed, you can add __next__ and __iter__ methods like so:

class Feed:
    def __init__(self, *args, **kwargs):
        self.file_obj = open(*args, **kwargs)
        self.reader = csv.reader(self.file_obj)
    def __enter__(self):
        return self
    def __exit__(self, type, value, traceback):
        self.close()
    def __next__(self):
        return self.get_next_tick()
    def __iter__(self):
        return self
    def get_next_tick(self):
        return next(self.reader)
    def close(self):
        self.file_obj.close()

This will allow you to loop over your feed or even call next(feed) directly, eg:

with Feed('feed.csv') as feed:
    for item in feed:
        print(item)

Python Tips has a great chapter on using Context Managers, which were introduced in PEP343.