def get_schema_tree()

in azure/Kqlmagic/database_html.py [0:0]

• 36 lines of code
• 24 McCabe index (conditional complexity)

    def get_schema_tree(cls, engine:KqlEngine, **options)->Dict[str,Any]:
        engine_class = (
            AriaEngine
            if isinstance(engine, AriaEngine) or (isinstance(engine, CacheEngine) and isinstance(engine.kql_engine, AriaEngine))
            # must be after AriaEngine, because AriaEngine class inherit from KustoEngine
            else KustoEngine
            if isinstance(engine, KustoEngine) or (isinstance(engine, CacheEngine) and isinstance(engine.kql_engine, KustoEngine))
            else LoganalyticsEngine
            if isinstance(engine, LoganalyticsEngine) or (isinstance(engine, CacheEngine) and isinstance(engine.kql_engine, LoganalyticsEngine))
            else AimonEngine
            if isinstance(engine, AimonEngine) or (isinstance(engine, CacheEngine) and isinstance(engine.kql_engine, AimonEngine))
            # must be after AimonEngine, because AimonEngine class inherit from AppinsightsEngine
            else AppinsightsEngine 
            if isinstance(engine, AppinsightsEngine) or (isinstance(engine, CacheEngine) and isinstance(engine.kql_engine, AppinsightsEngine))
            else None
        )

        if engine_class is not None:
            if isinstance(engine, CacheEngine):
                client_database_name = engine.kql_engine.get_client_database_name()
            else:
                client_database_name = engine.get_client_database_name()

            if engine_class in [KustoEngine, AriaEngine]:
                show_schema_query = f".show database ['{cls._adjustToKustoEntityNameRules(client_database_name)}'] schema"
                raw_query_result = engine.execute(show_schema_query, **options)
                raw_schema_table = raw_query_result.tables[0]
                database_metadata_tree = cls._create_database_metadata_tree(raw_schema_table.fetchall(), client_database_name)
                if options.get("cache") is not None and options.get("cache") != options.get("use_cache"):
                    CacheClient(**options).save(raw_query_result, engine, show_schema_query, **options)
                return database_metadata_tree

            elif engine_class in [AppinsightsEngine, LoganalyticsEngine, AimonEngine]:
                show_schema_query = ".show schema"
                metadata_result = engine.client_execute(show_schema_query, **options)
                metadata_schema_table = metadata_result.table
                database_metadata_tree = cls._create_database_draft_metadata_tree(metadata_schema_table)
                if options.get("cache") is not None and options.get("cache") != options.get("use_cache"):
                    CacheClient(**options).save(metadata_result, engine, show_schema_query, **options)
                return database_metadata_tree
        return None