def get_documents()

in python/src/tablestore_for_agent_memory/knowledge/knowledge_store.py [0:0]

• 26 lines of code
• 6 McCabe index (conditional complexity)

    def get_documents(self, document_id_list: List[str], tenant_id: Optional[str] = None) -> List[Optional[Document]]:
        if len(document_id_list) is None:
            return []
        tenant_id = self._check_enable_multi_tenant_id(tenant_id)
        documents = []
        total = len(document_id_list)
        batch_size = 100
        for start in range(0, total, batch_size):
            end = start + batch_size
            current_batch = document_id_list[start:end]
            rows_to_get = []
            for document_id in current_batch:
                primary_key = [('document_id', document_id), ('tenant_id', tenant_id)]
                rows_to_get.append(primary_key)
            request = BatchGetRowRequest()
            request.add(TableInBatchGetRowItem(self._table_name, rows_to_get, None, None, 1))
            result = self._client.batch_get_row(request)
            table_result = result.get_result_by_table(self._table_name)
            for i in range(len(table_result)):
                item = table_result[i]
                if item.is_ok:
                    document = TablestoreHelper.row_to_document(item.row, self._text_field, self._embedding_field)
                    documents.append(document)
                else:
                    raise ValueError(f"read failed for row:{rows_to_get[i]}, error code: {item.error_code}, error message: {item.error_message}")
        return documents