We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add the dbt target.name as a property on the context object
**Context**
I am sending slack messages and to keep things simple I have opted for one channel. (for some given set of models, that's handled through meta anyway i.e. slack_channel_id). 

However, doing this - it's hard to tell what the given environment was if you use one channel for all dbt invocations.

**Describe the solution you'd like**
I would like to be able to use context to access  dbt properties such as the target.name e.g.

```
    context.target.name # dev, test, prod, etc.
```
**Describe alternatives you've considered**
This can be worked around by injecting the {{ target.name }} as a meta field to a model.yml e.g:

```
version: 2

models:
  - name: my_funky_model
    meta:
      SLACK_CHANNEL_ID: XXXXXXXXXX
      TARGET: "{{ target.name }}"
      fal:
        scripts:
          - fal_scripts/notify_slack.py
```

However this does seem redundant/laborious to add to all .yml definitions 

--- 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/fal_script.py`
Content:
```
1 import os
2 import json
3 from typing import Dict, Any, List, Optional, Union, Callable
4 from pathlib import Path
5 from functools import partial
6 from dataclasses import dataclass, field
7 from deprecation import deprecated
8 
9 from faldbt.parse import normalize_path
10 from faldbt.project import DbtModel, FalDbt
11 
12 from dbt.contracts.results import RunStatus
13 from dbt.config.runtime import RuntimeConfig
14 from fal.logger import LOGGER
15 
16 from dbt.contracts.graph.parsed import ColumnInfo
17 
18 
19 class Hook:
20     path: str
21     arguments: Dict[str, Any]
22 
23 
24 @dataclass
25 class LocalHook(Hook):
26     path: str
27     arguments: Dict[str, Any] = field(default_factory=dict)
28 
29 
30 @dataclass
31 class IsolatedHook(Hook):
32     path: str
33     environment_name: str
34     arguments: Dict[str, Any] = field(default_factory=dict)
35 
36 
37 def _is_local_environment(environment_name: str) -> None:
38     return environment_name == "local"
39 
40 
41 def create_hook(raw_hook: Any, default_environment_name: Optional[str] = None) -> Hook:
42     if isinstance(raw_hook, str):
43         raw_hook = {"path": raw_hook}
44 
45     if not isinstance(raw_hook, dict):
46         raise ValueError(f"Unrecognized hook value: {raw_hook}")
47 
48     if "path" not in raw_hook:
49         raise ValueError(f"A hook must specify path.")
50 
51     environment_name = raw_hook.get("environment", default_environment_name)
52     if environment_name and not _is_local_environment(environment_name):
53         return IsolatedHook(
54             raw_hook["path"],
55             environment_name,
56             raw_hook.get("with", {}),
57         )
58     else:
59         return LocalHook(raw_hook["path"], raw_hook.get("with", {}))
60 
61 
62 @dataclass
63 class CurrentAdapterResponse:
64     message: str
65     code: Optional[str]
66     rows_affected: Optional[int]
67 
68 
69 @dataclass
70 class CurrentModel:
71     name: str
72     alias: str
73     status: RunStatus
74     columns: Dict[str, ColumnInfo]
75     tests: List[Any]
76     meta: Dict[Any, Any]
77     adapter_response: Optional[CurrentAdapterResponse]
78 
79 
80 @dataclass
81 class CurrentTest:
82     name: str
83     model_name: str
84     column: str
85     status: str
86 
87     @property
88     @deprecated(details="Use 'model_name' instead")
89     def modelname(self):
90         return self.model_name
91 
92 
93 @dataclass
94 class ContextConfig:
95     target_path: Path
96 
97     def __init__(self, config: RuntimeConfig):
98         self.target_path = Path(
99             os.path.realpath(os.path.join(config.project_root, config.target_path))
100         )
101 
102 
103 @dataclass
104 class Context:
105     current_model: Union[CurrentModel, None]
106     config: ContextConfig
107     _arguments: Optional[Dict[str, Any]] = field(repr=False, default=None)
108 
109     @property
110     def arguments(self) -> Dict[str, Any]:
111         if self._arguments is None:
112             raise ValueError(
113                 "'context.arguments' is only accessible from hooks, "
114                 "not from scripts/models"
115             )
116         return self._arguments
117 
118 
119 @dataclass(frozen=True, init=False)
120 class FalScript:
121     model: Optional[DbtModel]
122     path: Path
123     faldbt: FalDbt
124     hook_arguments: Optional[Dict[str, Any]]
125     is_hook: bool
126 
127     def __init__(
128         self,
129         faldbt: FalDbt,
130         model: Optional[DbtModel],
131         path: str,
132         hook_arguments: Optional[Dict[str, Any]] = None,
133         is_hook: bool = False,
134     ):
135         # Necessary because of frozen=True
136         object.__setattr__(self, "model", model)
137         object.__setattr__(self, "path", normalize_path(faldbt.scripts_dir, path))
138         object.__setattr__(self, "faldbt", faldbt)
139         object.__setattr__(self, "hook_arguments", hook_arguments)
140         object.__setattr__(self, "is_hook", is_hook)
141 
142     @classmethod
143     def from_hook(cls, faldbt: FalDbt, model: DbtModel, hook: Hook):
144         """
145         Creates a FalScript from a hook
146         """
147         assert isinstance(hook, LocalHook)
148         return cls(
149             faldbt=faldbt,
150             model=model,
151             path=hook.path,
152             hook_arguments=hook.arguments,
153             is_hook=True,
154         )
155 
156     @classmethod
157     def model_script(cls, faldbt: FalDbt, model: DbtModel):
158         script = FalScript(faldbt, model, "")
159         # HACK: Set the script path specially for this case
160         object.__setattr__(script, "path", model.python_model)
161         return script
162 
163     def exec(self):
164         """
165         Executes the script
166         """
167         # Enable local imports
168         try:
169             source_code = python_from_file(self.path)
170             program = compile(source_code, self.path, "exec")
171 
172             exec_globals = {
173                 "__name__": "__main__",
174                 "context": self._build_script_context(),
175                 "ref": self.faldbt.ref,
176                 "source": self.faldbt.source,
177                 "write_to_firestore": self.faldbt.write_to_firestore,
178                 "list_models": self.faldbt.list_models,
179                 "list_models_ids": self.faldbt.list_models_ids,
180                 "list_sources": self.faldbt.list_sources,
181                 "list_features": self.faldbt.list_features,
182                 "el": self.faldbt.el,
183                 "execute_sql": self.faldbt.execute_sql,
184             }
185 
186             if not self.is_hook:
187                 exec_globals["write_to_source"] = self.faldbt.write_to_source
188 
189                 if self.model is not None:
190                     # Hard-wire the model
191                     exec_globals["write_to_model"] = partial(
192                         self.faldbt.write_to_model,
193                         target_1=self.model.name,
194                         target_2=None,
195                     )
196 
197             else:
198                 exec_globals["write_to_source"] = _not_allowed_function_maker(
199                     "write_to_source"
200                 )
201                 exec_globals["write_to_model"] = _not_allowed_function_maker(
202                     "write_to_model"
203                 )
204             exec(program, exec_globals)
205         finally:
206             pass
207 
208     @property
209     def relative_path(self):
210         if self.is_model:
211             return self.path.relative_to(self.faldbt.project_dir)
212         else:
213             return self.path.relative_to(self.faldbt.scripts_dir)
214 
215     @property
216     def id(self):
217         if self.is_model:
218             return f"(model: {self.relative_path})"
219         else:
220             return f"({self.model_name}, {self.relative_path})"
221 
222     @property
223     def is_global(self):
224         return self.model is None
225 
226     @property
227     def is_model(self):
228         if self.model is not None and self.model.python_model is not None:
229             return self.model.python_model == self.path
230 
231     @property
232     def model_name(self):
233         return "<GLOBAL>" if self.is_global else self.model.name  # type: ignore
234 
235     def _build_script_context(self) -> Context:
236         context_config = ContextConfig(self.faldbt._config)
237         if self.is_global:
238             return Context(current_model=None, config=context_config)
239 
240         model: DbtModel = self.model  # type: ignore
241 
242         meta = model.meta or {}
243         _del_key(meta, self.faldbt.keyword)
244 
245         tests = _process_tests(model.tests)
246 
247         current_adapter_response = None
248         if model.adapter_response:
249             current_adapter_response = CurrentAdapterResponse(
250                 message=str(model.adapter_response),
251                 code=model.adapter_response.code,
252                 rows_affected=model.adapter_response.rows_affected,
253             )
254 
255         current_model = CurrentModel(
256             name=model.name,
257             alias=model.alias,
258             status=model.status,
259             columns=model.columns,
260             tests=tests,
261             meta=meta,
262             adapter_response=current_adapter_response,
263         )
264 
265         return Context(
266             current_model=current_model,
267             config=context_config,
268             _arguments=self.hook_arguments,
269         )
270 
271 
272 def _del_key(dict: Dict[str, Any], key: str):
273     try:
274         del dict[key]
275     except KeyError:
276         pass
277 
278 
279 def _process_tests(tests: List[Any]):
280     return list(
281         map(
282             lambda test: CurrentTest(
283                 name=test.name,
284                 column=test.column,
285                 status=test.status,
286                 model_name=test.model,
287             ),
288             tests,
289         )
290     )
291 
292 
293 def python_from_file(path: Path) -> str:
294     with open(path) as file:
295         raw_source_code = file.read()
296         if path.suffix == ".ipynb":
297             raw_source_code = _process_ipynb(raw_source_code)
298     return raw_source_code
299 
300 
301 def _process_ipynb(raw_source_code: str) -> str:
302     def strip_magic(source: List[str]) -> List[str]:
303         NOTEBOOK_LIB = "faldbt.magics"
304         return [item for item in source if item[0] != "%" and NOTEBOOK_LIB not in item]
305 
306     ipynb_struct = json.loads(raw_source_code)
307 
308     script_list = []
309     for cell in ipynb_struct["cells"]:
310         if cell["cell_type"] == "code":
311             source = strip_magic(cell["source"])
312             script_list.append("".join(source))
313 
314     joined_script = "\n #cell \n".join(script_list)
315 
316     LOGGER.debug(f"Joined .ipynb cells to:\n{joined_script}")
317 
318     return joined_script
319 
320 
321 def _not_allowed_function_maker(function_name: str) -> Callable[[Any], None]:
322     def not_allowed_function(*args, **kwargs):
323         raise Exception(
324             (
325                 f"{function_name} is not allowed in hooks."
326                 " Consider using a Python model."
327             )
328         )
329 
330     return not_allowed_function
331
```
--- 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/fal_script.py b/src/fal/fal_script.py
--- a/src/fal/fal_script.py
+++ b/src/fal/fal_script.py
@@ -99,11 +99,21 @@
             os.path.realpath(os.path.join(config.project_root, config.target_path))
         )
 
+@dataclass
+class ContextTarget:
+    def __init__(self, config: RuntimeConfig):
+        self.profile_name = config.profile_name
+        self.name = config.target_name
+        self.threads = config.threads
+        self.type = config.credentials.type
+        self.database = config.credentials.database
+        self.schema = config.credentials.schema
 
 @dataclass
 class Context:
     current_model: Union[CurrentModel, None]
     config: ContextConfig
+    target: ContextTarget
     _arguments: Optional[Dict[str, Any]] = field(repr=False, default=None)
 
     @property
@@ -233,9 +243,12 @@
         return "<GLOBAL>" if self.is_global else self.model.name  # type: ignore
 
     def _build_script_context(self) -> Context:
-        context_config = ContextConfig(self.faldbt._config)
+        config: RuntimeConfig = self.faldbt._config
+        context_config = ContextConfig(config)
+        target = ContextTarget(config)
+
         if self.is_global:
-            return Context(current_model=None, config=context_config)
+            return Context(current_model=None, target=target, config=context_config)
 
         model: DbtModel = self.model  # type: ignore
 
@@ -264,6 +277,7 @@
 
         return Context(
             current_model=current_model,
+            target=target,
             config=context_config,
             _arguments=self.hook_arguments,
         )

{"golden_diff": "diff --git a/src/fal/fal_script.py b/src/fal/fal_script.py\n--- a/src/fal/fal_script.py\n+++ b/src/fal/fal_script.py\n@@ -99,11 +99,21 @@\n             os.path.realpath(os.path.join(config.project_root, config.target_path))\n         )\n \n+@dataclass\n+class ContextTarget:\n+    def __init__(self, config: RuntimeConfig):\n+        self.profile_name = config.profile_name\n+        self.name = config.target_name\n+        self.threads = config.threads\n+        self.type = config.credentials.type\n+        self.database = config.credentials.database\n+        self.schema = config.credentials.schema\n \n @dataclass\n class Context:\n     current_model: Union[CurrentModel, None]\n     config: ContextConfig\n+    target: ContextTarget\n     _arguments: Optional[Dict[str, Any]] = field(repr=False, default=None)\n \n     @property\n@@ -233,9 +243,12 @@\n         return \"<GLOBAL>\" if self.is_global else self.model.name  # type: ignore\n \n     def _build_script_context(self) -> Context:\n-        context_config = ContextConfig(self.faldbt._config)\n+        config: RuntimeConfig = self.faldbt._config\n+        context_config = ContextConfig(config)\n+        target = ContextTarget(config)\n+\n         if self.is_global:\n-            return Context(current_model=None, config=context_config)\n+            return Context(current_model=None, target=target, config=context_config)\n \n         model: DbtModel = self.model  # type: ignore\n \n@@ -264,6 +277,7 @@\n \n         return Context(\n             current_model=current_model,\n+            target=target,\n             config=context_config,\n             _arguments=self.hook_arguments,\n         )\n", "issue": "Add the dbt target.name as a property on the context object\n**Context**\r\nI am sending slack messages and to keep things simple I have opted for one channel. (for some given set of models, that's handled through meta anyway i.e. slack_channel_id). \r\n\r\nHowever, doing this - it's hard to tell what the given environment was if you use one channel for all dbt invocations.\r\n\r\n**Describe the solution you'd like**\r\nI would like to be able to use context to access  dbt properties such as the target.name e.g.\r\n\r\n```\r\n    context.target.name # dev, test, prod, etc.\r\n```\r\n**Describe alternatives you've considered**\r\nThis can be worked around by injecting the {{ target.name }} as a meta field to a model.yml e.g:\r\n\r\n```\r\nversion: 2\r\n\r\nmodels:\r\n  - name: my_funky_model\r\n    meta:\r\n      SLACK_CHANNEL_ID: XXXXXXXXXX\r\n      TARGET: \"{{ target.name }}\"\r\n      fal:\r\n        scripts:\r\n          - fal_scripts/notify_slack.py\r\n```\r\n\r\nHowever this does seem redundant/laborious to add to all .yml definitions \n", "before_files": [{"content": "import os\nimport json\nfrom typing import Dict, Any, List, Optional, Union, Callable\nfrom pathlib import Path\nfrom functools import partial\nfrom dataclasses import dataclass, field\nfrom deprecation import deprecated\n\nfrom faldbt.parse import normalize_path\nfrom faldbt.project import DbtModel, FalDbt\n\nfrom dbt.contracts.results import RunStatus\nfrom dbt.config.runtime import RuntimeConfig\nfrom fal.logger import LOGGER\n\nfrom dbt.contracts.graph.parsed import ColumnInfo\n\n\nclass Hook:\n    path: str\n    arguments: Dict[str, Any]\n\n\n@dataclass\nclass LocalHook(Hook):\n    path: str\n    arguments: Dict[str, Any] = field(default_factory=dict)\n\n\n@dataclass\nclass IsolatedHook(Hook):\n    path: str\n    environment_name: str\n    arguments: Dict[str, Any] = field(default_factory=dict)\n\n\ndef _is_local_environment(environment_name: str) -> None:\n    return environment_name == \"local\"\n\n\ndef create_hook(raw_hook: Any, default_environment_name: Optional[str] = None) -> Hook:\n    if isinstance(raw_hook, str):\n        raw_hook = {\"path\": raw_hook}\n\n    if not isinstance(raw_hook, dict):\n        raise ValueError(f\"Unrecognized hook value: {raw_hook}\")\n\n    if \"path\" not in raw_hook:\n        raise ValueError(f\"A hook must specify path.\")\n\n    environment_name = raw_hook.get(\"environment\", default_environment_name)\n    if environment_name and not _is_local_environment(environment_name):\n        return IsolatedHook(\n            raw_hook[\"path\"],\n            environment_name,\n            raw_hook.get(\"with\", {}),\n        )\n    else:\n        return LocalHook(raw_hook[\"path\"], raw_hook.get(\"with\", {}))\n\n\n@dataclass\nclass CurrentAdapterResponse:\n    message: str\n    code: Optional[str]\n    rows_affected: Optional[int]\n\n\n@dataclass\nclass CurrentModel:\n    name: str\n    alias: str\n    status: RunStatus\n    columns: Dict[str, ColumnInfo]\n    tests: List[Any]\n    meta: Dict[Any, Any]\n    adapter_response: Optional[CurrentAdapterResponse]\n\n\n@dataclass\nclass CurrentTest:\n    name: str\n    model_name: str\n    column: str\n    status: str\n\n    @property\n    @deprecated(details=\"Use 'model_name' instead\")\n    def modelname(self):\n        return self.model_name\n\n\n@dataclass\nclass ContextConfig:\n    target_path: Path\n\n    def __init__(self, config: RuntimeConfig):\n        self.target_path = Path(\n            os.path.realpath(os.path.join(config.project_root, config.target_path))\n        )\n\n\n@dataclass\nclass Context:\n    current_model: Union[CurrentModel, None]\n    config: ContextConfig\n    _arguments: Optional[Dict[str, Any]] = field(repr=False, default=None)\n\n    @property\n    def arguments(self) -> Dict[str, Any]:\n        if self._arguments is None:\n            raise ValueError(\n                \"'context.arguments' is only accessible from hooks, \"\n                \"not from scripts/models\"\n            )\n        return self._arguments\n\n\n@dataclass(frozen=True, init=False)\nclass FalScript:\n    model: Optional[DbtModel]\n    path: Path\n    faldbt: FalDbt\n    hook_arguments: Optional[Dict[str, Any]]\n    is_hook: bool\n\n    def __init__(\n        self,\n        faldbt: FalDbt,\n        model: Optional[DbtModel],\n        path: str,\n        hook_arguments: Optional[Dict[str, Any]] = None,\n        is_hook: bool = False,\n    ):\n        # Necessary because of frozen=True\n        object.__setattr__(self, \"model\", model)\n        object.__setattr__(self, \"path\", normalize_path(faldbt.scripts_dir, path))\n        object.__setattr__(self, \"faldbt\", faldbt)\n        object.__setattr__(self, \"hook_arguments\", hook_arguments)\n        object.__setattr__(self, \"is_hook\", is_hook)\n\n    @classmethod\n    def from_hook(cls, faldbt: FalDbt, model: DbtModel, hook: Hook):\n        \"\"\"\n        Creates a FalScript from a hook\n        \"\"\"\n        assert isinstance(hook, LocalHook)\n        return cls(\n            faldbt=faldbt,\n            model=model,\n            path=hook.path,\n            hook_arguments=hook.arguments,\n            is_hook=True,\n        )\n\n    @classmethod\n    def model_script(cls, faldbt: FalDbt, model: DbtModel):\n        script = FalScript(faldbt, model, \"\")\n        # HACK: Set the script path specially for this case\n        object.__setattr__(script, \"path\", model.python_model)\n        return script\n\n    def exec(self):\n        \"\"\"\n        Executes the script\n        \"\"\"\n        # Enable local imports\n        try:\n            source_code = python_from_file(self.path)\n            program = compile(source_code, self.path, \"exec\")\n\n            exec_globals = {\n                \"__name__\": \"__main__\",\n                \"context\": self._build_script_context(),\n                \"ref\": self.faldbt.ref,\n                \"source\": self.faldbt.source,\n                \"write_to_firestore\": self.faldbt.write_to_firestore,\n                \"list_models\": self.faldbt.list_models,\n                \"list_models_ids\": self.faldbt.list_models_ids,\n                \"list_sources\": self.faldbt.list_sources,\n                \"list_features\": self.faldbt.list_features,\n                \"el\": self.faldbt.el,\n                \"execute_sql\": self.faldbt.execute_sql,\n            }\n\n            if not self.is_hook:\n                exec_globals[\"write_to_source\"] = self.faldbt.write_to_source\n\n                if self.model is not None:\n                    # Hard-wire the model\n                    exec_globals[\"write_to_model\"] = partial(\n                        self.faldbt.write_to_model,\n                        target_1=self.model.name,\n                        target_2=None,\n                    )\n\n            else:\n                exec_globals[\"write_to_source\"] = _not_allowed_function_maker(\n                    \"write_to_source\"\n                )\n                exec_globals[\"write_to_model\"] = _not_allowed_function_maker(\n                    \"write_to_model\"\n                )\n            exec(program, exec_globals)\n        finally:\n            pass\n\n    @property\n    def relative_path(self):\n        if self.is_model:\n            return self.path.relative_to(self.faldbt.project_dir)\n        else:\n            return self.path.relative_to(self.faldbt.scripts_dir)\n\n    @property\n    def id(self):\n        if self.is_model:\n            return f\"(model: {self.relative_path})\"\n        else:\n            return f\"({self.model_name}, {self.relative_path})\"\n\n    @property\n    def is_global(self):\n        return self.model is None\n\n    @property\n    def is_model(self):\n        if self.model is not None and self.model.python_model is not None:\n            return self.model.python_model == self.path\n\n    @property\n    def model_name(self):\n        return \"<GLOBAL>\" if self.is_global else self.model.name  # type: ignore\n\n    def _build_script_context(self) -> Context:\n        context_config = ContextConfig(self.faldbt._config)\n        if self.is_global:\n            return Context(current_model=None, config=context_config)\n\n        model: DbtModel = self.model  # type: ignore\n\n        meta = model.meta or {}\n        _del_key(meta, self.faldbt.keyword)\n\n        tests = _process_tests(model.tests)\n\n        current_adapter_response = None\n        if model.adapter_response:\n            current_adapter_response = CurrentAdapterResponse(\n                message=str(model.adapter_response),\n                code=model.adapter_response.code,\n                rows_affected=model.adapter_response.rows_affected,\n            )\n\n        current_model = CurrentModel(\n            name=model.name,\n            alias=model.alias,\n            status=model.status,\n            columns=model.columns,\n            tests=tests,\n            meta=meta,\n            adapter_response=current_adapter_response,\n        )\n\n        return Context(\n            current_model=current_model,\n            config=context_config,\n            _arguments=self.hook_arguments,\n        )\n\n\ndef _del_key(dict: Dict[str, Any], key: str):\n    try:\n        del dict[key]\n    except KeyError:\n        pass\n\n\ndef _process_tests(tests: List[Any]):\n    return list(\n        map(\n            lambda test: CurrentTest(\n                name=test.name,\n                column=test.column,\n                status=test.status,\n                model_name=test.model,\n            ),\n            tests,\n        )\n    )\n\n\ndef python_from_file(path: Path) -> str:\n    with open(path) as file:\n        raw_source_code = file.read()\n        if path.suffix == \".ipynb\":\n            raw_source_code = _process_ipynb(raw_source_code)\n    return raw_source_code\n\n\ndef _process_ipynb(raw_source_code: str) -> str:\n    def strip_magic(source: List[str]) -> List[str]:\n        NOTEBOOK_LIB = \"faldbt.magics\"\n        return [item for item in source if item[0] != \"%\" and NOTEBOOK_LIB not in item]\n\n    ipynb_struct = json.loads(raw_source_code)\n\n    script_list = []\n    for cell in ipynb_struct[\"cells\"]:\n        if cell[\"cell_type\"] == \"code\":\n            source = strip_magic(cell[\"source\"])\n            script_list.append(\"\".join(source))\n\n    joined_script = \"\\n #cell \\n\".join(script_list)\n\n    LOGGER.debug(f\"Joined .ipynb cells to:\\n{joined_script}\")\n\n    return joined_script\n\n\ndef _not_allowed_function_maker(function_name: str) -> Callable[[Any], None]:\n    def not_allowed_function(*args, **kwargs):\n        raise Exception(\n            (\n                f\"{function_name} is not allowed in hooks.\"\n                \" Consider using a Python model.\"\n            )\n        )\n\n    return not_allowed_function\n", "path": "src/fal/fal_script.py"}], "after_files": [{"content": "import os\nimport json\nfrom typing import Dict, Any, List, Optional, Union, Callable\nfrom pathlib import Path\nfrom functools import partial\nfrom dataclasses import dataclass, field\nfrom deprecation import deprecated\n\nfrom faldbt.parse import normalize_path\nfrom faldbt.project import DbtModel, FalDbt\n\nfrom dbt.contracts.results import RunStatus\nfrom dbt.config.runtime import RuntimeConfig\nfrom fal.logger import LOGGER\n\nfrom dbt.contracts.graph.parsed import ColumnInfo\n\n\nclass Hook:\n    path: str\n    arguments: Dict[str, Any]\n\n\n@dataclass\nclass LocalHook(Hook):\n    path: str\n    arguments: Dict[str, Any] = field(default_factory=dict)\n\n\n@dataclass\nclass IsolatedHook(Hook):\n    path: str\n    environment_name: str\n    arguments: Dict[str, Any] = field(default_factory=dict)\n\n\ndef _is_local_environment(environment_name: str) -> None:\n    return environment_name == \"local\"\n\n\ndef create_hook(raw_hook: Any, default_environment_name: Optional[str] = None) -> Hook:\n    if isinstance(raw_hook, str):\n        raw_hook = {\"path\": raw_hook}\n\n    if not isinstance(raw_hook, dict):\n        raise ValueError(f\"Unrecognized hook value: {raw_hook}\")\n\n    if \"path\" not in raw_hook:\n        raise ValueError(f\"A hook must specify path.\")\n\n    environment_name = raw_hook.get(\"environment\", default_environment_name)\n    if environment_name and not _is_local_environment(environment_name):\n        return IsolatedHook(\n            raw_hook[\"path\"],\n            environment_name,\n            raw_hook.get(\"with\", {}),\n        )\n    else:\n        return LocalHook(raw_hook[\"path\"], raw_hook.get(\"with\", {}))\n\n\n@dataclass\nclass CurrentAdapterResponse:\n    message: str\n    code: Optional[str]\n    rows_affected: Optional[int]\n\n\n@dataclass\nclass CurrentModel:\n    name: str\n    alias: str\n    status: RunStatus\n    columns: Dict[str, ColumnInfo]\n    tests: List[Any]\n    meta: Dict[Any, Any]\n    adapter_response: Optional[CurrentAdapterResponse]\n\n\n@dataclass\nclass CurrentTest:\n    name: str\n    model_name: str\n    column: str\n    status: str\n\n    @property\n    @deprecated(details=\"Use 'model_name' instead\")\n    def modelname(self):\n        return self.model_name\n\n\n@dataclass\nclass ContextConfig:\n    target_path: Path\n\n    def __init__(self, config: RuntimeConfig):\n        self.target_path = Path(\n            os.path.realpath(os.path.join(config.project_root, config.target_path))\n        )\n\n@dataclass\nclass ContextTarget:\n    def __init__(self, config: RuntimeConfig):\n        self.profile_name = config.profile_name\n        self.name = config.target_name\n        self.threads = config.threads\n        self.type = config.credentials.type\n        self.database = config.credentials.database\n        self.schema = config.credentials.schema\n\n@dataclass\nclass Context:\n    current_model: Union[CurrentModel, None]\n    config: ContextConfig\n    target: ContextTarget\n    _arguments: Optional[Dict[str, Any]] = field(repr=False, default=None)\n\n    @property\n    def arguments(self) -> Dict[str, Any]:\n        if self._arguments is None:\n            raise ValueError(\n                \"'context.arguments' is only accessible from hooks, \"\n                \"not from scripts/models\"\n            )\n        return self._arguments\n\n\n@dataclass(frozen=True, init=False)\nclass FalScript:\n    model: Optional[DbtModel]\n    path: Path\n    faldbt: FalDbt\n    hook_arguments: Optional[Dict[str, Any]]\n    is_hook: bool\n\n    def __init__(\n        self,\n        faldbt: FalDbt,\n        model: Optional[DbtModel],\n        path: str,\n        hook_arguments: Optional[Dict[str, Any]] = None,\n        is_hook: bool = False,\n    ):\n        # Necessary because of frozen=True\n        object.__setattr__(self, \"model\", model)\n        object.__setattr__(self, \"path\", normalize_path(faldbt.scripts_dir, path))\n        object.__setattr__(self, \"faldbt\", faldbt)\n        object.__setattr__(self, \"hook_arguments\", hook_arguments)\n        object.__setattr__(self, \"is_hook\", is_hook)\n\n    @classmethod\n    def from_hook(cls, faldbt: FalDbt, model: DbtModel, hook: Hook):\n        \"\"\"\n        Creates a FalScript from a hook\n        \"\"\"\n        assert isinstance(hook, LocalHook)\n        return cls(\n            faldbt=faldbt,\n            model=model,\n            path=hook.path,\n            hook_arguments=hook.arguments,\n            is_hook=True,\n        )\n\n    @classmethod\n    def model_script(cls, faldbt: FalDbt, model: DbtModel):\n        script = FalScript(faldbt, model, \"\")\n        # HACK: Set the script path specially for this case\n        object.__setattr__(script, \"path\", model.python_model)\n        return script\n\n    def exec(self):\n        \"\"\"\n        Executes the script\n        \"\"\"\n        # Enable local imports\n        try:\n            source_code = python_from_file(self.path)\n            program = compile(source_code, self.path, \"exec\")\n\n            exec_globals = {\n                \"__name__\": \"__main__\",\n                \"context\": self._build_script_context(),\n                \"ref\": self.faldbt.ref,\n                \"source\": self.faldbt.source,\n                \"write_to_firestore\": self.faldbt.write_to_firestore,\n                \"list_models\": self.faldbt.list_models,\n                \"list_models_ids\": self.faldbt.list_models_ids,\n                \"list_sources\": self.faldbt.list_sources,\n                \"list_features\": self.faldbt.list_features,\n                \"el\": self.faldbt.el,\n                \"execute_sql\": self.faldbt.execute_sql,\n            }\n\n            if not self.is_hook:\n                exec_globals[\"write_to_source\"] = self.faldbt.write_to_source\n\n                if self.model is not None:\n                    # Hard-wire the model\n                    exec_globals[\"write_to_model\"] = partial(\n                        self.faldbt.write_to_model,\n                        target_1=self.model.name,\n                        target_2=None,\n                    )\n\n            else:\n                exec_globals[\"write_to_source\"] = _not_allowed_function_maker(\n                    \"write_to_source\"\n                )\n                exec_globals[\"write_to_model\"] = _not_allowed_function_maker(\n                    \"write_to_model\"\n                )\n            exec(program, exec_globals)\n        finally:\n            pass\n\n    @property\n    def relative_path(self):\n        if self.is_model:\n            return self.path.relative_to(self.faldbt.project_dir)\n        else:\n            return self.path.relative_to(self.faldbt.scripts_dir)\n\n    @property\n    def id(self):\n        if self.is_model:\n            return f\"(model: {self.relative_path})\"\n        else:\n            return f\"({self.model_name}, {self.relative_path})\"\n\n    @property\n    def is_global(self):\n        return self.model is None\n\n    @property\n    def is_model(self):\n        if self.model is not None and self.model.python_model is not None:\n            return self.model.python_model == self.path\n\n    @property\n    def model_name(self):\n        return \"<GLOBAL>\" if self.is_global else self.model.name  # type: ignore\n\n    def _build_script_context(self) -> Context:\n        config: RuntimeConfig = self.faldbt._config\n        context_config = ContextConfig(config)\n        target = ContextTarget(config)\n\n        if self.is_global:\n            return Context(current_model=None, target=target, config=context_config)\n\n        model: DbtModel = self.model  # type: ignore\n\n        meta = model.meta or {}\n        _del_key(meta, self.faldbt.keyword)\n\n        tests = _process_tests(model.tests)\n\n        current_adapter_response = None\n        if model.adapter_response:\n            current_adapter_response = CurrentAdapterResponse(\n                message=str(model.adapter_response),\n                code=model.adapter_response.code,\n                rows_affected=model.adapter_response.rows_affected,\n            )\n\n        current_model = CurrentModel(\n            name=model.name,\n            alias=model.alias,\n            status=model.status,\n            columns=model.columns,\n            tests=tests,\n            meta=meta,\n            adapter_response=current_adapter_response,\n        )\n\n        return Context(\n            current_model=current_model,\n            target=target,\n            config=context_config,\n            _arguments=self.hook_arguments,\n        )\n\n\ndef _del_key(dict: Dict[str, Any], key: str):\n    try:\n        del dict[key]\n    except KeyError:\n        pass\n\n\ndef _process_tests(tests: List[Any]):\n    return list(\n        map(\n            lambda test: CurrentTest(\n                name=test.name,\n                column=test.column,\n                status=test.status,\n                model_name=test.model,\n            ),\n            tests,\n        )\n    )\n\n\ndef python_from_file(path: Path) -> str:\n    with open(path) as file:\n        raw_source_code = file.read()\n        if path.suffix == \".ipynb\":\n            raw_source_code = _process_ipynb(raw_source_code)\n    return raw_source_code\n\n\ndef _process_ipynb(raw_source_code: str) -> str:\n    def strip_magic(source: List[str]) -> List[str]:\n        NOTEBOOK_LIB = \"faldbt.magics\"\n        return [item for item in source if item[0] != \"%\" and NOTEBOOK_LIB not in item]\n\n    ipynb_struct = json.loads(raw_source_code)\n\n    script_list = []\n    for cell in ipynb_struct[\"cells\"]:\n        if cell[\"cell_type\"] == \"code\":\n            source = strip_magic(cell[\"source\"])\n            script_list.append(\"\".join(source))\n\n    joined_script = \"\\n #cell \\n\".join(script_list)\n\n    LOGGER.debug(f\"Joined .ipynb cells to:\\n{joined_script}\")\n\n    return joined_script\n\n\ndef _not_allowed_function_maker(function_name: str) -> Callable[[Any], None]:\n    def not_allowed_function(*args, **kwargs):\n        raise Exception(\n            (\n                f\"{function_name} is not allowed in hooks.\"\n                \" Consider using a Python model.\"\n            )\n        )\n\n    return not_allowed_function\n", "path": "src/fal/fal_script.py"}]}