This method is called from the manager. It must return a list or a generator
        of BaseRecord objects.
        When it has nothing else to read, it must set class variable "finished" to True.
555529
def get_next_batch(self):
        
        messages = self.get_from_kafka()
        if messages:
            for message in messages:
                item = BaseRecord(message)
                self.increase_read()
                yield item

        self.logger.debug('Done reading batch')
        self.last_position = self.consumer.offsets
420613
def get_all_fields(cls, class_obj=None, fields=None):
        

        def return_fields(obj):

            internal_fields = fields
            if internal_fields is None:
                internal_fields = {}

            for attribute in dir(obj):

                try:
                    attr_val = getattr(obj, attribute)
                    attr_cls = attr_val.__class__

                    # If it is a model field, call on init
                    if issubclass(attr_cls, ModelField):
                        internal_fields[attribute] = attr_val
                except:
                    pass

            return internal_fields

        if class_obj is None:
            class_obj = cls

            fields = return_fields(class_obj)
            for parent_class in cls.__bases__:
                parent_fields = cls.get_all_fields(parent_class, fields)
                for field_name, field_value in list(parent_fields.items()):
                    if not field_name in fields:
                        fields[field_name] = field_value

            return fields

        else:
            if not isinstance(class_obj, CollectionModel):
                return fields
435395
def objects(self, json_result=False):
        
        self.rows_num = 0
        query = self.query()
        if not isinstance(query, (orm.Result, list, dict)):
            query = do_(query)
        for record in query:
            self.rows_num += 1
            r = self.object(record, json_result)
            self._cal_sum(record)
            yield r
        total = self._render_sum(True)
        if total:
            yield total
1237958
public synchronized void finished()
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "finished");
    
    checkEntryParent();
    
    synchronized(parentList)
    {
      //remove the cursor from the list
      //this will set the parent list reference to null and make the
      //cursor seubsequently unusable
      forceRemove();      
    }    
    
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "finished");        
  }
449019
def make_factory(self, cls, count):
        
        field_names = cls._meta.get_all_field_names()
        fields = {}
        text = []
        finalizer = None
        scaffold = scaffolding.scaffold_for_model(cls)

        for field_name in field_names:
            generator = getattr(scaffold, field_name, None)
            if generator:
                if hasattr(generator, 'set_up'):
                    generator.set_up(cls, count)
                fields[field_name] = generator
                text.append(u'%s: %s; ' % (field_name, fields[field_name]))
        try:
            self.stdout.write(u'Generator for %s: %s\n' % (cls, u''.join(text)))
        except models.ObjectDoesNotExist:
            self.stdout.write(u'Generator for %s\n' % u''.join(text))

        if hasattr(scaffold, 'finalize') and hasattr(scaffold.finalize, '__call__'):
            finalizer = scaffold.finalize

        return fields, finalizer
1270642
public <T extends DataObject> List<T> collect(Class<T> type, Callable<T> factory) {
        Generator<T> generator = new Generator<T>(type, factory);
        _generator = generator;
        return generator.list;
    }
1292784
def find_all(self, cls):
        """"""
        final_results = []
        table = self.get_class_table(cls)
        for db_result in table.scan():
            obj = cls.from_data(db_result['value'])
            final_results.append(obj)

        return final_results
210403
@Override
	public ReversibleIterator<BasicLogRecord> iterator() {
		logMgrLock.lock();
		try {
			flush();
			return new LogIterator(currentBlk);
		} finally {
			logMgrLock.unlock();
		}
	}
1050226
public List<?> range(Value begin, Value end) {
		List<Object> results = new ArrayList<Object>();
		Key[] elementKeys = getElementKeys();
		if (elementKeys != null && elementKeys.length > 0) {
			List<Record> records = fetchSubRecords(elementKeys);
			for (Record record : records) {
				if (record != null && filterBinByRange(record, ListElementBinName, begin, end)) {
					results.add(record.getValue(ListElementBinName));
				}
			}
		}
		return results;
	}
733374
def _objectListGenerator(self, request, objectList):
        
        return self._topLevelObjectGenerator(
            request, len(objectList), lambda index: objectList[index])
1170051
def as_completed(jobs):
  '''  '''

  jobs = tuple(jobs)
  event = threading.Event()
  callback = lambda f, ev: event.set()
  [job.add_listener(Job.SUCCESS, callback, once=True) for job in jobs]
  [job.add_listener(Job.ERROR, callback, once=True) for job in jobs]

  while jobs:
    event.wait()
    event.clear()

    jobs, finished = split_list_by(jobs, lambda x: x.finished)
    for job in finished:
      yield job
481940
def has_completed(self):
        """"""
        objects_in_db = WorkflowObjectModel.query.filter(
            WorkflowObjectModel.id_workflow == self.uuid,
            WorkflowObjectModel.id_parent == None,  # noqa
        ).filter(WorkflowObjectModel.status.in_([
            workflow_object_class.known_statuses.COMPLETED
        ])).count()
        return objects_in_db == len(list(self.objects))
1049263
private Iterator<D> getIterator() throws DataRecordException, IOException {
    if (Boolean.valueOf(this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_IS_SPECIFIC_API_ACTIVE))) {
      return this.getRecordSetFromSourceApi(this.schema, this.entity, this.workUnit, this.predicateList);
    }
    return this.getRecordSet(this.schema, this.entity, this.workUnit, this.predicateList);
  }
60129
def next_trials(self):
        
        trials = list(self._trial_generator)
        if self._shuffle:
            random.shuffle(trials)
        self._finished = True
        return trials
934376
def each
      # Need to close the connection, teh result_set, AND the result_set.getStatement when
      # we're done.
      connection = open_connection!

      # We're going to need to ask for item/copy info while in the
      # middle of streaming our results. JDBC is happier and more performant
      # if we use a seperate connection for this.
      extra_connection = open_connection! if include_some_holdings?

      # We're going to make our marc records in batches, and only yield
      # them to caller in batches, so we can fetch copy/item info in batches
      # for efficiency.
      batch_size = settings["horizon.batch_size"].to_i
      record_batch = []

      exclude_tags = (settings["horizon.exclude_tags"] || "").split(",")


      rs = self.fetch_result_set!(connection)

      current_bib_id = nil
      record = nil
      record_count = 0

      error_handler = org.marc4j.ErrorHandler.new

      while(rs.next)
        bib_id      = rs.getInt("bib#");

        if bib_id != current_bib_id
          record_count += 1

          if settings["debug_ascii_progress"] &&  (record_count % settings["solrj_writer.batch_size"] == 0)
            $stderr.write ","
          end

          # new record! Put old one on batch queue.
          record_batch << record if record

          # prepare and yield batch?
          if (record_count % batch_size == 0)
            enhance_batch!(extra_connection, record_batch)
            record_batch.each do |r|
              # set current_bib_id for error logging
              current_bib_id = r['001'].value
              yield r
            end
            record_batch.clear
          end

          # And start new record we've encountered.
          error_handler = org.marc4j.ErrorHandler.new
          current_bib_id = bib_id
          record = MARC::Record.new
          record.append MARC::ControlField.new("001", bib_id.to_s)
        end

        tagord      = rs.getInt("tagord");
        tag         = rs.getString("tag")

        # just silently skip it, some weird row in the horizon db, it happens.
        # plus any of our exclude_tags.
        next if tag.nil? || tag == "" || exclude_tags.include?(tag)

        indicators = rs.getString("indicators")

        # a packed byte array could be in various columns, in order of preference...
        # the xref stuff is joined in from the auth table
        # Have to get it as bytes and then convert it to String to avoid JDBC messing
        # up the encoding marc8 grr
        authtext = rs.getBytes("xref_longtext") || rs.getBytes("xref_text")
        text     = rs.getBytes("longtext") || rs.getBytes("text")


        if tag == "000"
          # Horizon puts a \x1E marc field terminator on the end of hte
          # leader in the db too, but that's not really part of it.
          record.leader =  String.from_java_bytes(text).chomp("\x1E")

          fix_leader!(record.leader)
        elsif tag != "001"
          # we add an 001 ourselves with bib id in another part of code.
          field = build_marc_field!(error_handler, tag, indicators, text, authtext)
          record.append field unless field.nil?
        end
      end

      # last one
      record_batch << record if record

      # yield last batch
      enhance_batch!(extra_connection, record_batch)

      record_batch.each do |r|
        yield r
      end
      record_batch.clear

    rescue Exception => e
      logger.fatal "HorizonReader, unexpected exception at bib id:#{current_bib_id}: #{e}"    
      raise e
    ensure
      logger.info("HorizonReader: Closing all JDBC objects...")

      # have to cancel the statement to keep us from waiting on entire
      # result set when exception is raised in the middle of stream.
      statement = rs && rs.getStatement
      if statement
        statement.cancel
        statement.close
      end

      rs.close if rs

      # shouldn't actually need to close the resultset and statement if we cancel, I think.
      connection.close if connection

      extra_connection.close if extra_connection

      logger.info("HorizonReader: Closed JDBC objects")
    end
1184780
def _check_base_classes(base_classes, check_for_type):
    
    return_value = False
    for base in base_classes:
        if base.__name__ == check_for_type:
            return_value = True
            break
        else:
            return_value = _check_base_classes(base.__bases__, check_for_type)
            if return_value is True:
                break

    return return_value
614647
function resultRaw (results) {
    // throw error if results required and not retrieved
    if (this.args.required && !results.length) {
        // throw error
        throw this.error('no records found')
    }
    // if only ids were selected the return result object to fetch records
    if (this.select.isSelectOnlyIds) {
        // build result object
        var result = new ImmutableCoreModelResult({
            query: this,
            results: results,
        })
        // return result or all records if specified
        return this.args.all ? result.all() : result
    }
    // otherwise return ImmutableCoreModelInstance record(s)
    else {
        // query is for a single record
        if (this.args.limit === 1) {
            // return new instance or undefined
            return results.length
                ? this.newInstance(results[0])
                : undefined
        }
        // query is for multiple records
        else {
            return _.map(results, result => this.newInstance(result))
        }
    }
}