parameters: (self, query, file_path, file_format='CSV',
                       delimiter=',', header=True)
docstring: Runs a query as the current user and saves the result to a file.
  
          query can be a sql query or table reference.
docstring_summary: Runs a query as the current user and saves the result to a file.
docstring_tokens: Runs a query as the current user and saves the result to a file .
function: def export_query(self, query, file_path, file_format='CSV',
                       delimiter=',', header=True):
          """
          Runs a query as the current user and saves the result to a file.
  
          query can be a sql query or table reference.
          """
          header_option = 'HEADER' if header else ''
          query = query.split(';')[0].strip()
  
          self._check_for_injections(file_format)
          self._check_for_injections(header_option)
  
          meta_query = 'COPY (%s) TO STDOUT WITH %s %s DELIMITER %s;'
          params = (AsIs(query), AsIs(file_format),
                    AsIs(header_option), delimiter)
  
          cur = self.connection.cursor()
          query = cur.mogrify(meta_query, params)
  
          # Store pending exports in a temporary location so they're aren't
          # discoverable while being exported.
          tmp_path = '/tmp/user_exports/{0}-{1}'.format(
              uuid4().hex, hashlib.sha256(query).hexdigest())
          try:
              os.makedirs('/tmp/user_exports')
          except OSError as e:
              if e.errno != errno.EEXIST:
                  raise e
  
          try:
              with open(tmp_path, 'w') as f:
                  cur.copy_expert(query, f)
          except psycopg2.Error as e:
              # Delete the temporary files of failed exports.
              os.remove(tmp_path)
              _convert_pg_exception(e)
          finally:
              cur.close()
          # Move successful exports into the user's data folder.
          # os.rename() would fail here if /tmp and /user_data are stored on
          # different filesystems, so use shutil.move() instead.
          shutil.move(tmp_path, file_path)
function_tokens: def export_query ( self , query , file_path , file_format = 'CSV' , delimiter = ',' , header = True ) : header_option = 'HEADER' if header else '' query = query . split ( ';' ) [ 0 ] . strip ( ) self . _check_for_injections ( file_format ) self . _check_for_injections ( header_option ) meta_query = 'COPY (%s) TO STDOUT WITH %s %s DELIMITER %s;' params = ( AsIs ( query ) , AsIs ( file_format ) , AsIs ( header_option ) , delimiter ) cur = self . connection . cursor ( ) query = cur . mogrify ( meta_query , params ) # Store pending exports in a temporary location so they're aren't # discoverable while being exported. tmp_path = '/tmp/user_exports/{0}-{1}' . format ( uuid4 ( ) . hex , hashlib . sha256 ( query ) . hexdigest ( ) ) try : os . makedirs ( '/tmp/user_exports' ) except OSError as e : if e . errno != errno . EEXIST : raise e try : with open ( tmp_path , 'w' ) as f : cur . copy_expert ( query , f ) except psycopg2 . Error as e : # Delete the temporary files of failed exports. os . remove ( tmp_path ) _convert_pg_exception ( e ) finally : cur . close ( ) # Move successful exports into the user's data folder. # os.rename() would fail here if /tmp and /user_data are stored on # different filesystems, so use shutil.move() instead. shutil . move ( tmp_path , file_path )