sources.py

class CsvFileSource(beam.io.filebasedsource.FileBasedSource):

  def read_records(self, file_name, range_tracker):
[...]
    reader = csv.reader(_File2Iterator(self._file), delimiter=self.delimiter)
[...]


class _File2Iterator(object):
  def __init__(self, obj):
    self._obj = obj

  def __iter__(self):
    return self

  def next(self):
    line = self._obj.readline()
    if line == None or line == '':
      raise StopIteration
    return line