def execute_data_collection()

in src/dma/collector/workflows/readiness_check/base.py [0:0]

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

    def execute_data_collection(self) -> None:
        canonical_query_manager = next(
            provide_canonical_queries(local_db=self.local_db, working_path=self.working_path)
        )

        self.collection_extractor = CollectionExtractor(
            local_db=self.local_db,
            src_info=self.src_info,
            database=self.database,
            canonical_query_manager=canonical_query_manager,
            console=self.console,
            collection_identifier=self.collection_identifier,
        )
        self.collection_extractor.execute()
        self.db_version = self.collection_extractor.get_db_version()