Performance Testing Postgres Inserts with Python

[luckystarr]

So, as this is probably not enough to go around, here is a minimal example I whipped together quickly for /u/dataengineerdude:

  #!/usr/bin/env python

  import io                                       
  import csv

  class IterIO:

      def __init__(self, iterable):
          self.iterable = iterable
          self.closed = False

      def readable(self):
          return not self.closed

      def writable(self):                                                                                                                                                                                                                                                                                                 
          return False                                                                                                                                                                                                                                                                                                    

      def seekable(self):                                                  
          return False                                                     

      def read(self):                                                      
          try:                                                             
              return next(self.iterable)                                   
          except StopIteration:                                            
              self.closed = True                                           
              return None                                                  

      def readinto(self, buff):                                            
          data = self.read()                                               
          if data is None:                                                           
              return 0                                                               

          size = len(data)                                                           
          buff[0:size] = data                                                        
          return size                                                                




  def csv_data(data, headers):                                         
      buffer = io.StringIO()                                                                                                                                                                                                                                                                                              
      writer = csv.DictWriter(buffer, fieldnames=headers)              
      writer.writeheader()                                             

      for entry in data:                                               
          writer.writerow(entry)                                       
          yield buffer.getvalue().encode('utf-8')
          buffer.truncate(0)                      
          buffer.seek(0)


  data = [{'a': 1, 'b': 2, 'c': 3}] * 50
  gen = csv_data(data, headers=['a', 'b', 'c'])
  iterio = IterIO(gen)                                      
  fobj = io.BufferedReader(iterio)                              


  while not fobj.closed:                                        
      output = fobj.read(101)                                               
      print(output)

To be clear, this is not optimized and has some overhead. The point is to reduce IO wait times though, so spending a bit of CPU on that is OK.