fix(ingestion/lookml): liquid template resolution and view-to-view cll

metadata-ingestion/src/datahub/ingestion/source/looker/looker_config.py

@@ -1,13 +1,14 @@
 import dataclasses
 import os
 import re
-from typing import Any, ClassVar, Dict, List, Optional, Union, cast
+from typing import Any, ClassVar, Dict, List, Optional, Tuple, Union, cast
 
 import pydantic
+from looker_sdk.sdk.api40.models import DBConnection
 from pydantic import Field, validator
 
 from datahub.configuration import ConfigModel
-from datahub.configuration.common import AllowDenyPattern
+from datahub.configuration.common import AllowDenyPattern, ConfigurationError
 from datahub.configuration.source_common import (
     EnvConfigMixin,
     PlatformInstanceConfigMixin,
@@ -143,6 +144,82 @@ class LookerCommonConfig(EnvConfigMixin, PlatformInstanceConfigMixin):
     )
 
 
+def _get_bigquery_definition(
+    looker_connection: DBConnection,
+) -> Tuple[str, Optional[str], Optional[str]]:
+    platform = "bigquery"
+    # bigquery project ids are returned in the host field
+    db = looker_connection.host
+    schema = looker_connection.database
+    return platform, db, schema
+
+
+def _get_generic_definition(
+    looker_connection: DBConnection, platform: Optional[str] = None
+) -> Tuple[str, Optional[str], Optional[str]]:
+    if platform is None:
+        # We extract the platform from the dialect name
+        dialect_name = looker_connection.dialect_name
+        assert dialect_name is not None
+        # generally the first part of the dialect name before _ is the name of the platform
+        # versions are encoded as numbers and can be removed
+        # e.g. spark1 or hive2 or druid_18
+        platform = re.sub(r"[0-9]+", "", dialect_name.split("_")[0])
+
+    assert (
+        platform is not None
+    ), f"Failed to extract a valid platform from connection {looker_connection}"
+    db = looker_connection.database
+    schema = looker_connection.schema  # ok for this to be None
+    return platform, db, schema
+
+
+class LookerConnectionDefinition(ConfigModel):
+    platform: str
+    default_db: str
+    default_schema: Optional[str]  # Optional since some sources are two-level only
+    platform_instance: Optional[str] = None
+    platform_env: Optional[str] = Field(
+        default=None,
+        description="The environment that the platform is located in. Leaving this empty will inherit defaults from "
+        "the top level Looker configuration",
+    )
+
+    @validator("platform_env")
+    def platform_env_must_be_one_of(cls, v: Optional[str]) -> Optional[str]:
+        if v is not None:
+            return EnvConfigMixin.env_must_be_one_of(v)
+        return v
+
+    @validator("platform", "default_db", "default_schema")
+    def lower_everything(cls, v):
+        """We lower case all strings passed in to avoid casing issues later"""
+        if v is not None:
+            return v.lower()
+
+    @classmethod
+    def from_looker_connection(
+        cls, looker_connection: DBConnection
+    ) -> "LookerConnectionDefinition":
+        """Dialect definitions are here: https://docs.looker.com/setup-and-management/database-config"""
+        extractors: Dict[str, Any] = {
+            "^bigquery": _get_bigquery_definition,
+            ".*": _get_generic_definition,
+        }
+
+        if looker_connection.dialect_name is None:
+            raise ConfigurationError(
+                f"Unable to fetch a fully filled out connection for {looker_connection.name}. Please check your API permissions."
+            )
+        for extractor_pattern, extracting_function in extractors.items():
+            if re.match(extractor_pattern, looker_connection.dialect_name):
+                (platform, db, schema) = extracting_function(looker_connection)
+                return cls(platform=platform, default_db=db, default_schema=schema)
+        raise ConfigurationError(
+            f"Could not find an appropriate platform for looker_connection: {looker_connection.name} with dialect: {looker_connection.dialect_name}"
+        )
+
+
 class LookerDashboardSourceConfig(
     LookerAPIConfig,
     LookerCommonConfig,

metadata-ingestion/src/datahub/ingestion/source/looker/looker_connection.py

@@ -1,85 +1,55 @@
-import re
-from typing import Any, Dict, Optional, Tuple
+import logging
+from typing import Optional
 
+from looker_sdk.error import SDKError
 from looker_sdk.sdk.api40.models import DBConnection
-from pydantic import Field, validator
 
-from datahub.configuration import ConfigModel
 from datahub.configuration.common import ConfigurationError
-from datahub.configuration.source_common import EnvConfigMixin
-
-
-def _get_bigquery_definition(
-    looker_connection: DBConnection,
-) -> Tuple[str, Optional[str], Optional[str]]:
-    platform = "bigquery"
-    # bigquery project ids are returned in the host field
-    db = looker_connection.host
-    schema = looker_connection.database
-    return (platform, db, schema)
-
-
-def _get_generic_definition(
-    looker_connection: DBConnection, platform: Optional[str] = None
-) -> Tuple[str, Optional[str], Optional[str]]:
-    if platform is None:
-        # We extract the platform from the dialect name
-        dialect_name = looker_connection.dialect_name
-        assert dialect_name is not None
-        # generally the first part of the dialect name before _ is the name of the platform
-        # versions are encoded as numbers and can be removed
-        # e.g. spark1 or hive2 or druid_18
-        platform = re.sub(r"[0-9]+", "", dialect_name.split("_")[0])
-
-    assert (
-        platform is not None
-    ), f"Failed to extract a valid platform from connection {looker_connection}"
-    db = looker_connection.database
-    schema = looker_connection.schema  # ok for this to be None
-    return (platform, db, schema)
-
-
-class LookerConnectionDefinition(ConfigModel):
-    platform: str
-    default_db: str
-    default_schema: Optional[str]  # Optional since some sources are two-level only
-    platform_instance: Optional[str] = None
-    platform_env: Optional[str] = Field(
-        default=None,
-        description="The environment that the platform is located in. Leaving this empty will inherit defaults from "
-        "the top level Looker configuration",
-    )
-
-    @validator("platform_env")
-    def platform_env_must_be_one_of(cls, v: Optional[str]) -> Optional[str]:
-        if v is not None:
-            return EnvConfigMixin.env_must_be_one_of(v)
-        return v
-
-    @validator("platform", "default_db", "default_schema")
-    def lower_everything(cls, v):
-        """We lower case all strings passed in to avoid casing issues later"""
-        if v is not None:
-            return v.lower()
-
-    @classmethod
-    def from_looker_connection(
-        cls, looker_connection: DBConnection
-    ) -> "LookerConnectionDefinition":
-        """Dialect definitions are here: https://docs.looker.com/setup-and-management/database-config"""
-        extractors: Dict[str, Any] = {
-            "^bigquery": _get_bigquery_definition,
-            ".*": _get_generic_definition,
-        }
-
-        if looker_connection.dialect_name is None:
-            raise ConfigurationError(
-                f"Unable to fetch a fully filled out connection for {looker_connection.name}. Please check your API permissions."
+from datahub.ingestion.source.looker.looker_config import LookerConnectionDefinition
+from datahub.ingestion.source.looker.looker_lib_wrapper import LookerAPI
+from datahub.ingestion.source.looker.lookml_config import (
+    LookMLSourceConfig,
+    LookMLSourceReport,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def get_connection_def_based_on_connection_string(
+    connection: str,
+    source_config: LookMLSourceConfig,
+    looker_client: Optional[LookerAPI],
+    reporter: LookMLSourceReport,
+) -> Optional[LookerConnectionDefinition]:
+    if source_config.connection_to_platform_map is None:
+        source_config.connection_to_platform_map = {}
+
+    assert source_config.connection_to_platform_map is not None
+
+    connection_def: Optional[LookerConnectionDefinition] = None
+
+    if connection in source_config.connection_to_platform_map:
+        connection_def = source_config.connection_to_platform_map[connection]
+    elif looker_client:
+        try:
+            looker_connection: DBConnection = looker_client.connection(connection)
+        except SDKError:
+            logger.error(
+                f"Failed to retrieve connection {connection} from Looker. This usually happens when the "
+                f"credentials provided are not admin credentials."
             )
-        for extractor_pattern, extracting_function in extractors.items():
-            if re.match(extractor_pattern, looker_connection.dialect_name):
-                (platform, db, schema) = extracting_function(looker_connection)
-                return cls(platform=platform, default_db=db, default_schema=schema)
-        raise ConfigurationError(
-            f"Could not find an appropriate platform for looker_connection: {looker_connection.name} with dialect: {looker_connection.dialect_name}"
-        )
+        else:
+            try:
+                connection_def = LookerConnectionDefinition.from_looker_connection(
+                    looker_connection
+                )
+
+                # Populate the cache (using the config map) to avoid calling looker again for this connection
+                source_config.connection_to_platform_map[connection] = connection_def
+            except ConfigurationError:
+                reporter.report_warning(
+                    f"connection-{connection}",
+                    "Failed to load connection from Looker",
+                )
+
+    return connection_def

metadata-ingestion/src/datahub/ingestion/source/looker/looker_file_loader.py

@@ -4,16 +4,16 @@
 from typing import Any, Dict, Optional
 
 from datahub.ingestion.source.looker.lkml_patched import load_lkml
-from datahub.ingestion.source.looker.looker_connection import LookerConnectionDefinition
+from datahub.ingestion.source.looker.looker_config import LookerConnectionDefinition
 from datahub.ingestion.source.looker.looker_dataclasses import LookerViewFile
+from datahub.ingestion.source.looker.looker_template_language import (
+    resolve_liquid_variable_in_view_dict,
+)
 from datahub.ingestion.source.looker.lookml_config import (
     _EXPLORE_FILE_EXTENSION,
     _VIEW_FILE_EXTENSION,
     LookMLSourceReport,
 )
-from datahub.ingestion.source.looker.template_lang_resolver import (
-    resolve_liquid_variable_in_view_dict,
-)
 
 logger = logging.getLogger(__name__)
 

metadata-ingestion/src/datahub/ingestion/source/looker/looker_liquid_tag.py

@@ -0,0 +1,104 @@
+from functools import lru_cache
+from typing import ClassVar, Optional, TextIO, cast
+
+from liquid import Environment
+from liquid.ast import Node
+from liquid.context import Context
+from liquid.parse import expect, get_parser
+from liquid.stream import TokenStream
+from liquid.tag import Tag
+from liquid.template import BoundTemplate
+from liquid.token import TOKEN_EXPRESSION, TOKEN_LITERAL, TOKEN_TAG, Token
+
+
+class CustomTagException(Exception):
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class ConditionNode(Node):
+    def __init__(self, tok: Token, sql_or_lookml_reference: str, filter_name: str):
+        self.tok = tok
+
+        self.sql_or_lookml_reference = sql_or_lookml_reference
+
+        self.filter_name = filter_name
+
+    def render_to_output(self, context: Context, buffer: TextIO) -> Optional[bool]:
+        filter_value: Optional[str] = cast(
+            str, context.globals.get(self.filter_name)
+        )  # to silent lint
+
+        if filter_value is None:
+            raise CustomTagException(
+                f'filter {self.filter_name} value is not provided for "condition" tag'
+            )
+
+        filter_value = filter_value.strip()
+
+        buffer.write(f"{self.sql_or_lookml_reference}='{filter_value}'")
+
+        return True
+
+
+# Define the custom tag
+class ConditionTag(Tag):
+    """
+    ConditionTag is the equivalent implementation of looker's custom liquid tag "condition".
+    Refer doc: https://cloud.google.com/looker/docs/templated-filters#basic_usage
+
+    Refer doc to see how to write liquid custom tag: https://jg-rp.github.io/liquid/guides/custom-tags
+
+    This class render the below tag as order.region='ap-south-1' if order_region is provided in config.liquid_variables
+    as order_region: 'ap-south-1'
+        {% condition order_region %} order.region {% endcondition %}
+
+    """
+
+    TAG_START: ClassVar[str] = "condition"
+    TAG_END: ClassVar[str] = "endcondition"
+    name: str = "condition"
+
+    def __init__(self, env: Environment):
+        super().__init__(env)
+        self.parser = get_parser(self.env)
+
+    def parse(self, stream: TokenStream) -> Node:
+        expect(stream, TOKEN_TAG, value=ConditionTag.TAG_START)
+
+        start_token = stream.current
+
+        stream.next_token()
+        expect(stream, TOKEN_EXPRESSION)
+        filter_name: str = stream.current.value.strip()
+
+        stream.next_token()
+        expect(stream, TOKEN_LITERAL)
+
+        sql_or_lookml_reference: str = stream.current.value.strip()
+
+        stream.next_token()
+        expect(stream, TOKEN_TAG, value=ConditionTag.TAG_END)
+
+        return ConditionNode(
+            tok=start_token,
+            sql_or_lookml_reference=sql_or_lookml_reference,
+            filter_name=filter_name,
+        )
+
+
+custom_tags = [ConditionTag]
+
+
+@lru_cache(maxsize=1)
+def _create_env() -> Environment:
+    env: Environment = Environment()
+    # register tag. One time activity
+    for custom_tag in custom_tags:
+        env.add_tag(custom_tag)
+    return env
+
+
+def create_template(text: str) -> BoundTemplate:
+    env: Environment = _create_env()
+    return env.from_string(text)

metadata-ingestion/src/datahub/ingestion/source/looker/looker_source.py

@@ -370,7 +370,8 @@ def _get_input_fields_from_query(
             if field is None:
                 continue
 
-            # we haven't loaded in metadata about the explore yet, so we need to wait until explores are populated later to fetch this
+            # we haven't loaded in metadata about the explore yet, so we need to wait until explores are populated
+            # later to fetch this
             result.append(
                 InputFieldElement(
                     name=field, view_field=None, model=query.model, explore=query.view