We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Datasets without `variables.json` should still deploy properly
## Description

Some datasets (and their pipelines) may not need Airflow variables at all. 

During the deploy step, when a dataset's `variables.json` file is missing, the deploy script errors out with a `FileNotFound` error. We should let such datasets skip importing Airflow variables and proceed with deploying the pipeline DAGs.

## Checklist
- [x] I created this issue in accordance with the [Code of Conduct](https://github.com/GoogleCloudPlatform/public-datasets-pipelines/blob/main/CODE_OF_CONDUCT.md).
- [x] This issue is appropriately labeled.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `scripts/deploy_dag.py`
Content:
```
1 # Copyright 2021 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 
16 import argparse
17 import pathlib
18 import subprocess
19 import typing
20 
21 CURRENT_PATH = pathlib.Path(__file__).resolve().parent
22 PROJECT_ROOT = CURRENT_PATH.parent
23 DATASETS_PATH = PROJECT_ROOT / "datasets"
24 
25 
26 def main(
27     local: bool,
28     env_path: pathlib.Path,
29     dataset_id: str,
30     pipeline: str = None,
31     airflow_home: pathlib.Path = None,
32     composer_env: str = None,
33     composer_bucket: str = None,
34     composer_region: str = None,
35 ):
36     print("\n========== AIRFLOW VARIABLES ==========")
37     copy_variables_to_airflow_data_folder(
38         local, env_path, dataset_id, airflow_home, composer_bucket
39     )
40     import_variables_to_airflow_env(
41         local, env_path, dataset_id, composer_env, composer_bucket, composer_region
42     )
43 
44     print("========== AIRFLOW DAGS ==========")
45     if pipeline:
46         pipelines = [env_path / "datasets" / pipeline]
47     else:
48         pipelines = list_subdirs(env_path / "datasets" / dataset_id)
49 
50     for pipeline_path in pipelines:
51         copy_custom_callables_to_airflow_dags_folder(
52             local,
53             env_path,
54             dataset_id,
55             pipeline_path.name,
56             composer_bucket,
57             airflow_home,
58         )
59 
60         copy_generated_dag_to_airflow_dags_folder(
61             local,
62             env_path,
63             dataset_id,
64             pipeline_path.name,
65             composer_bucket,
66             airflow_home,
67         )
68 
69 
70 def copy_variables_to_airflow_data_folder(
71     local: bool,
72     env_path: pathlib.Path,
73     dataset_id: str,
74     airflow_home: pathlib.Path = None,
75     composer_bucket: str = None,
76 ):
77     """
78     cd .{ENV}/{DATASET_ID}
79     """
80     cwd = env_path / "datasets" / dataset_id
81     filename = f"{dataset_id}_variables.json"
82 
83     check_existence_of_variables_file(cwd / filename)
84 
85     if local:
86         """
87         cp {DATASET_ID}_variables.json {AIRFLOW_HOME}/data/variables/{DATASET_ID}_variables.json
88         """
89         target_path = airflow_home / "data" / "variables" / filename
90         target_path.mkdir(parents=True, exist_ok=True)
91         print(
92             "\nCopying variables JSON file into Airflow data folder\n\n"
93             f"  Source:\n  {cwd / filename}\n\n"
94             f"  Destination:\n  {target_path}\n"
95         )
96 
97         subprocess.check_call(["cp", "-rf", filename, str(target_path)], cwd=cwd)
98     else:
99         """
100         [remote]
101         gsutil cp {DATASET_ID}_variables.json gs://{COMPOSER_BUCKET}/data/variables/{DATASET_ID}_variables.json...
102         """
103         gcs_uri = f"gs://{composer_bucket}/data/variables/{filename}"
104         print(
105             "\nCopying variables JSON file into Cloud Composer data folder\n\n"
106             f"  Source:\n  {cwd / filename}\n\n"
107             f"  Destination:\n  {gcs_uri}\n"
108         )
109         subprocess.check_call(["gsutil", "cp", filename, gcs_uri], cwd=cwd)
110 
111 
112 def import_variables_to_airflow_env(
113     local: bool,
114     env_path: pathlib.Path,
115     dataset_id: str,
116     composer_env: str = None,
117     composer_bucket: str = None,
118     composer_region: str = None,
119 ):
120     """
121     [local]
122     airflow variables import .{ENV}/datasets/{DATASET_ID}/variables.json
123 
124     [remote]
125     gcloud composer environments run COMPOSER_ENV --location COMPOSER_REGION variables -- --import /home/airflow/gcs/data/variables/{DATASET_ID}_variables.json
126     """
127     cwd = env_path / "datasets" / dataset_id
128     filename = f"{dataset_id}_variables.json"
129 
130     if local:
131         print(f"\nImporting Airflow variables from {cwd / filename}...\n")
132         subprocess.check_call(
133             ["airflow", "variables", "import", str(cwd / filename)], cwd=cwd
134         )
135     else:
136         gcs_uri = f"gs://{composer_bucket}/data/variables/{filename}"
137         airflow_path = f"/home/airflow/gcs/data/variables/{filename}"
138         print(f"\nImporting Airflow variables from {gcs_uri} ({airflow_path})...\n")
139         subprocess.check_call(
140             [
141                 "gcloud",
142                 "composer",
143                 "environments",
144                 "run",
145                 str(composer_env),
146                 "--location",
147                 str(composer_region),
148                 "variables",
149                 "--",
150                 "--import",
151                 str(airflow_path),
152             ],
153             cwd=cwd,
154         )
155 
156 
157 def copy_generated_dag_to_airflow_dags_folder(
158     local: bool,
159     env_path: pathlib.Path,
160     dataset_id: str,
161     pipeline_id: str,
162     composer_bucket: str = None,
163     airflow_home: pathlib.Path = None,
164 ):
165     """
166     cd {DATASET_ID}/{PIPELINE_ID}
167 
168     [local]
169     cp {PIPELINE_ID}_dag.py {AIRFLOW_HOME}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py
170 
171     [remote]
172     gsutil cp {PIPELINE_ID}_dag.py gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py
173     """
174     cwd = env_path / "datasets" / dataset_id / pipeline_id
175     filename = f"{pipeline_id}_dag.py"
176 
177     if local:
178         target = airflow_home / "dags" / f"{dataset_id}__{pipeline_id}_dag.py"
179         print(
180             f"\nCopying DAG file for pipeline `{pipeline_id}` into Airflow DAGs folder\n\n"
181             f"  Source:\n  {cwd / filename}\n\n"
182             f"  Destination:\n  {target}\n"
183         )
184         subprocess.check_call(["cp", "-rf", filename, str(target)], cwd=cwd)
185     else:
186         target = f"gs://{composer_bucket}/dags/{dataset_id}__{pipeline_id}_dag.py"
187         print(
188             f"\nCopying DAG file for pipeline `{pipeline_id}` into Cloud Composer DAG folder\n\n"
189             f"  Source:\n  {cwd / filename}\n\n"
190             f"  Destination:\n  {target}\n"
191         )
192         subprocess.check_call(["gsutil", "cp", filename, target], cwd=cwd)
193 
194 
195 def copy_custom_callables_to_airflow_dags_folder(
196     local: bool,
197     env_path: pathlib.Path,
198     dataset_id: str,
199     pipeline_id: str,
200     composer_bucket: str = None,
201     airflow_home: pathlib.Path = None,
202 ):
203     """
204     cd {DATASET_ID}/{PIPELINE_ID}
205 
206     [local]
207     mkdir -p {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom
208     cp -rf custom {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom
209 
210     [remote]
211     gsutil cp -r custom gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}/{PIPELINE_ID}/
212     """
213     cwd = env_path / "datasets" / dataset_id / pipeline_id
214 
215     if not (cwd / "custom").exists():
216         return
217 
218     if local:
219         target_parent = airflow_home / "dags" / dataset_id / pipeline_id
220         target_parent.mkdir(parents=True, exist_ok=True)
221         print(
222             f"\nCopying custom callables folder for pipeline `{pipeline_id}` into Airflow DAGs folder\n\n"
223             f"  Source:\n  {cwd / 'custom'}\n\n"
224             f"  Destination:\n  {target_parent / 'custom'}\n"
225         )
226         subprocess.check_call(["cp", "-rf", "custom", str(target_parent)], cwd=cwd)
227     else:
228         target = f"gs://{composer_bucket}/dags/{dataset_id}/{pipeline_id}/"
229         print(
230             f"\nCopying custom callables folder for pipeline `{pipeline_id}` into Cloud Composer DAG folder\n\n"
231             f"  Source:\n  {cwd / 'custom'}\n\n"
232             f"  Destination:\n  {target}\n"
233         )
234         subprocess.check_call(["gsutil", "cp", "-r", "custom", target], cwd=cwd)
235 
236 
237 def check_existence_of_variables_file(file_path: pathlib.Path):
238     if not file_path:
239         raise FileNotFoundError(f"Airflow variables file {file_path} does not exist.")
240 
241 
242 def list_subdirs(path: pathlib.Path) -> typing.List[pathlib.Path]:
243     """Returns a list of subdirectories"""
244     subdirs = [f for f in path.iterdir() if f.is_dir() and not f.name[0] in (".", "_")]
245     return subdirs
246 
247 
248 if __name__ == "__main__":
249     parser = argparse.ArgumentParser(
250         description="Deploy DAGs and variables to an Airflow environment"
251     )
252     parser.add_argument(
253         "-d",
254         "--dataset",
255         required=True,
256         type=str,
257         dest="dataset",
258         help="The directory name of the dataset.",
259     )
260     parser.add_argument(
261         "-e",
262         "--env",
263         type=str,
264         default="dev",
265         dest="env",
266         help="The stage used for the resources: dev|staging|prod",
267     )
268     parser.add_argument(
269         "-n",
270         "--composer-env",
271         type=str,
272         dest="composer_env",
273         help="The Google Cloud Composer environment name",
274     )
275     parser.add_argument(
276         "-b",
277         "--composer-bucket",
278         type=str,
279         dest="composer_bucket",
280         help="The Google Cloud Composer bucket name",
281     )
282     parser.add_argument(
283         "-r",
284         "--composer-region",
285         type=str,
286         dest="composer_region",
287         help="The region of the Google Cloud Composer environment",
288     )
289     parser.add_argument(
290         "-p",
291         "--pipeline",
292         required=False,
293         type=str,
294         dest="pipeline",
295         help="The directory name of the pipeline",
296     )
297     parser.add_argument(
298         "-a",
299         "--airflow-home",
300         type=str,
301         default="~/airflow",
302         dest="airflow_home",
303         help="pathlib.Path to the Airflow home directory (defaults to `~/airflow`)",
304     )
305     parser.add_argument("--local", required=False, dest="local", action="store_true")
306 
307     args = parser.parse_args()
308     airflow_path = pathlib.Path(args.airflow_home).expanduser()
309 
310     if args.local:
311         if not airflow_path.exists() and airflow_path.is_dir():
312             raise ValueError(
313                 "Argument `-a|--airflow-home` must exist and be a directory"
314             )
315     else:
316         if not args.composer_env:
317             raise ValueError(
318                 "Argument `-n|--composer-env` (Composer environment name) not specified"
319             )
320 
321         if not args.composer_bucket:
322             raise ValueError(
323                 "Argument `-b|--composer-bucket` (Composer bucket name) not specified"
324             )
325 
326         if not args.composer_region:
327             raise ValueError(
328                 "Argument `-r|--composer-region` (Composer environment region) not specified"
329             )
330 
331     main(
332         local=args.local,
333         env_path=PROJECT_ROOT / f".{args.env}",
334         dataset_id=args.dataset,
335         pipeline=args.pipeline,
336         airflow_home=airflow_path,
337         composer_env=args.composer_env,
338         composer_bucket=args.composer_bucket,
339         composer_region=args.composer_region,
340     )
341
```
--- 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/scripts/deploy_dag.py b/scripts/deploy_dag.py
--- a/scripts/deploy_dag.py
+++ b/scripts/deploy_dag.py
@@ -17,6 +17,7 @@
 import pathlib
 import subprocess
 import typing
+import warnings
 
 CURRENT_PATH = pathlib.Path(__file__).resolve().parent
 PROJECT_ROOT = CURRENT_PATH.parent
@@ -67,6 +68,10 @@
         )
 
 
+def run_gsutil_cmd(args: typing.List[str], cwd: pathlib.Path):
+    subprocess.check_call(["gsutil"] + args, cwd=cwd)
+
+
 def copy_variables_to_airflow_data_folder(
     local: bool,
     env_path: pathlib.Path,
@@ -80,7 +85,9 @@
     cwd = env_path / "datasets" / dataset_id
     filename = f"{dataset_id}_variables.json"
 
-    check_existence_of_variables_file(cwd / filename)
+    if not (cwd / filename).exists():
+        warnings.warn(f"Airflow variables file {filename} does not exist.")
+        return
 
     if local:
         """
@@ -106,7 +113,31 @@
             f"  Source:\n  {cwd / filename}\n\n"
             f"  Destination:\n  {gcs_uri}\n"
         )
-        subprocess.check_call(["gsutil", "cp", filename, gcs_uri], cwd=cwd)
+        run_gsutil_cmd(["cp", filename, gcs_uri], cwd=cwd)
+
+
+def run_cloud_composer_vars_import(
+    composer_env: str,
+    composer_region: str,
+    airflow_path: pathlib.Path,
+    cwd: pathlib.Path,
+):
+    subprocess.check_call(
+        [
+            "gcloud",
+            "composer",
+            "environments",
+            "run",
+            str(composer_env),
+            "--location",
+            str(composer_region),
+            "variables",
+            "--",
+            "--import",
+            str(airflow_path),
+        ],
+        cwd=cwd,
+    )
 
 
 def import_variables_to_airflow_env(
@@ -136,21 +167,8 @@
         gcs_uri = f"gs://{composer_bucket}/data/variables/{filename}"
         airflow_path = f"/home/airflow/gcs/data/variables/{filename}"
         print(f"\nImporting Airflow variables from {gcs_uri} ({airflow_path})...\n")
-        subprocess.check_call(
-            [
-                "gcloud",
-                "composer",
-                "environments",
-                "run",
-                str(composer_env),
-                "--location",
-                str(composer_region),
-                "variables",
-                "--",
-                "--import",
-                str(airflow_path),
-            ],
-            cwd=cwd,
+        run_cloud_composer_vars_import(
+            composer_env, composer_region, airflow_path, cwd=cwd
         )
 
 
@@ -189,7 +207,7 @@
             f"  Source:\n  {cwd / filename}\n\n"
             f"  Destination:\n  {target}\n"
         )
-        subprocess.check_call(["gsutil", "cp", filename, target], cwd=cwd)
+        run_gsutil_cmd(["cp", filename, target], cwd=cwd)
 
 
 def copy_custom_callables_to_airflow_dags_folder(
@@ -231,7 +249,7 @@
             f"  Source:\n  {cwd / 'custom'}\n\n"
             f"  Destination:\n  {target}\n"
         )
-        subprocess.check_call(["gsutil", "cp", "-r", "custom", target], cwd=cwd)
+        run_gsutil_cmd(["cp", "-r", "custom", target], cwd=cwd)
 
 
 def check_existence_of_variables_file(file_path: pathlib.Path):

{"golden_diff": "diff --git a/scripts/deploy_dag.py b/scripts/deploy_dag.py\n--- a/scripts/deploy_dag.py\n+++ b/scripts/deploy_dag.py\n@@ -17,6 +17,7 @@\n import pathlib\n import subprocess\n import typing\n+import warnings\n \n CURRENT_PATH = pathlib.Path(__file__).resolve().parent\n PROJECT_ROOT = CURRENT_PATH.parent\n@@ -67,6 +68,10 @@\n         )\n \n \n+def run_gsutil_cmd(args: typing.List[str], cwd: pathlib.Path):\n+    subprocess.check_call([\"gsutil\"] + args, cwd=cwd)\n+\n+\n def copy_variables_to_airflow_data_folder(\n     local: bool,\n     env_path: pathlib.Path,\n@@ -80,7 +85,9 @@\n     cwd = env_path / \"datasets\" / dataset_id\n     filename = f\"{dataset_id}_variables.json\"\n \n-    check_existence_of_variables_file(cwd / filename)\n+    if not (cwd / filename).exists():\n+        warnings.warn(f\"Airflow variables file {filename} does not exist.\")\n+        return\n \n     if local:\n         \"\"\"\n@@ -106,7 +113,31 @@\n             f\"  Source:\\n  {cwd / filename}\\n\\n\"\n             f\"  Destination:\\n  {gcs_uri}\\n\"\n         )\n-        subprocess.check_call([\"gsutil\", \"cp\", filename, gcs_uri], cwd=cwd)\n+        run_gsutil_cmd([\"cp\", filename, gcs_uri], cwd=cwd)\n+\n+\n+def run_cloud_composer_vars_import(\n+    composer_env: str,\n+    composer_region: str,\n+    airflow_path: pathlib.Path,\n+    cwd: pathlib.Path,\n+):\n+    subprocess.check_call(\n+        [\n+            \"gcloud\",\n+            \"composer\",\n+            \"environments\",\n+            \"run\",\n+            str(composer_env),\n+            \"--location\",\n+            str(composer_region),\n+            \"variables\",\n+            \"--\",\n+            \"--import\",\n+            str(airflow_path),\n+        ],\n+        cwd=cwd,\n+    )\n \n \n def import_variables_to_airflow_env(\n@@ -136,21 +167,8 @@\n         gcs_uri = f\"gs://{composer_bucket}/data/variables/{filename}\"\n         airflow_path = f\"/home/airflow/gcs/data/variables/{filename}\"\n         print(f\"\\nImporting Airflow variables from {gcs_uri} ({airflow_path})...\\n\")\n-        subprocess.check_call(\n-            [\n-                \"gcloud\",\n-                \"composer\",\n-                \"environments\",\n-                \"run\",\n-                str(composer_env),\n-                \"--location\",\n-                str(composer_region),\n-                \"variables\",\n-                \"--\",\n-                \"--import\",\n-                str(airflow_path),\n-            ],\n-            cwd=cwd,\n+        run_cloud_composer_vars_import(\n+            composer_env, composer_region, airflow_path, cwd=cwd\n         )\n \n \n@@ -189,7 +207,7 @@\n             f\"  Source:\\n  {cwd / filename}\\n\\n\"\n             f\"  Destination:\\n  {target}\\n\"\n         )\n-        subprocess.check_call([\"gsutil\", \"cp\", filename, target], cwd=cwd)\n+        run_gsutil_cmd([\"cp\", filename, target], cwd=cwd)\n \n \n def copy_custom_callables_to_airflow_dags_folder(\n@@ -231,7 +249,7 @@\n             f\"  Source:\\n  {cwd / 'custom'}\\n\\n\"\n             f\"  Destination:\\n  {target}\\n\"\n         )\n-        subprocess.check_call([\"gsutil\", \"cp\", \"-r\", \"custom\", target], cwd=cwd)\n+        run_gsutil_cmd([\"cp\", \"-r\", \"custom\", target], cwd=cwd)\n \n \n def check_existence_of_variables_file(file_path: pathlib.Path):\n", "issue": "Datasets without `variables.json` should still deploy properly\n## Description\r\n\r\nSome datasets (and their pipelines) may not need Airflow variables at all. \r\n\r\nDuring the deploy step, when a dataset's `variables.json` file is missing, the deploy script errors out with a `FileNotFound` error. We should let such datasets skip importing Airflow variables and proceed with deploying the pipeline DAGs.\r\n\r\n## Checklist\r\n- [x] I created this issue in accordance with the [Code of Conduct](https://github.com/GoogleCloudPlatform/public-datasets-pipelines/blob/main/CODE_OF_CONDUCT.md).\r\n- [x] This issue is appropriately labeled.\r\n\n", "before_files": [{"content": "# Copyright 2021 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport argparse\nimport pathlib\nimport subprocess\nimport typing\n\nCURRENT_PATH = pathlib.Path(__file__).resolve().parent\nPROJECT_ROOT = CURRENT_PATH.parent\nDATASETS_PATH = PROJECT_ROOT / \"datasets\"\n\n\ndef main(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline: str = None,\n    airflow_home: pathlib.Path = None,\n    composer_env: str = None,\n    composer_bucket: str = None,\n    composer_region: str = None,\n):\n    print(\"\\n========== AIRFLOW VARIABLES ==========\")\n    copy_variables_to_airflow_data_folder(\n        local, env_path, dataset_id, airflow_home, composer_bucket\n    )\n    import_variables_to_airflow_env(\n        local, env_path, dataset_id, composer_env, composer_bucket, composer_region\n    )\n\n    print(\"========== AIRFLOW DAGS ==========\")\n    if pipeline:\n        pipelines = [env_path / \"datasets\" / pipeline]\n    else:\n        pipelines = list_subdirs(env_path / \"datasets\" / dataset_id)\n\n    for pipeline_path in pipelines:\n        copy_custom_callables_to_airflow_dags_folder(\n            local,\n            env_path,\n            dataset_id,\n            pipeline_path.name,\n            composer_bucket,\n            airflow_home,\n        )\n\n        copy_generated_dag_to_airflow_dags_folder(\n            local,\n            env_path,\n            dataset_id,\n            pipeline_path.name,\n            composer_bucket,\n            airflow_home,\n        )\n\n\ndef copy_variables_to_airflow_data_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    airflow_home: pathlib.Path = None,\n    composer_bucket: str = None,\n):\n    \"\"\"\n    cd .{ENV}/{DATASET_ID}\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id\n    filename = f\"{dataset_id}_variables.json\"\n\n    check_existence_of_variables_file(cwd / filename)\n\n    if local:\n        \"\"\"\n        cp {DATASET_ID}_variables.json {AIRFLOW_HOME}/data/variables/{DATASET_ID}_variables.json\n        \"\"\"\n        target_path = airflow_home / \"data\" / \"variables\" / filename\n        target_path.mkdir(parents=True, exist_ok=True)\n        print(\n            \"\\nCopying variables JSON file into Airflow data folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target_path}\\n\"\n        )\n\n        subprocess.check_call([\"cp\", \"-rf\", filename, str(target_path)], cwd=cwd)\n    else:\n        \"\"\"\n        [remote]\n        gsutil cp {DATASET_ID}_variables.json gs://{COMPOSER_BUCKET}/data/variables/{DATASET_ID}_variables.json...\n        \"\"\"\n        gcs_uri = f\"gs://{composer_bucket}/data/variables/{filename}\"\n        print(\n            \"\\nCopying variables JSON file into Cloud Composer data folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {gcs_uri}\\n\"\n        )\n        subprocess.check_call([\"gsutil\", \"cp\", filename, gcs_uri], cwd=cwd)\n\n\ndef import_variables_to_airflow_env(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    composer_env: str = None,\n    composer_bucket: str = None,\n    composer_region: str = None,\n):\n    \"\"\"\n    [local]\n    airflow variables import .{ENV}/datasets/{DATASET_ID}/variables.json\n\n    [remote]\n    gcloud composer environments run COMPOSER_ENV --location COMPOSER_REGION variables -- --import /home/airflow/gcs/data/variables/{DATASET_ID}_variables.json\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id\n    filename = f\"{dataset_id}_variables.json\"\n\n    if local:\n        print(f\"\\nImporting Airflow variables from {cwd / filename}...\\n\")\n        subprocess.check_call(\n            [\"airflow\", \"variables\", \"import\", str(cwd / filename)], cwd=cwd\n        )\n    else:\n        gcs_uri = f\"gs://{composer_bucket}/data/variables/{filename}\"\n        airflow_path = f\"/home/airflow/gcs/data/variables/{filename}\"\n        print(f\"\\nImporting Airflow variables from {gcs_uri} ({airflow_path})...\\n\")\n        subprocess.check_call(\n            [\n                \"gcloud\",\n                \"composer\",\n                \"environments\",\n                \"run\",\n                str(composer_env),\n                \"--location\",\n                str(composer_region),\n                \"variables\",\n                \"--\",\n                \"--import\",\n                str(airflow_path),\n            ],\n            cwd=cwd,\n        )\n\n\ndef copy_generated_dag_to_airflow_dags_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline_id: str,\n    composer_bucket: str = None,\n    airflow_home: pathlib.Path = None,\n):\n    \"\"\"\n    cd {DATASET_ID}/{PIPELINE_ID}\n\n    [local]\n    cp {PIPELINE_ID}_dag.py {AIRFLOW_HOME}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py\n\n    [remote]\n    gsutil cp {PIPELINE_ID}_dag.py gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id / pipeline_id\n    filename = f\"{pipeline_id}_dag.py\"\n\n    if local:\n        target = airflow_home / \"dags\" / f\"{dataset_id}__{pipeline_id}_dag.py\"\n        print(\n            f\"\\nCopying DAG file for pipeline `{pipeline_id}` into Airflow DAGs folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        subprocess.check_call([\"cp\", \"-rf\", filename, str(target)], cwd=cwd)\n    else:\n        target = f\"gs://{composer_bucket}/dags/{dataset_id}__{pipeline_id}_dag.py\"\n        print(\n            f\"\\nCopying DAG file for pipeline `{pipeline_id}` into Cloud Composer DAG folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        subprocess.check_call([\"gsutil\", \"cp\", filename, target], cwd=cwd)\n\n\ndef copy_custom_callables_to_airflow_dags_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline_id: str,\n    composer_bucket: str = None,\n    airflow_home: pathlib.Path = None,\n):\n    \"\"\"\n    cd {DATASET_ID}/{PIPELINE_ID}\n\n    [local]\n    mkdir -p {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom\n    cp -rf custom {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom\n\n    [remote]\n    gsutil cp -r custom gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}/{PIPELINE_ID}/\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id / pipeline_id\n\n    if not (cwd / \"custom\").exists():\n        return\n\n    if local:\n        target_parent = airflow_home / \"dags\" / dataset_id / pipeline_id\n        target_parent.mkdir(parents=True, exist_ok=True)\n        print(\n            f\"\\nCopying custom callables folder for pipeline `{pipeline_id}` into Airflow DAGs folder\\n\\n\"\n            f\"  Source:\\n  {cwd / 'custom'}\\n\\n\"\n            f\"  Destination:\\n  {target_parent / 'custom'}\\n\"\n        )\n        subprocess.check_call([\"cp\", \"-rf\", \"custom\", str(target_parent)], cwd=cwd)\n    else:\n        target = f\"gs://{composer_bucket}/dags/{dataset_id}/{pipeline_id}/\"\n        print(\n            f\"\\nCopying custom callables folder for pipeline `{pipeline_id}` into Cloud Composer DAG folder\\n\\n\"\n            f\"  Source:\\n  {cwd / 'custom'}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        subprocess.check_call([\"gsutil\", \"cp\", \"-r\", \"custom\", target], cwd=cwd)\n\n\ndef check_existence_of_variables_file(file_path: pathlib.Path):\n    if not file_path:\n        raise FileNotFoundError(f\"Airflow variables file {file_path} does not exist.\")\n\n\ndef list_subdirs(path: pathlib.Path) -> typing.List[pathlib.Path]:\n    \"\"\"Returns a list of subdirectories\"\"\"\n    subdirs = [f for f in path.iterdir() if f.is_dir() and not f.name[0] in (\".\", \"_\")]\n    return subdirs\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(\n        description=\"Deploy DAGs and variables to an Airflow environment\"\n    )\n    parser.add_argument(\n        \"-d\",\n        \"--dataset\",\n        required=True,\n        type=str,\n        dest=\"dataset\",\n        help=\"The directory name of the dataset.\",\n    )\n    parser.add_argument(\n        \"-e\",\n        \"--env\",\n        type=str,\n        default=\"dev\",\n        dest=\"env\",\n        help=\"The stage used for the resources: dev|staging|prod\",\n    )\n    parser.add_argument(\n        \"-n\",\n        \"--composer-env\",\n        type=str,\n        dest=\"composer_env\",\n        help=\"The Google Cloud Composer environment name\",\n    )\n    parser.add_argument(\n        \"-b\",\n        \"--composer-bucket\",\n        type=str,\n        dest=\"composer_bucket\",\n        help=\"The Google Cloud Composer bucket name\",\n    )\n    parser.add_argument(\n        \"-r\",\n        \"--composer-region\",\n        type=str,\n        dest=\"composer_region\",\n        help=\"The region of the Google Cloud Composer environment\",\n    )\n    parser.add_argument(\n        \"-p\",\n        \"--pipeline\",\n        required=False,\n        type=str,\n        dest=\"pipeline\",\n        help=\"The directory name of the pipeline\",\n    )\n    parser.add_argument(\n        \"-a\",\n        \"--airflow-home\",\n        type=str,\n        default=\"~/airflow\",\n        dest=\"airflow_home\",\n        help=\"pathlib.Path to the Airflow home directory (defaults to `~/airflow`)\",\n    )\n    parser.add_argument(\"--local\", required=False, dest=\"local\", action=\"store_true\")\n\n    args = parser.parse_args()\n    airflow_path = pathlib.Path(args.airflow_home).expanduser()\n\n    if args.local:\n        if not airflow_path.exists() and airflow_path.is_dir():\n            raise ValueError(\n                \"Argument `-a|--airflow-home` must exist and be a directory\"\n            )\n    else:\n        if not args.composer_env:\n            raise ValueError(\n                \"Argument `-n|--composer-env` (Composer environment name) not specified\"\n            )\n\n        if not args.composer_bucket:\n            raise ValueError(\n                \"Argument `-b|--composer-bucket` (Composer bucket name) not specified\"\n            )\n\n        if not args.composer_region:\n            raise ValueError(\n                \"Argument `-r|--composer-region` (Composer environment region) not specified\"\n            )\n\n    main(\n        local=args.local,\n        env_path=PROJECT_ROOT / f\".{args.env}\",\n        dataset_id=args.dataset,\n        pipeline=args.pipeline,\n        airflow_home=airflow_path,\n        composer_env=args.composer_env,\n        composer_bucket=args.composer_bucket,\n        composer_region=args.composer_region,\n    )\n", "path": "scripts/deploy_dag.py"}], "after_files": [{"content": "# Copyright 2021 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\nimport argparse\nimport pathlib\nimport subprocess\nimport typing\nimport warnings\n\nCURRENT_PATH = pathlib.Path(__file__).resolve().parent\nPROJECT_ROOT = CURRENT_PATH.parent\nDATASETS_PATH = PROJECT_ROOT / \"datasets\"\n\n\ndef main(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline: str = None,\n    airflow_home: pathlib.Path = None,\n    composer_env: str = None,\n    composer_bucket: str = None,\n    composer_region: str = None,\n):\n    print(\"\\n========== AIRFLOW VARIABLES ==========\")\n    copy_variables_to_airflow_data_folder(\n        local, env_path, dataset_id, airflow_home, composer_bucket\n    )\n    import_variables_to_airflow_env(\n        local, env_path, dataset_id, composer_env, composer_bucket, composer_region\n    )\n\n    print(\"========== AIRFLOW DAGS ==========\")\n    if pipeline:\n        pipelines = [env_path / \"datasets\" / pipeline]\n    else:\n        pipelines = list_subdirs(env_path / \"datasets\" / dataset_id)\n\n    for pipeline_path in pipelines:\n        copy_custom_callables_to_airflow_dags_folder(\n            local,\n            env_path,\n            dataset_id,\n            pipeline_path.name,\n            composer_bucket,\n            airflow_home,\n        )\n\n        copy_generated_dag_to_airflow_dags_folder(\n            local,\n            env_path,\n            dataset_id,\n            pipeline_path.name,\n            composer_bucket,\n            airflow_home,\n        )\n\n\ndef run_gsutil_cmd(args: typing.List[str], cwd: pathlib.Path):\n    subprocess.check_call([\"gsutil\"] + args, cwd=cwd)\n\n\ndef copy_variables_to_airflow_data_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    airflow_home: pathlib.Path = None,\n    composer_bucket: str = None,\n):\n    \"\"\"\n    cd .{ENV}/{DATASET_ID}\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id\n    filename = f\"{dataset_id}_variables.json\"\n\n    if not (cwd / filename).exists():\n        warnings.warn(f\"Airflow variables file {filename} does not exist.\")\n        return\n\n    if local:\n        \"\"\"\n        cp {DATASET_ID}_variables.json {AIRFLOW_HOME}/data/variables/{DATASET_ID}_variables.json\n        \"\"\"\n        target_path = airflow_home / \"data\" / \"variables\" / filename\n        target_path.mkdir(parents=True, exist_ok=True)\n        print(\n            \"\\nCopying variables JSON file into Airflow data folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target_path}\\n\"\n        )\n\n        subprocess.check_call([\"cp\", \"-rf\", filename, str(target_path)], cwd=cwd)\n    else:\n        \"\"\"\n        [remote]\n        gsutil cp {DATASET_ID}_variables.json gs://{COMPOSER_BUCKET}/data/variables/{DATASET_ID}_variables.json...\n        \"\"\"\n        gcs_uri = f\"gs://{composer_bucket}/data/variables/{filename}\"\n        print(\n            \"\\nCopying variables JSON file into Cloud Composer data folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {gcs_uri}\\n\"\n        )\n        run_gsutil_cmd([\"cp\", filename, gcs_uri], cwd=cwd)\n\n\ndef run_cloud_composer_vars_import(\n    composer_env: str,\n    composer_region: str,\n    airflow_path: pathlib.Path,\n    cwd: pathlib.Path,\n):\n    subprocess.check_call(\n        [\n            \"gcloud\",\n            \"composer\",\n            \"environments\",\n            \"run\",\n            str(composer_env),\n            \"--location\",\n            str(composer_region),\n            \"variables\",\n            \"--\",\n            \"--import\",\n            str(airflow_path),\n        ],\n        cwd=cwd,\n    )\n\n\ndef import_variables_to_airflow_env(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    composer_env: str = None,\n    composer_bucket: str = None,\n    composer_region: str = None,\n):\n    \"\"\"\n    [local]\n    airflow variables import .{ENV}/datasets/{DATASET_ID}/variables.json\n\n    [remote]\n    gcloud composer environments run COMPOSER_ENV --location COMPOSER_REGION variables -- --import /home/airflow/gcs/data/variables/{DATASET_ID}_variables.json\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id\n    filename = f\"{dataset_id}_variables.json\"\n\n    if local:\n        print(f\"\\nImporting Airflow variables from {cwd / filename}...\\n\")\n        subprocess.check_call(\n            [\"airflow\", \"variables\", \"import\", str(cwd / filename)], cwd=cwd\n        )\n    else:\n        gcs_uri = f\"gs://{composer_bucket}/data/variables/{filename}\"\n        airflow_path = f\"/home/airflow/gcs/data/variables/{filename}\"\n        print(f\"\\nImporting Airflow variables from {gcs_uri} ({airflow_path})...\\n\")\n        run_cloud_composer_vars_import(\n            composer_env, composer_region, airflow_path, cwd=cwd\n        )\n\n\ndef copy_generated_dag_to_airflow_dags_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline_id: str,\n    composer_bucket: str = None,\n    airflow_home: pathlib.Path = None,\n):\n    \"\"\"\n    cd {DATASET_ID}/{PIPELINE_ID}\n\n    [local]\n    cp {PIPELINE_ID}_dag.py {AIRFLOW_HOME}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py\n\n    [remote]\n    gsutil cp {PIPELINE_ID}_dag.py gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}__{PIPELINE_ID}_dag.py\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id / pipeline_id\n    filename = f\"{pipeline_id}_dag.py\"\n\n    if local:\n        target = airflow_home / \"dags\" / f\"{dataset_id}__{pipeline_id}_dag.py\"\n        print(\n            f\"\\nCopying DAG file for pipeline `{pipeline_id}` into Airflow DAGs folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        subprocess.check_call([\"cp\", \"-rf\", filename, str(target)], cwd=cwd)\n    else:\n        target = f\"gs://{composer_bucket}/dags/{dataset_id}__{pipeline_id}_dag.py\"\n        print(\n            f\"\\nCopying DAG file for pipeline `{pipeline_id}` into Cloud Composer DAG folder\\n\\n\"\n            f\"  Source:\\n  {cwd / filename}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        run_gsutil_cmd([\"cp\", filename, target], cwd=cwd)\n\n\ndef copy_custom_callables_to_airflow_dags_folder(\n    local: bool,\n    env_path: pathlib.Path,\n    dataset_id: str,\n    pipeline_id: str,\n    composer_bucket: str = None,\n    airflow_home: pathlib.Path = None,\n):\n    \"\"\"\n    cd {DATASET_ID}/{PIPELINE_ID}\n\n    [local]\n    mkdir -p {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom\n    cp -rf custom {AIRFLOW_HOME}/dags/DATASET_ID/PIPELINE_ID/custom\n\n    [remote]\n    gsutil cp -r custom gs://{COMPOSER_BUCKET}/dags/{DATASET_ID}/{PIPELINE_ID}/\n    \"\"\"\n    cwd = env_path / \"datasets\" / dataset_id / pipeline_id\n\n    if not (cwd / \"custom\").exists():\n        return\n\n    if local:\n        target_parent = airflow_home / \"dags\" / dataset_id / pipeline_id\n        target_parent.mkdir(parents=True, exist_ok=True)\n        print(\n            f\"\\nCopying custom callables folder for pipeline `{pipeline_id}` into Airflow DAGs folder\\n\\n\"\n            f\"  Source:\\n  {cwd / 'custom'}\\n\\n\"\n            f\"  Destination:\\n  {target_parent / 'custom'}\\n\"\n        )\n        subprocess.check_call([\"cp\", \"-rf\", \"custom\", str(target_parent)], cwd=cwd)\n    else:\n        target = f\"gs://{composer_bucket}/dags/{dataset_id}/{pipeline_id}/\"\n        print(\n            f\"\\nCopying custom callables folder for pipeline `{pipeline_id}` into Cloud Composer DAG folder\\n\\n\"\n            f\"  Source:\\n  {cwd / 'custom'}\\n\\n\"\n            f\"  Destination:\\n  {target}\\n\"\n        )\n        run_gsutil_cmd([\"cp\", \"-r\", \"custom\", target], cwd=cwd)\n\n\ndef check_existence_of_variables_file(file_path: pathlib.Path):\n    if not file_path:\n        raise FileNotFoundError(f\"Airflow variables file {file_path} does not exist.\")\n\n\ndef list_subdirs(path: pathlib.Path) -> typing.List[pathlib.Path]:\n    \"\"\"Returns a list of subdirectories\"\"\"\n    subdirs = [f for f in path.iterdir() if f.is_dir() and not f.name[0] in (\".\", \"_\")]\n    return subdirs\n\n\nif __name__ == \"__main__\":\n    parser = argparse.ArgumentParser(\n        description=\"Deploy DAGs and variables to an Airflow environment\"\n    )\n    parser.add_argument(\n        \"-d\",\n        \"--dataset\",\n        required=True,\n        type=str,\n        dest=\"dataset\",\n        help=\"The directory name of the dataset.\",\n    )\n    parser.add_argument(\n        \"-e\",\n        \"--env\",\n        type=str,\n        default=\"dev\",\n        dest=\"env\",\n        help=\"The stage used for the resources: dev|staging|prod\",\n    )\n    parser.add_argument(\n        \"-n\",\n        \"--composer-env\",\n        type=str,\n        dest=\"composer_env\",\n        help=\"The Google Cloud Composer environment name\",\n    )\n    parser.add_argument(\n        \"-b\",\n        \"--composer-bucket\",\n        type=str,\n        dest=\"composer_bucket\",\n        help=\"The Google Cloud Composer bucket name\",\n    )\n    parser.add_argument(\n        \"-r\",\n        \"--composer-region\",\n        type=str,\n        dest=\"composer_region\",\n        help=\"The region of the Google Cloud Composer environment\",\n    )\n    parser.add_argument(\n        \"-p\",\n        \"--pipeline\",\n        required=False,\n        type=str,\n        dest=\"pipeline\",\n        help=\"The directory name of the pipeline\",\n    )\n    parser.add_argument(\n        \"-a\",\n        \"--airflow-home\",\n        type=str,\n        default=\"~/airflow\",\n        dest=\"airflow_home\",\n        help=\"pathlib.Path to the Airflow home directory (defaults to `~/airflow`)\",\n    )\n    parser.add_argument(\"--local\", required=False, dest=\"local\", action=\"store_true\")\n\n    args = parser.parse_args()\n    airflow_path = pathlib.Path(args.airflow_home).expanduser()\n\n    if args.local:\n        if not airflow_path.exists() and airflow_path.is_dir():\n            raise ValueError(\n                \"Argument `-a|--airflow-home` must exist and be a directory\"\n            )\n    else:\n        if not args.composer_env:\n            raise ValueError(\n                \"Argument `-n|--composer-env` (Composer environment name) not specified\"\n            )\n\n        if not args.composer_bucket:\n            raise ValueError(\n                \"Argument `-b|--composer-bucket` (Composer bucket name) not specified\"\n            )\n\n        if not args.composer_region:\n            raise ValueError(\n                \"Argument `-r|--composer-region` (Composer environment region) not specified\"\n            )\n\n    main(\n        local=args.local,\n        env_path=PROJECT_ROOT / f\".{args.env}\",\n        dataset_id=args.dataset,\n        pipeline=args.pipeline,\n        airflow_home=airflow_path,\n        composer_env=args.composer_env,\n        composer_bucket=args.composer_bucket,\n        composer_region=args.composer_region,\n    )\n", "path": "scripts/deploy_dag.py"}]}