code: def csv(ctx, force, threads, mapping, data):
      
  
      # Find out how many lines there are (for the progress bar).
      lines = 0
      for line in DictReader(data):
          lines += 1
      data.seek(0)
  
      # set up objects
      mapping = yaml.load(mapping)
      mapping_loader = MappingLoader(ctx.obj['grano'], mapping)
  
      def process_row(row):
          try:
              mapping_loader.load(row)
          except GranoException, ge:
              msg = '\nServer error: %s' % ge.message
              click.secho(msg, fg='red', bold=True)
              if not force:
                  os._exit(1)
          except RowException, re:
              if not force:
                  msg = '\nRow %s: %s' % (row['__row_id__'], re.message)
                  click.secho(msg, fg='red', bold=True)
                  os._exit(1)
  
      def generate():
          with click.progressbar(DictReader(data),
                                 label=data.name,
                                 length=lines) as bar:
              for i, row in enumerate(bar):
                  row['__row_id__'] = i
                  yield row
  
      threaded(generate(), process_row, num_threads=threads,
               max_queue=1)
docs: Load CSV data into a grano instance using a mapping specification.