We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Stop trying to run sources in dbt runner
<!-- *** Make sure you have searched for an existing bug report for this issue *** -->

**Describe the bug**
fal takes sources as valid nodes to run in dbt and tries to run them.
It mostly ends up in a no-op operation, but if the source shares name with a model it can run the wrong model.

**Expected behavior**
fal should not try to run sources in dbt.

**Screenshots**

![image](https://user-images.githubusercontent.com/2745502/181391848-7c2fc583-06be-44d9-8812-af867a7e3bc7.png)



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/fal/cli/selectors.py`
Content:
```
1 import itertools
2 import re
3 from dataclasses import dataclass
4 from typing import List, Optional, Union, Iterator
5 from fal.node_graph import NodeGraph
6 from faldbt.project import CompileArgs, FalDbt
7 from dbt.task.compile import CompileTask
8 from enum import Enum
9 from functools import reduce
10 import networkx as nx
11 from dbt.logger import GLOBAL_LOGGER as logger
12 
13 
14 class ExecutionPlan:
15     """
16     Represents a fal flow excution
17     """
18 
19     before_scripts: List[str]
20     dbt_models: List[str]
21     after_scripts: List[str]
22     project_name: str
23 
24     def __init__(self, unique_ids: List[str], project_name):
25         self.before_scripts = []
26         self.dbt_models = []
27         self.after_scripts = []
28         self.project_name = project_name
29         for id in unique_ids:
30             if _is_before_script(id):
31                 self.before_scripts.append(id)
32             elif _is_after_script(id):
33                 self.after_scripts.append(id)
34             else:
35                 self.dbt_models.append(id)
36 
37     @property
38     def nodes(self) -> List[str]:
39         return self.before_scripts + self.after_scripts + self.dbt_models
40 
41     @classmethod
42     def create_plan_from_graph(cls, parsed, nodeGraph: NodeGraph, fal_dbt: FalDbt):
43         """
44         Creates and ExecutionPlan from the cli arguments
45         """
46         unique_ids = list(nodeGraph.graph.nodes.keys())
47         ids_to_execute = []
48 
49         ids_to_exclude = []
50 
51         if parsed.select:
52             ids_to_execute = _filter_node_ids(
53                 unique_ids, fal_dbt, list(parsed.select), nodeGraph
54             )
55 
56         else:
57             ids_to_execute.extend(unique_ids)
58 
59         if "exclude" in parsed and parsed.exclude:
60             ids_to_exclude = _filter_node_ids(
61                 unique_ids, fal_dbt, list(parsed.exclude), nodeGraph
62             )
63 
64         ids_to_execute = [i for i in ids_to_execute if i not in ids_to_exclude]
65         return cls(list(set(ids_to_execute)), fal_dbt.project_name)
66 
67 
68 @dataclass
69 class SelectionUnion:
70     components: List[str]
71 
72 
73 @dataclass
74 class SelectionIntersection:
75     components: List[str]
76 
77 
78 def parse_union(
79     components: List[str],
80 ) -> SelectionUnion:
81     # Based on the original implemention at dbt-core.
82 
83     # turn ['a b', 'c'] -> ['a', 'b', 'c']
84     raw_specs = itertools.chain.from_iterable(r.split(OP_SET_UNION) for r in components)
85     union_components = []
86 
87     # ['a', 'b', 'c,d'] -> union('a', 'b', intersection('c', 'd'))
88     for raw_spec in raw_specs:
89         union_components.append(
90             SelectionIntersection(
91                 raw_spec.split(OP_SET_INTERSECTION),
92             )
93         )
94     return SelectionUnion(
95         union_components,
96     )
97 
98 
99 def _filter_node_ids(
100     unique_ids: List[str],
101     fal_dbt: FalDbt,
102     selectors: List[str],
103     nodeGraph: NodeGraph,
104 ) -> List[str]:
105     """Filter list of unique_ids according to a selector."""
106     output = set()
107 
108     union = parse_union(selectors)
109     for intersection in union.components:
110         try:
111             plan_outputs = [
112                 set(SelectorPlan(selector, unique_ids, fal_dbt).execute(nodeGraph))
113                 for selector in intersection.components
114                 if selector
115             ]
116         except nx.NetworkXError:
117             # When the user selects a non-existent node, don't fail immediately
118             # but rather just continue processing the rest of selectors.
119             plan_outputs = []
120 
121         if plan_outputs:
122             output |= set.intersection(*plan_outputs)
123 
124     return list(output)
125 
126 
127 def _get_children_with_parents(node_id: str, nodeGraph: NodeGraph) -> List[str]:
128     children = nodeGraph.get_descendants(node_id)
129     output = reduce(lambda l, ch: l + nodeGraph.get_ancestors(ch), children, children)
130 
131     output = list(set(output))
132 
133     return output
134 
135 
136 def _expand_script(script_name: str, unique_ids: List[str]) -> List[str]:
137     """
138     Expands the selected script name to unique id format.
139     for example [scripta.py] to [script.modelB.AFTER.scripta.py, script.modelA.BEFORE.scripta.py]
140     """
141 
142     def contains_script_name(id: str):
143         return script_name in id
144 
145     return list(filter(contains_script_name, unique_ids))
146 
147 
148 class SelectType(Enum):
149     MODEL = 1
150     SCRIPT = 2
151     COMPLEX = 3
152 
153 
154 @dataclass(init=False)
155 class SelectorPlan:
156     """
157     Represents a single selector, for example in the command
158 
159     fal flow run --select script.py+
160 
161     script.py+ is the SelectorPlan with needs_children attribute set to true
162     """
163 
164     unique_ids: List[str]
165     children: bool
166     children_levels: Optional[int]
167     children_with_parents: bool
168     parents: bool
169     parents_levels: Optional[int]
170     type: SelectType
171     raw: str
172 
173     def __init__(self, selector: str, unique_ids: List[str], fal_dbt: FalDbt):
174         self.raw = selector
175         self.children_with_parents = OP_CHILDREN_WITH_PARENTS.match(selector)
176         selector = OP_CHILDREN_WITH_PARENTS.rest(selector)
177 
178         self.parents = OP_PARENTS.match(selector)
179         self.parents_levels = OP_PARENTS.depth(selector)
180         selector = OP_PARENTS.rest(selector)
181 
182         self.children = OP_CHILDREN.match(selector)
183         self.children_levels = OP_CHILDREN.depth(selector)
184         selector = OP_CHILDREN.rest(selector)
185 
186         self.type = _to_select_type(selector)
187 
188         if self.type == SelectType.MODEL:
189             self.unique_ids = [f"model.{fal_dbt.project_name}.{selector}"]
190         elif self.type == SelectType.SCRIPT:
191             self.unique_ids = _expand_script(selector, unique_ids)
192         elif self.type == SelectType.COMPLEX:
193             self.unique_ids = unique_ids_from_complex_selector(selector, fal_dbt)
194 
195     def __post_init__(self):
196         if self.children and self.children_with_parents:
197             raise RuntimeError(
198                 f'Invalid node spec {self.raw} - "@" prefix and "+" suffix are incompatible'
199             )
200 
201     def execute(self, nodeGraph: NodeGraph) -> Iterator[str]:
202         for id in self.unique_ids:
203             yield id
204 
205             if self.children:
206                 if self.children_levels is None:
207                     children = nodeGraph.get_descendants(id)
208                 else:
209                     children = nodeGraph.get_successors(id, self.children_levels)
210                 yield from children
211 
212             if self.parents:
213                 if self.parents_levels is None:
214                     parents = nodeGraph.get_ancestors(id)
215                 else:
216                     parents = nodeGraph.get_predecessors(id, self.parents_levels)
217                 yield from parents
218 
219             if self.children_with_parents:
220                 ids = _get_children_with_parents(id, nodeGraph)
221                 yield from ids
222 
223 
224 def unique_ids_from_complex_selector(select, fal_dbt: FalDbt) -> List[str]:
225     args = CompileArgs(None, [select], [select], tuple(), fal_dbt._state, None)
226     compile_task = CompileTask(args, fal_dbt._config)
227     compile_task._runtime_initialize()
228     spec = compile_task.get_selection_spec()
229     graph = compile_task.get_node_selector().get_graph_queue(spec)
230     return list(graph.queued)
231 
232 
233 def _to_select_type(selector: str) -> SelectType:
234     if ":" in selector:
235         return SelectType.COMPLEX
236     else:
237         if _is_script_node(selector):
238             return SelectType.SCRIPT
239         else:
240             return SelectType.MODEL
241 
242 
243 def _is_script_node(node_name: str) -> bool:
244     return node_name.endswith(".py") or node_name.endswith(".ipynb")
245 
246 
247 class SelectorGraphOp:
248     _regex: re.Pattern
249 
250     def __init__(self, regex: re.Pattern):
251         self._regex = regex
252         assert (
253             "rest" in regex.groupindex
254         ), 'rest must be in regex. Use `re.compile("something(?P<rest>.*)")`'
255 
256     def _select(self, selector: str, group: Union[str, int]) -> Optional[str]:
257         match = self._regex.match(selector)
258         if match:
259             return match.group(group)
260 
261     def match(self, selector: str) -> bool:
262         return self._select(selector, 0) is not None
263 
264     def rest(self, selector: str) -> str:
265         rest = self._select(selector, "rest")
266         if rest is not None:
267             return rest
268         return selector
269 
270 
271 class SelectorGraphOpDepth(SelectorGraphOp):
272     def depth(self, selector: str) -> Optional[int]:
273         depth = self._select(selector, "depth")
274         if depth:
275             return int(depth)
276 
277 
278 # Graph operators from their regex Patterns
279 OP_CHILDREN_WITH_PARENTS = SelectorGraphOp(re.compile("^\\@(?P<rest>.*)"))
280 OP_PARENTS = SelectorGraphOpDepth(re.compile("^(?P<depth>\\d*)\\+(?P<rest>.*)"))
281 OP_CHILDREN = SelectorGraphOpDepth(re.compile("(?P<rest>.*)\\+(?P<depth>\\d*)$"))
282 
283 # Logic based set operators
284 OP_SET_UNION = " "
285 OP_SET_INTERSECTION = ","
286 
287 IS_BEFORE_SCRIPT_REGEX = re.compile("^script\\..+\\.BEFORE\\..+\\.(py|ipynb)$")
288 IS_AFTER_SCRIPT_REGEX = re.compile("^script\\..+\\.AFTER\\..+\\.(py|ipynb)")
289 
290 
291 def _is_before_script(id: str) -> bool:
292     return bool(IS_BEFORE_SCRIPT_REGEX.match(id))
293 
294 
295 def _is_after_script(id: str) -> bool:
296     return bool(IS_AFTER_SCRIPT_REGEX.match(id))
297
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/fal/cli/selectors.py b/src/fal/cli/selectors.py
--- a/src/fal/cli/selectors.py
+++ b/src/fal/cli/selectors.py
@@ -44,24 +44,23 @@
         Creates and ExecutionPlan from the cli arguments
         """
         unique_ids = list(nodeGraph.graph.nodes.keys())
-        ids_to_execute = []
-
-        ids_to_exclude = []
 
+        ids_to_execute = unique_ids
         if parsed.select:
             ids_to_execute = _filter_node_ids(
                 unique_ids, fal_dbt, list(parsed.select), nodeGraph
             )
 
-        else:
-            ids_to_execute.extend(unique_ids)
-
+        ids_to_exclude = []
         if "exclude" in parsed and parsed.exclude:
             ids_to_exclude = _filter_node_ids(
                 unique_ids, fal_dbt, list(parsed.exclude), nodeGraph
             )
 
         ids_to_execute = [i for i in ids_to_execute if i not in ids_to_exclude]
+
+        # Remove non-model nodes (sources, maybe more?) by making sure they are in the node_lookup dict
+        ids_to_execute = [i for i in ids_to_execute if i in nodeGraph.node_lookup]
         return cls(list(set(ids_to_execute)), fal_dbt.project_name)

{"golden_diff": "diff --git a/src/fal/cli/selectors.py b/src/fal/cli/selectors.py\n--- a/src/fal/cli/selectors.py\n+++ b/src/fal/cli/selectors.py\n@@ -44,24 +44,23 @@\n         Creates and ExecutionPlan from the cli arguments\n         \"\"\"\n         unique_ids = list(nodeGraph.graph.nodes.keys())\n-        ids_to_execute = []\n-\n-        ids_to_exclude = []\n \n+        ids_to_execute = unique_ids\n         if parsed.select:\n             ids_to_execute = _filter_node_ids(\n                 unique_ids, fal_dbt, list(parsed.select), nodeGraph\n             )\n \n-        else:\n-            ids_to_execute.extend(unique_ids)\n-\n+        ids_to_exclude = []\n         if \"exclude\" in parsed and parsed.exclude:\n             ids_to_exclude = _filter_node_ids(\n                 unique_ids, fal_dbt, list(parsed.exclude), nodeGraph\n             )\n \n         ids_to_execute = [i for i in ids_to_execute if i not in ids_to_exclude]\n+\n+        # Remove non-model nodes (sources, maybe more?) by making sure they are in the node_lookup dict\n+        ids_to_execute = [i for i in ids_to_execute if i in nodeGraph.node_lookup]\n         return cls(list(set(ids_to_execute)), fal_dbt.project_name)\n", "issue": "Stop trying to run sources in dbt runner\n<!-- *** Make sure you have searched for an existing bug report for this issue *** -->\r\n\r\n**Describe the bug**\r\nfal takes sources as valid nodes to run in dbt and tries to run them.\r\nIt mostly ends up in a no-op operation, but if the source shares name with a model it can run the wrong model.\r\n\r\n**Expected behavior**\r\nfal should not try to run sources in dbt.\r\n\r\n**Screenshots**\r\n\r\n![image](https://user-images.githubusercontent.com/2745502/181391848-7c2fc583-06be-44d9-8812-af867a7e3bc7.png)\r\n\r\n\n", "before_files": [{"content": "import itertools\nimport re\nfrom dataclasses import dataclass\nfrom typing import List, Optional, Union, Iterator\nfrom fal.node_graph import NodeGraph\nfrom faldbt.project import CompileArgs, FalDbt\nfrom dbt.task.compile import CompileTask\nfrom enum import Enum\nfrom functools import reduce\nimport networkx as nx\nfrom dbt.logger import GLOBAL_LOGGER as logger\n\n\nclass ExecutionPlan:\n    \"\"\"\n    Represents a fal flow excution\n    \"\"\"\n\n    before_scripts: List[str]\n    dbt_models: List[str]\n    after_scripts: List[str]\n    project_name: str\n\n    def __init__(self, unique_ids: List[str], project_name):\n        self.before_scripts = []\n        self.dbt_models = []\n        self.after_scripts = []\n        self.project_name = project_name\n        for id in unique_ids:\n            if _is_before_script(id):\n                self.before_scripts.append(id)\n            elif _is_after_script(id):\n                self.after_scripts.append(id)\n            else:\n                self.dbt_models.append(id)\n\n    @property\n    def nodes(self) -> List[str]:\n        return self.before_scripts + self.after_scripts + self.dbt_models\n\n    @classmethod\n    def create_plan_from_graph(cls, parsed, nodeGraph: NodeGraph, fal_dbt: FalDbt):\n        \"\"\"\n        Creates and ExecutionPlan from the cli arguments\n        \"\"\"\n        unique_ids = list(nodeGraph.graph.nodes.keys())\n        ids_to_execute = []\n\n        ids_to_exclude = []\n\n        if parsed.select:\n            ids_to_execute = _filter_node_ids(\n                unique_ids, fal_dbt, list(parsed.select), nodeGraph\n            )\n\n        else:\n            ids_to_execute.extend(unique_ids)\n\n        if \"exclude\" in parsed and parsed.exclude:\n            ids_to_exclude = _filter_node_ids(\n                unique_ids, fal_dbt, list(parsed.exclude), nodeGraph\n            )\n\n        ids_to_execute = [i for i in ids_to_execute if i not in ids_to_exclude]\n        return cls(list(set(ids_to_execute)), fal_dbt.project_name)\n\n\n@dataclass\nclass SelectionUnion:\n    components: List[str]\n\n\n@dataclass\nclass SelectionIntersection:\n    components: List[str]\n\n\ndef parse_union(\n    components: List[str],\n) -> SelectionUnion:\n    # Based on the original implemention at dbt-core.\n\n    # turn ['a b', 'c'] -> ['a', 'b', 'c']\n    raw_specs = itertools.chain.from_iterable(r.split(OP_SET_UNION) for r in components)\n    union_components = []\n\n    # ['a', 'b', 'c,d'] -> union('a', 'b', intersection('c', 'd'))\n    for raw_spec in raw_specs:\n        union_components.append(\n            SelectionIntersection(\n                raw_spec.split(OP_SET_INTERSECTION),\n            )\n        )\n    return SelectionUnion(\n        union_components,\n    )\n\n\ndef _filter_node_ids(\n    unique_ids: List[str],\n    fal_dbt: FalDbt,\n    selectors: List[str],\n    nodeGraph: NodeGraph,\n) -> List[str]:\n    \"\"\"Filter list of unique_ids according to a selector.\"\"\"\n    output = set()\n\n    union = parse_union(selectors)\n    for intersection in union.components:\n        try:\n            plan_outputs = [\n                set(SelectorPlan(selector, unique_ids, fal_dbt).execute(nodeGraph))\n                for selector in intersection.components\n                if selector\n            ]\n        except nx.NetworkXError:\n            # When the user selects a non-existent node, don't fail immediately\n            # but rather just continue processing the rest of selectors.\n            plan_outputs = []\n\n        if plan_outputs:\n            output |= set.intersection(*plan_outputs)\n\n    return list(output)\n\n\ndef _get_children_with_parents(node_id: str, nodeGraph: NodeGraph) -> List[str]:\n    children = nodeGraph.get_descendants(node_id)\n    output = reduce(lambda l, ch: l + nodeGraph.get_ancestors(ch), children, children)\n\n    output = list(set(output))\n\n    return output\n\n\ndef _expand_script(script_name: str, unique_ids: List[str]) -> List[str]:\n    \"\"\"\n    Expands the selected script name to unique id format.\n    for example [scripta.py] to [script.modelB.AFTER.scripta.py, script.modelA.BEFORE.scripta.py]\n    \"\"\"\n\n    def contains_script_name(id: str):\n        return script_name in id\n\n    return list(filter(contains_script_name, unique_ids))\n\n\nclass SelectType(Enum):\n    MODEL = 1\n    SCRIPT = 2\n    COMPLEX = 3\n\n\n@dataclass(init=False)\nclass SelectorPlan:\n    \"\"\"\n    Represents a single selector, for example in the command\n\n    fal flow run --select script.py+\n\n    script.py+ is the SelectorPlan with needs_children attribute set to true\n    \"\"\"\n\n    unique_ids: List[str]\n    children: bool\n    children_levels: Optional[int]\n    children_with_parents: bool\n    parents: bool\n    parents_levels: Optional[int]\n    type: SelectType\n    raw: str\n\n    def __init__(self, selector: str, unique_ids: List[str], fal_dbt: FalDbt):\n        self.raw = selector\n        self.children_with_parents = OP_CHILDREN_WITH_PARENTS.match(selector)\n        selector = OP_CHILDREN_WITH_PARENTS.rest(selector)\n\n        self.parents = OP_PARENTS.match(selector)\n        self.parents_levels = OP_PARENTS.depth(selector)\n        selector = OP_PARENTS.rest(selector)\n\n        self.children = OP_CHILDREN.match(selector)\n        self.children_levels = OP_CHILDREN.depth(selector)\n        selector = OP_CHILDREN.rest(selector)\n\n        self.type = _to_select_type(selector)\n\n        if self.type == SelectType.MODEL:\n            self.unique_ids = [f\"model.{fal_dbt.project_name}.{selector}\"]\n        elif self.type == SelectType.SCRIPT:\n            self.unique_ids = _expand_script(selector, unique_ids)\n        elif self.type == SelectType.COMPLEX:\n            self.unique_ids = unique_ids_from_complex_selector(selector, fal_dbt)\n\n    def __post_init__(self):\n        if self.children and self.children_with_parents:\n            raise RuntimeError(\n                f'Invalid node spec {self.raw} - \"@\" prefix and \"+\" suffix are incompatible'\n            )\n\n    def execute(self, nodeGraph: NodeGraph) -> Iterator[str]:\n        for id in self.unique_ids:\n            yield id\n\n            if self.children:\n                if self.children_levels is None:\n                    children = nodeGraph.get_descendants(id)\n                else:\n                    children = nodeGraph.get_successors(id, self.children_levels)\n                yield from children\n\n            if self.parents:\n                if self.parents_levels is None:\n                    parents = nodeGraph.get_ancestors(id)\n                else:\n                    parents = nodeGraph.get_predecessors(id, self.parents_levels)\n                yield from parents\n\n            if self.children_with_parents:\n                ids = _get_children_with_parents(id, nodeGraph)\n                yield from ids\n\n\ndef unique_ids_from_complex_selector(select, fal_dbt: FalDbt) -> List[str]:\n    args = CompileArgs(None, [select], [select], tuple(), fal_dbt._state, None)\n    compile_task = CompileTask(args, fal_dbt._config)\n    compile_task._runtime_initialize()\n    spec = compile_task.get_selection_spec()\n    graph = compile_task.get_node_selector().get_graph_queue(spec)\n    return list(graph.queued)\n\n\ndef _to_select_type(selector: str) -> SelectType:\n    if \":\" in selector:\n        return SelectType.COMPLEX\n    else:\n        if _is_script_node(selector):\n            return SelectType.SCRIPT\n        else:\n            return SelectType.MODEL\n\n\ndef _is_script_node(node_name: str) -> bool:\n    return node_name.endswith(\".py\") or node_name.endswith(\".ipynb\")\n\n\nclass SelectorGraphOp:\n    _regex: re.Pattern\n\n    def __init__(self, regex: re.Pattern):\n        self._regex = regex\n        assert (\n            \"rest\" in regex.groupindex\n        ), 'rest must be in regex. Use `re.compile(\"something(?P<rest>.*)\")`'\n\n    def _select(self, selector: str, group: Union[str, int]) -> Optional[str]:\n        match = self._regex.match(selector)\n        if match:\n            return match.group(group)\n\n    def match(self, selector: str) -> bool:\n        return self._select(selector, 0) is not None\n\n    def rest(self, selector: str) -> str:\n        rest = self._select(selector, \"rest\")\n        if rest is not None:\n            return rest\n        return selector\n\n\nclass SelectorGraphOpDepth(SelectorGraphOp):\n    def depth(self, selector: str) -> Optional[int]:\n        depth = self._select(selector, \"depth\")\n        if depth:\n            return int(depth)\n\n\n# Graph operators from their regex Patterns\nOP_CHILDREN_WITH_PARENTS = SelectorGraphOp(re.compile(\"^\\\\@(?P<rest>.*)\"))\nOP_PARENTS = SelectorGraphOpDepth(re.compile(\"^(?P<depth>\\\\d*)\\\\+(?P<rest>.*)\"))\nOP_CHILDREN = SelectorGraphOpDepth(re.compile(\"(?P<rest>.*)\\\\+(?P<depth>\\\\d*)$\"))\n\n# Logic based set operators\nOP_SET_UNION = \" \"\nOP_SET_INTERSECTION = \",\"\n\nIS_BEFORE_SCRIPT_REGEX = re.compile(\"^script\\\\..+\\\\.BEFORE\\\\..+\\\\.(py|ipynb)$\")\nIS_AFTER_SCRIPT_REGEX = re.compile(\"^script\\\\..+\\\\.AFTER\\\\..+\\\\.(py|ipynb)\")\n\n\ndef _is_before_script(id: str) -> bool:\n    return bool(IS_BEFORE_SCRIPT_REGEX.match(id))\n\n\ndef _is_after_script(id: str) -> bool:\n    return bool(IS_AFTER_SCRIPT_REGEX.match(id))\n", "path": "src/fal/cli/selectors.py"}], "after_files": [{"content": "import itertools\nimport re\nfrom dataclasses import dataclass\nfrom typing import List, Optional, Union, Iterator\nfrom fal.node_graph import NodeGraph\nfrom faldbt.project import CompileArgs, FalDbt\nfrom dbt.task.compile import CompileTask\nfrom enum import Enum\nfrom functools import reduce\nimport networkx as nx\nfrom dbt.logger import GLOBAL_LOGGER as logger\n\n\nclass ExecutionPlan:\n    \"\"\"\n    Represents a fal flow excution\n    \"\"\"\n\n    before_scripts: List[str]\n    dbt_models: List[str]\n    after_scripts: List[str]\n    project_name: str\n\n    def __init__(self, unique_ids: List[str], project_name):\n        self.before_scripts = []\n        self.dbt_models = []\n        self.after_scripts = []\n        self.project_name = project_name\n        for id in unique_ids:\n            if _is_before_script(id):\n                self.before_scripts.append(id)\n            elif _is_after_script(id):\n                self.after_scripts.append(id)\n            else:\n                self.dbt_models.append(id)\n\n    @property\n    def nodes(self) -> List[str]:\n        return self.before_scripts + self.after_scripts + self.dbt_models\n\n    @classmethod\n    def create_plan_from_graph(cls, parsed, nodeGraph: NodeGraph, fal_dbt: FalDbt):\n        \"\"\"\n        Creates and ExecutionPlan from the cli arguments\n        \"\"\"\n        unique_ids = list(nodeGraph.graph.nodes.keys())\n\n        ids_to_execute = unique_ids\n        if parsed.select:\n            ids_to_execute = _filter_node_ids(\n                unique_ids, fal_dbt, list(parsed.select), nodeGraph\n            )\n\n        ids_to_exclude = []\n        if \"exclude\" in parsed and parsed.exclude:\n            ids_to_exclude = _filter_node_ids(\n                unique_ids, fal_dbt, list(parsed.exclude), nodeGraph\n            )\n\n        ids_to_execute = [i for i in ids_to_execute if i not in ids_to_exclude]\n\n        # Remove non-model nodes (sources, maybe more?) by making sure they are in the node_lookup dict\n        ids_to_execute = [i for i in ids_to_execute if i in nodeGraph.node_lookup]\n        return cls(list(set(ids_to_execute)), fal_dbt.project_name)\n\n\n@dataclass\nclass SelectionUnion:\n    components: List[str]\n\n\n@dataclass\nclass SelectionIntersection:\n    components: List[str]\n\n\ndef parse_union(\n    components: List[str],\n) -> SelectionUnion:\n    # Based on the original implemention at dbt-core.\n\n    # turn ['a b', 'c'] -> ['a', 'b', 'c']\n    raw_specs = itertools.chain.from_iterable(r.split(OP_SET_UNION) for r in components)\n    union_components = []\n\n    # ['a', 'b', 'c,d'] -> union('a', 'b', intersection('c', 'd'))\n    for raw_spec in raw_specs:\n        union_components.append(\n            SelectionIntersection(\n                raw_spec.split(OP_SET_INTERSECTION),\n            )\n        )\n    return SelectionUnion(\n        union_components,\n    )\n\n\ndef _filter_node_ids(\n    unique_ids: List[str],\n    fal_dbt: FalDbt,\n    selectors: List[str],\n    nodeGraph: NodeGraph,\n) -> List[str]:\n    \"\"\"Filter list of unique_ids according to a selector.\"\"\"\n    output = set()\n\n    union = parse_union(selectors)\n    for intersection in union.components:\n        try:\n            plan_outputs = [\n                set(SelectorPlan(selector, unique_ids, fal_dbt).execute(nodeGraph))\n                for selector in intersection.components\n                if selector\n            ]\n        except nx.NetworkXError:\n            # When the user selects a non-existent node, don't fail immediately\n            # but rather just continue processing the rest of selectors.\n            plan_outputs = []\n\n        if plan_outputs:\n            output |= set.intersection(*plan_outputs)\n\n    return list(output)\n\n\ndef _get_children_with_parents(node_id: str, nodeGraph: NodeGraph) -> List[str]:\n    children = nodeGraph.get_descendants(node_id)\n    output = reduce(lambda l, ch: l + nodeGraph.get_ancestors(ch), children, children)\n\n    output = list(set(output))\n\n    return output\n\n\ndef _expand_script(script_name: str, unique_ids: List[str]) -> List[str]:\n    \"\"\"\n    Expands the selected script name to unique id format.\n    for example [scripta.py] to [script.modelB.AFTER.scripta.py, script.modelA.BEFORE.scripta.py]\n    \"\"\"\n\n    def contains_script_name(id: str):\n        return script_name in id\n\n    return list(filter(contains_script_name, unique_ids))\n\n\nclass SelectType(Enum):\n    MODEL = 1\n    SCRIPT = 2\n    COMPLEX = 3\n\n\n@dataclass(init=False)\nclass SelectorPlan:\n    \"\"\"\n    Represents a single selector, for example in the command\n\n    fal flow run --select script.py+\n\n    script.py+ is the SelectorPlan with needs_children attribute set to true\n    \"\"\"\n\n    unique_ids: List[str]\n    children: bool\n    children_levels: Optional[int]\n    children_with_parents: bool\n    parents: bool\n    parents_levels: Optional[int]\n    type: SelectType\n    raw: str\n\n    def __init__(self, selector: str, unique_ids: List[str], fal_dbt: FalDbt):\n        self.raw = selector\n        self.children_with_parents = OP_CHILDREN_WITH_PARENTS.match(selector)\n        selector = OP_CHILDREN_WITH_PARENTS.rest(selector)\n\n        self.parents = OP_PARENTS.match(selector)\n        self.parents_levels = OP_PARENTS.depth(selector)\n        selector = OP_PARENTS.rest(selector)\n\n        self.children = OP_CHILDREN.match(selector)\n        self.children_levels = OP_CHILDREN.depth(selector)\n        selector = OP_CHILDREN.rest(selector)\n\n        self.type = _to_select_type(selector)\n\n        if self.type == SelectType.MODEL:\n            self.unique_ids = [f\"model.{fal_dbt.project_name}.{selector}\"]\n        elif self.type == SelectType.SCRIPT:\n            self.unique_ids = _expand_script(selector, unique_ids)\n        elif self.type == SelectType.COMPLEX:\n            self.unique_ids = unique_ids_from_complex_selector(selector, fal_dbt)\n\n    def __post_init__(self):\n        if self.children and self.children_with_parents:\n            raise RuntimeError(\n                f'Invalid node spec {self.raw} - \"@\" prefix and \"+\" suffix are incompatible'\n            )\n\n    def execute(self, nodeGraph: NodeGraph) -> Iterator[str]:\n        for id in self.unique_ids:\n            yield id\n\n            if self.children:\n                if self.children_levels is None:\n                    children = nodeGraph.get_descendants(id)\n                else:\n                    children = nodeGraph.get_successors(id, self.children_levels)\n                yield from children\n\n            if self.parents:\n                if self.parents_levels is None:\n                    parents = nodeGraph.get_ancestors(id)\n                else:\n                    parents = nodeGraph.get_predecessors(id, self.parents_levels)\n                yield from parents\n\n            if self.children_with_parents:\n                ids = _get_children_with_parents(id, nodeGraph)\n                yield from ids\n\n\ndef unique_ids_from_complex_selector(select, fal_dbt: FalDbt) -> List[str]:\n    args = CompileArgs(None, [select], [select], tuple(), fal_dbt._state, None)\n    compile_task = CompileTask(args, fal_dbt._config)\n    compile_task._runtime_initialize()\n    spec = compile_task.get_selection_spec()\n    graph = compile_task.get_node_selector().get_graph_queue(spec)\n    return list(graph.queued)\n\n\ndef _to_select_type(selector: str) -> SelectType:\n    if \":\" in selector:\n        return SelectType.COMPLEX\n    else:\n        if _is_script_node(selector):\n            return SelectType.SCRIPT\n        else:\n            return SelectType.MODEL\n\n\ndef _is_script_node(node_name: str) -> bool:\n    return node_name.endswith(\".py\") or node_name.endswith(\".ipynb\")\n\n\nclass SelectorGraphOp:\n    _regex: re.Pattern\n\n    def __init__(self, regex: re.Pattern):\n        self._regex = regex\n        assert (\n            \"rest\" in regex.groupindex\n        ), 'rest must be in regex. Use `re.compile(\"something(?P<rest>.*)\")`'\n\n    def _select(self, selector: str, group: Union[str, int]) -> Optional[str]:\n        match = self._regex.match(selector)\n        if match:\n            return match.group(group)\n\n    def match(self, selector: str) -> bool:\n        return self._select(selector, 0) is not None\n\n    def rest(self, selector: str) -> str:\n        rest = self._select(selector, \"rest\")\n        if rest is not None:\n            return rest\n        return selector\n\n\nclass SelectorGraphOpDepth(SelectorGraphOp):\n    def depth(self, selector: str) -> Optional[int]:\n        depth = self._select(selector, \"depth\")\n        if depth:\n            return int(depth)\n\n\n# Graph operators from their regex Patterns\nOP_CHILDREN_WITH_PARENTS = SelectorGraphOp(re.compile(\"^\\\\@(?P<rest>.*)\"))\nOP_PARENTS = SelectorGraphOpDepth(re.compile(\"^(?P<depth>\\\\d*)\\\\+(?P<rest>.*)\"))\nOP_CHILDREN = SelectorGraphOpDepth(re.compile(\"(?P<rest>.*)\\\\+(?P<depth>\\\\d*)$\"))\n\n# Logic based set operators\nOP_SET_UNION = \" \"\nOP_SET_INTERSECTION = \",\"\n\nIS_BEFORE_SCRIPT_REGEX = re.compile(\"^script\\\\..+\\\\.BEFORE\\\\..+\\\\.(py|ipynb)$\")\nIS_AFTER_SCRIPT_REGEX = re.compile(\"^script\\\\..+\\\\.AFTER\\\\..+\\\\.(py|ipynb)\")\n\n\ndef _is_before_script(id: str) -> bool:\n    return bool(IS_BEFORE_SCRIPT_REGEX.match(id))\n\n\ndef _is_after_script(id: str) -> bool:\n    return bool(IS_AFTER_SCRIPT_REGEX.match(id))\n", "path": "src/fal/cli/selectors.py"}]}