sqlmesh.core.reference

22    @property
23    def columns(self) -> t.List[str]:
24        expression = self.expression
25        if isinstance(expression, exp.Alias):
26            expression = expression.this
27        expression = expression.unnest()
28        if isinstance(expression, (exp.Array, exp.Tuple)):
29            return [e.output_name for e in expression.expressions]
30        return [expression.output_name]

32    @property
33    def name(self) -> str:
34        if not self._name:
35            keys = []
36
37            if isinstance(self.expression, (exp.Tuple, exp.Array)):
38                for e in self.expression.expressions:
39                    if not e.output_name:
40                        raise ConfigError(
41                            f"Reference '{e}' must have an inferrable name or explicit alias."
42                        )
43                    keys.append(e.output_name)
44            elif self.expression.output_name:
45                keys.append(self.expression.output_name)
46            else:
47                raise ConfigError(
48                    f"Reference '{self.expression}' must have an inferrable name or explicit alias."
49                )
50
51            self._name = "__".join(keys)
52        return self._name

358def init_private_attributes(self: BaseModel, context: Any, /) -> None:
359    """This function is meant to behave like a BaseModel method to initialise private attributes.
360
361    It takes context as an argument since that's what pydantic-core passes when calling it.
362
363    Args:
364        self: The BaseModel instance.
365        context: The context.
366    """
367    if getattr(self, '__pydantic_private__', None) is None:
368        pydantic_private = {}
369        for name, private_attr in self.__private_attributes__.items():
370            default = private_attr.get_default()
371            if default is not PydanticUndefined:
372                pydantic_private[name] = default
373        object_setattr(self, '__pydantic_private__', pydantic_private)

56    def __init__(self, models: t.Iterable[Model]):
57        self._model_refs: t.Dict[str, t.Dict[str, Reference]] = {}
58        self._ref_models: t.Dict[str, t.Set[str]] = {}
59        self._dim_models: t.Dict[str, t.Set[str]] = {}
60
61        for model in models:
62            self.add_model(model)

64    def add_model(self, model: Model) -> None:
65        """Add a model and its references to the graph.
66
67        Args:
68            model: the model to add.
69        """
70        for column in model.columns_to_types or {}:
71            self._dim_models.setdefault(column, set())
72            self._dim_models[column].add(model.name)
73
74        for ref in model.all_references:
75            self._model_refs.setdefault(model.name, {})
76            self._model_refs[model.name][ref.name] = ref
77            self._ref_models.setdefault(ref.name, set())
78            self._ref_models[ref.name].add(model.name)

 80    def models_for_column(self, source: str, column: str, max_depth: int = 3) -> t.List[str]:
 81        """Find all the models with a column that join to a source within max_depth.
 82
 83        Args:
 84            source: The source model.
 85            column: The column to look for.
 86            max_depth: The maximum number of models to join to find a path.
 87
 88        Returns:
 89            The list of models that fit the criteria of the search.
 90        """
 91        models = []
 92
 93        for model in self._dim_models[column]:
 94            try:
 95                if model != source:
 96                    self.find_path(source, model)
 97                models.append(model)
 98            except SQLMeshError:
 99                pass
100
101        return sorted(models)

103    def find_path(self, source: str, target: str, max_depth: int = 3) -> t.List[Reference]:
104        """Find a path from source model to target model with max depth.
105
106        Args:
107            source: The source model.
108            target: The target model.
109            max_depth: The maximum number of models to join to find a path.
110
111        Returns:
112            The list of references representing the join path of source to target.
113        """
114        if source not in self._model_refs:
115            return []
116
117        queue = deque(([ref] for ref in self._model_refs[source].values()))
118
119        while queue:
120            path = queue.popleft()
121            visited = set()
122            many = False
123
124            for ref in path:
125                visited.add(ref.model_name)
126                many = many or not ref.unique
127
128            ref_name = path[-1].name
129
130            for model_name in sorted(self._ref_models[ref_name]):
131                for ref in self._model_refs[model_name].values():
132                    # paths cannot have loops or contain many to many refs
133                    if model_name in visited or (many and not ref.unique):
134                        continue
135
136                    new_path = path + [ref]
137
138                    if model_name == target:
139                        return new_path
140
141                    if len(new_path) < max_depth:
142                        queue.append(new_path)
143
144        raise SQLMeshError(
145            f"Cannot find path between '{source}' and '{target}'. Make sure that references/grains are configured and that a many to many join is not occurring."
146        )