def call_batch_processor()

in components/specialized-parser/src/runner.py [0:0]

• 31 lines of code
• 1 McCabe index (conditional complexity)

    def call_batch_processor(self) -> Operation:
        opts = ClientOptions(
            api_endpoint=f"{self.processor_config.location}-documentai.googleapis.com"
        )
        client_info = ClientInfo(user_agent=USER_AGENT)
        client = documentai.DocumentProcessorServiceClient(
            client_options=opts, client_info=client_info
        )

        gcs_prefix = documentai.GcsPrefix(
            gcs_uri_prefix=self.job_config.gcs_input_prefix
        )
        input_config = documentai.BatchDocumentsInputConfig(gcs_prefix=gcs_prefix)

        gcs_output_config = documentai.DocumentOutputConfig.GcsOutputConfig(
            gcs_uri=self.job_config.gcs_output_uri
        )
        output_config = documentai.DocumentOutputConfig(
            gcs_output_config=gcs_output_config
        )

        processor_name = client.processor_path(
            self.processor_config.project,
            self.processor_config.location,
            self.processor_config.processor_id,
        )
        request = documentai.BatchProcessRequest(
            name=processor_name,
            input_documents=input_config,
            document_output_config=output_config,
        )
        operation: Operation = client.batch_process_documents(request)
        logger.info("Started batch process")
        return operation