根据查询获取最多一个对象。

        :return: 查询结果
        :rtype: Object
        :raise: LeanCloudError
503465
def first(self):
        
        params = self.dump()
        params['limit'] = 1
        content = self._do_request(params)
        results = content['results']
        if not results:
            raise LeanCloudError(101, 'Object not found')
        obj = self._new_object()
        obj._update_data(self._process_result(results[0]))
        return obj
615294
def chunks(self):
        
        if not hasattr(self, '_it'):
            class ChunkIterator(object):
                def __iter__(iter):
                    return iter

                def __next__(iter):
                    try:
                        chunk = self._next_chunk()
                    except StopIteration:
                        if self.loop:
                            self._init_stretching()
                            return iter.__next__()

                        raise

                    return chunk
                next = __next__

            self._it = ChunkIterator()

        return self._it
225964
def poll_error(self):
        
        if self.block:
            return self.error

        new_list = self.error[self.old_error_size:]
        self.old_error_size += len(new_list)
        return new_list
166486
def block(self, *args, **kwargs):
        
        LOG.debug(
            '%s on %s (awaitable %s async %s provider %s)',
            'blocking',
            self._func,
            self._is_awaitable,
            self._is_asyncio_provider,
            self._concurrency_provider
        )

        if self._deferable:
            raise RuntimeError('Already activated this call by deferring it')
        with self._lock:
            if not hasattr(self, '_result'):
                try:
                    self._result = (
                        self._concurrency_provider.run_until_complete(self.defer(*args, **kwargs))
                        if self._is_asyncio_provider else self._func(*args, **kwargs)
                    )
                except queue.Empty:
                    raise CloudTimeoutError("No data received after %.1f seconds." % kwargs.get("timeout", 0))
            self._blocked = True
        return self._result
158387
async def main():
    redis = await create_pool(RedisSettings())

    # no id, random id will be generated
    job1 = await redis.enqueue_job('the_task')
    print(job1)
    

    # random id again, again the job will be enqueued and a job will be returned
    job2 = await redis.enqueue_job('the_task')
    print(job2)
    """
    >  <arq job 7d2163c056e54b62a4d8404921094f05>
    """

    # custom job id, job will be enqueued
    job3 = await redis.enqueue_job('the_task', _job_id='foobar')
    print(job3)
    """
    >  <arq job foobar>
    """

    # same custom job id, job will not be enqueued and enqueue_job will return None
    job4 = await redis.enqueue_job('the_task', _job_id='foobar')
    print(job4)
    """
    >  None
    """
9431
def result(self):
        
        if self._result is None:
            self.await_result()
        chunks, exception = self._result
        if exception is None:
            return chunks
        raise exception
1157711
def fetch(self):
        
        params = values.of({})

        payload = self._version.fetch(
            'GET',
            self._uri,
            params=params,
        )

        return TaskQueueInstance(
            self._version,
            payload,
            workspace_sid=self._solution['workspace_sid'],
            sid=self._solution['sid'],
        )
1020488
async def fini(self):
        '''
        
        '''
        assert self.anitted, f'{self.__class__.__name__} initialized improperly.  Must use Base.anit class method.'

        if self.isfini:
            return

        if __debug__:
            import synapse.lib.threads as s_threads  # avoid import cycle
            assert s_threads.iden() == self.tid

        self._syn_refs -= 1
        if self._syn_refs > 0:
            return self._syn_refs

        self.isfini = True

        fevt = self.finievt

        if fevt is not None:
            fevt.set()

        for base in list(self.tofini):
            await base.fini()

        try:
            await self._kill_active_tasks()
        except:
            logger.exception(f'{self} - Exception during _kill_active_tasks')

        for fini in self._fini_funcs:
            try:
                await s_coro.ornot(fini)
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception(f'{self} - fini function failed: {fini}')

        self._syn_funcs.clear()
        self._fini_funcs.clear()
        return 0
1318680
def run(self):
        
        try:
            chunk_index = self.chunk_index_gen(self.array.shape,
                                               self.iteration_order)
            for key in chunk_index:
                # Now we have the slices that describe the next chunk.
                # For example, key might be equivalent to
                # `[11:12, 0:3, :, :]`.
                # Simply "realise" the data for that region and emit it
                # as a Chunk to all registered output queues.
                if self.masked:
                    data = self.array[key].masked_array()
                else:
                    data = self.array[key].ndarray()
                output_chunk = Chunk(key, data)
                self.output(output_chunk)
        except:
            self.abort()
            raise
        else:
            for queue in self.output_queues:
                queue.put(QUEUE_FINISHED)
321433
def error(self):
        
        if self.__timed_out:
            return TimeoutError(self.session, self, "timeout")
        if self.__exit_code is not None and \
                self.__expected_exit_code is not None and \
                self.__exit_code != self.__expected_exit_code:
            return ExitCodeError(self.session, self,
                                 'bad exit code: Got %s' % self.__exit_code)
543865
def sequence(self):
        
        self.open()
        seq = lvm_vg_get_seqno(self.handle)
        self.close()
        return seq
1144151
def main(self):
        
        assert self.params.func, "No subcommand defined in `ElastiCluster.main()"
        try:
            return self.params.func()
        except Exception as err:
            log.error("Error: %s", err)
            if self.params.verbose > 2:
                import traceback
                traceback.print_exc()
            print("Aborting because of errors: {err}.".format(err=err))
            sys.exit(1)
50130
def run(self):
    

    if self.__target is not None:
      return self.__target(self, *self.__args, **self.__kwargs)
    raise NotImplementedError
208174
def get(self, object_id):
        
        if not object_id:
            raise LeanCloudError(code=101, error='Object not found.')
        obj = self._query_class.create_without_data(object_id)
        obj.fetch(select=self._select, include=self._include)
        return obj
840679
def run(self):
        
        region = AWSServiceRegion(access_key=self.key, secret_key=self.secret,
                                  uri=self.endpoint)
        query = self.query_factory(action=self.action, creds=region.creds,
                                   endpoint=region.ec2_endpoint,
                                   other_params=self.parameters)

        def write_response(response):
            print >> self.output, "URL: %s" % query.client.url
            print >> self.output
            print >> self.output, "HTTP status code: %s" % query.client.status
            print >> self.output
            print >> self.output, response

        def write_error(failure):
            if failure.check(AWSError):
                message = failure.value.original
            else:
                message = failure.getErrorMessage()
                if message.startswith("Error Message: "):
                    message = message[len("Error Message: "):]

            print >> self.output, "URL: %s" % query.client.url
            print >> self.output
            if getattr(query.client, "status", None) is not None:
                print >> self.output, "HTTP status code: %s" % (
                    query.client.status,)
                print >> self.output
            print >> self.output, message

            if getattr(failure.value, "response", None) is not None:
                print >> self.output
                print >> self.output, failure.value.response

        deferred = query.submit()
        deferred.addCallback(write_response)
        deferred.addErrback(write_error)
        return deferred
398729
def get(self):
        
        try:
            item = self._queue.get_nowait()
        except (Empty, PersistEmpty):
            return None

        if self._persistence_path:
            self._queue.task_done()

        return item
52002
def join(self):
        
        self._jobs.join()
        try:
            return self._results, self._errors
        finally:
            self._clear()