@wrap_exceptions<EOL><INDENT>def bulk_upsert(self, docs, namespace, timestamp):<DEDENT>
def docs_to_upsert():<EOL><INDENT>doc = None<EOL>for doc in docs:<EOL><INDENT>index, doc_type = self._index_and_mapping(namespace)<EOL>doc_id = str(doc.pop("<STR_LIT>"))<EOL>document_action = {<EOL>"<STR_LIT>": index,<EOL>"<STR_LIT>": doc_type,<EOL>"<STR_LIT>": doc_id,<EOL>"<STR_LIT>": self._formatter.format_document(doc),<EOL>}<EOL>document_meta = {<EOL>"<STR_LIT>": self.meta_index_name,<EOL>"<STR_LIT>": self.meta_type,<EOL>"<STR_LIT>": doc_id,<EOL>"<STR_LIT>": {"<STR_LIT>": namespace, "<STR_LIT>": timestamp},<EOL>}<EOL>yield document_action<EOL>yield document_meta<EOL><DEDENT>if doc is None:<EOL><INDENT>raise errors.EmptyDocsError(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>"<EOL>)<EOL><DEDENT><DEDENT>try:<EOL><INDENT>kw = {}<EOL>if self.chunk_size > <NUM_LIT:0>:<EOL><INDENT>kw["<STR_LIT>"] = self.chunk_size<EOL><DEDENT>responses = streaming_bulk(<EOL>client=self.elastic, actions=docs_to_upsert(), **kw<EOL>)<EOL>for ok, resp in responses:<EOL><INDENT>if not ok:<EOL><INDENT>LOG.error(<EOL>"<STR_LIT>"<EOL>"<STR_LIT>" % resp<EOL>)<EOL><DEDENT><DEDENT>if self.auto_commit_interval == <NUM_LIT:0>:<EOL><INDENT>self.commit()<EOL><DEDENT><DEDENT>except errors.EmptyDocsError:<EOL><INDENT>pass<EOL><DEDENT>
Insert multiple documents into Elasticsearch.