We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
CI: Fix Snowflake temp tables filling up
@utkarsharma2 @sunank200 - Can one of you add more details to this -- we need to investigate and fix the tests that is filling up our Snowflake instances with tables

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python-sdk/example_dags/example_amazon_s3_snowflake_transform.py`
Content:
```
1 import os
2 import time
3 from datetime import datetime, timedelta
4 
5 import pandas as pd
6 
7 # Uses data from https://www.kaggle.com/c/shelter-animal-outcomes
8 from airflow.decorators import dag
9 
10 from astro import sql as aql
11 from astro.files import File
12 from astro.table import Metadata, Table
13 
14 
15 @aql.transform()
16 def combine_data(center_1: Table, center_2: Table):
17     return """SELECT * FROM {{center_1}}
18     UNION SELECT * FROM {{center_2}}"""
19 
20 
21 @aql.transform()
22 def clean_data(input_table: Table):
23     return """SELECT *
24     FROM {{input_table}} WHERE type NOT LIKE 'Guinea Pig'
25     """
26 
27 
28 # [START dataframe_example_1]
29 @aql.dataframe(columns_names_capitalization="original")
30 def aggregate_data(df: pd.DataFrame):
31     new_df = df.pivot_table(index="date", values="name", columns=["type"], aggfunc="count").reset_index()
32     new_df.columns = new_df.columns.str.lower()
33     return new_df
34 
35 
36 # [END dataframe_example_1]
37 
38 
39 @dag(
40     start_date=datetime(2021, 1, 1),
41     max_active_runs=1,
42     schedule_interval="@daily",
43     default_args={
44         "email_on_failure": False,
45         "retries": 0,
46         "retry_delay": timedelta(minutes=5),
47     },
48     catchup=False,
49 )
50 def example_amazon_s3_snowflake_transform():
51 
52     s3_bucket = os.getenv("S3_BUCKET", "s3://tmp9")
53 
54     input_table_1 = Table(
55         name="ADOPTION_CENTER_1",
56         metadata=Metadata(
57             database=os.environ["SNOWFLAKE_DATABASE"],
58             schema=os.environ["SNOWFLAKE_SCHEMA"],
59         ),
60         conn_id="snowflake_conn",
61     )
62     # [START metadata_example_snowflake]
63     input_table_2 = Table(
64         name="ADOPTION_CENTER_2",
65         metadata=Metadata(
66             database=os.environ["SNOWFLAKE_DATABASE"],
67             schema=os.environ["SNOWFLAKE_SCHEMA"],
68         ),
69         conn_id="snowflake_conn",
70     )
71     # [END metadata_example_snowflake]
72 
73     temp_table_1 = aql.load_file(
74         input_file=File(path=f"{s3_bucket}/ADOPTION_CENTER_1_unquoted.csv"),
75         output_table=input_table_1,
76     )
77     temp_table_2 = aql.load_file(
78         input_file=File(path=f"{s3_bucket}/ADOPTION_CENTER_2_unquoted.csv"),
79         output_table=input_table_2,
80     )
81 
82     combined_data = combine_data(
83         center_1=temp_table_1,
84         center_2=temp_table_2,
85     )
86 
87     cleaned_data = clean_data(combined_data)
88     # [START dataframe_example_2]
89     aggregate_data(
90         cleaned_data,
91         output_table=Table(
92             name="aggregated_adoptions_" + str(int(time.time())),
93             metadata=Metadata(
94                 schema=os.environ["SNOWFLAKE_SCHEMA"],
95                 database=os.environ["SNOWFLAKE_DATABASE"],
96             ),
97             conn_id="snowflake_conn",
98         ),
99     )
100     # [END dataframe_example_2]
101     aql.cleanup()
102 
103 
104 dag = example_amazon_s3_snowflake_transform()
105
```

Path: `python-sdk/example_dags/example_snowflake_partial_table_with_append.py`
Content:
```
1 """
2 Example ETL DAG highlighting Astro functionality
3 DAG requires 2 "Homes" csv's (found in this repo), and a supported database
4 General flow of the DAG is to extract the data from csv's and combine using SQL,
5 then switch to Python for a melt transformation, then back to SQL for final
6 filtering. The data is then loaded by appending to an existing reporting table.
7 
8 This example DAG creates the reporting table & truncates it by the end of the execution.
9 """
10 
11 import os
12 from datetime import datetime
13 
14 import pandas as pd
15 from airflow.decorators import dag
16 
17 from astro.files import File
18 from astro.sql import append, cleanup, dataframe, load_file, run_raw_sql, transform
19 from astro.table import Metadata, Table
20 
21 SNOWFLAKE_CONN_ID = "snowflake_conn"
22 dir_path = os.path.dirname(os.path.realpath(__file__))
23 
24 FILE_PATH = dir_path + "/data/"
25 
26 
27 # The first transformation combines data from the two source csv's
28 @transform
29 def extract_data(homes1: Table, homes2: Table):
30     return """
31     SELECT *
32     FROM {{homes1}}
33     UNION
34     SELECT *
35     FROM {{homes2}}
36     """
37 
38 
39 # Switch to Python (Pandas) for melting transformation to get data into long format
40 @dataframe
41 def transform_data(df: pd.DataFrame):
42     df.columns = df.columns.str.lower()
43     melted_df = df.melt(id_vars=["sell", "list"], value_vars=["living", "rooms", "beds", "baths", "age"])
44 
45     return melted_df
46 
47 
48 # Back to SQL to filter data
49 @transform
50 def filter_data(homes_long: Table):
51     return """
52     SELECT *
53     FROM {{homes_long}}
54     WHERE SELL > 200
55     """
56 
57 
58 # [START howto_run_raw_sql_snowflake_1]
59 @run_raw_sql
60 def create_table(table: Table):
61     """Create the reporting data which will be the target of the append method"""
62     return """
63     CREATE TABLE IF NOT EXISTS {{table}} (
64       sell number,
65       list number,
66       variable varchar,
67       value number
68     );
69     """
70 
71 
72 @dag(start_date=datetime(2021, 12, 1), schedule_interval="@daily", catchup=False)
73 def example_snowflake_partial_table_with_append():
74     homes_reporting = Table(conn_id=SNOWFLAKE_CONN_ID)
75     create_results_table = create_table(table=homes_reporting, conn_id=SNOWFLAKE_CONN_ID)
76     # [END howto_run_raw_sql_snowflake_1]
77 
78     # Initial load of homes data csv's into Snowflake
79     homes_data1 = load_file(
80         input_file=File(path=FILE_PATH + "homes.csv"),
81         output_table=Table(
82             conn_id=SNOWFLAKE_CONN_ID,
83             metadata=Metadata(
84                 database=os.getenv("SNOWFLAKE_DATABASE"),
85                 schema=os.getenv("SNOWFLAKE_SCHEMA"),
86             ),
87         ),
88     )
89 
90     homes_data2 = load_file(
91         input_file=File(path=FILE_PATH + "homes2.csv"),
92         output_table=Table(
93             conn_id=SNOWFLAKE_CONN_ID,
94             metadata=Metadata(
95                 database=os.getenv("SNOWFLAKE_DATABASE"),
96                 schema=os.getenv("SNOWFLAKE_SCHEMA"),
97             ),
98         ),
99     )
100 
101     # Define task dependencies
102     extracted_data = extract_data(
103         homes1=homes_data1,
104         homes2=homes_data2,
105         output_table=Table(name="combined_homes_data"),
106     )
107 
108     transformed_data = transform_data(df=extracted_data, output_table=Table(name="homes_data_long"))
109 
110     filtered_data = filter_data(
111         homes_long=transformed_data,
112         output_table=Table(),
113     )
114 
115     # Append transformed & filtered data to reporting table
116     # Dependency is inferred by passing the previous `filtered_data` task to `append_table` param
117     # [START append_example_with_columns_list]
118     record_results = append(
119         source_table=filtered_data,
120         target_table=homes_reporting,
121         columns=["sell", "list", "variable", "value"],
122     )
123     # [END append_example_with_columns_list]
124     record_results.set_upstream(create_results_table)
125 
126     cleanup()
127 
128 
129 example_snowflake_partial_table_dag = example_snowflake_partial_table_with_append()
130
```
--- 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/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py b/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py
--- a/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py
+++ b/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py
@@ -52,21 +52,23 @@
     s3_bucket = os.getenv("S3_BUCKET", "s3://tmp9")
 
     input_table_1 = Table(
-        name="ADOPTION_CENTER_1",
+        name="ADOPTION_CENTER_1_" + str(int(time.time())),
         metadata=Metadata(
             database=os.environ["SNOWFLAKE_DATABASE"],
             schema=os.environ["SNOWFLAKE_SCHEMA"],
         ),
         conn_id="snowflake_conn",
+        temp=True,
     )
     # [START metadata_example_snowflake]
     input_table_2 = Table(
-        name="ADOPTION_CENTER_2",
+        name="ADOPTION_CENTER_2_" + str(int(time.time())),
         metadata=Metadata(
             database=os.environ["SNOWFLAKE_DATABASE"],
             schema=os.environ["SNOWFLAKE_SCHEMA"],
         ),
         conn_id="snowflake_conn",
+        temp=True,
     )
     # [END metadata_example_snowflake]
 
@@ -86,16 +88,18 @@
 
     cleaned_data = clean_data(combined_data)
     # [START dataframe_example_2]
+    snowflake_output_table = Table(
+        name="aggregated_adoptions_" + str(int(time.time())),
+        metadata=Metadata(
+            schema=os.environ["SNOWFLAKE_SCHEMA"],
+            database=os.environ["SNOWFLAKE_DATABASE"],
+        ),
+        conn_id="snowflake_conn",
+        temp=True,
+    )
     aggregate_data(
         cleaned_data,
-        output_table=Table(
-            name="aggregated_adoptions_" + str(int(time.time())),
-            metadata=Metadata(
-                schema=os.environ["SNOWFLAKE_SCHEMA"],
-                database=os.environ["SNOWFLAKE_DATABASE"],
-            ),
-            conn_id="snowflake_conn",
-        ),
+        output_table=snowflake_output_table,
     )
     # [END dataframe_example_2]
     aql.cleanup()
diff --git a/python-sdk/example_dags/example_snowflake_partial_table_with_append.py b/python-sdk/example_dags/example_snowflake_partial_table_with_append.py
--- a/python-sdk/example_dags/example_snowflake_partial_table_with_append.py
+++ b/python-sdk/example_dags/example_snowflake_partial_table_with_append.py
@@ -9,6 +9,7 @@
 """
 
 import os
+import time
 from datetime import datetime
 
 import pandas as pd
@@ -60,7 +61,7 @@
 def create_table(table: Table):
     """Create the reporting data which will be the target of the append method"""
     return """
-    CREATE TABLE IF NOT EXISTS {{table}} (
+      CREATE OR REPLACE TABLE {{table}} (
       sell number,
       list number,
       variable varchar,
@@ -71,7 +72,7 @@
 
 @dag(start_date=datetime(2021, 12, 1), schedule_interval="@daily", catchup=False)
 def example_snowflake_partial_table_with_append():
-    homes_reporting = Table(conn_id=SNOWFLAKE_CONN_ID)
+    homes_reporting = Table(name="homes_reporting_data", temp=True, conn_id=SNOWFLAKE_CONN_ID)
     create_results_table = create_table(table=homes_reporting, conn_id=SNOWFLAKE_CONN_ID)
     # [END howto_run_raw_sql_snowflake_1]
 
@@ -102,10 +103,12 @@
     extracted_data = extract_data(
         homes1=homes_data1,
         homes2=homes_data2,
-        output_table=Table(name="combined_homes_data"),
+        output_table=Table(name="combined_homes_data_" + str(int(time.time())), temp=True),
     )
 
-    transformed_data = transform_data(df=extracted_data, output_table=Table(name="homes_data_long"))
+    transformed_data = transform_data(
+        df=extracted_data, output_table=Table(name="homes_data_long_" + str(int(time.time())), temp=True)
+    )
 
     filtered_data = filter_data(
         homes_long=transformed_data,

{"golden_diff": "diff --git a/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py b/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py\n--- a/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py\n+++ b/python-sdk/example_dags/example_amazon_s3_snowflake_transform.py\n@@ -52,21 +52,23 @@\n     s3_bucket = os.getenv(\"S3_BUCKET\", \"s3://tmp9\")\n \n     input_table_1 = Table(\n-        name=\"ADOPTION_CENTER_1\",\n+        name=\"ADOPTION_CENTER_1_\" + str(int(time.time())),\n         metadata=Metadata(\n             database=os.environ[\"SNOWFLAKE_DATABASE\"],\n             schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n         ),\n         conn_id=\"snowflake_conn\",\n+        temp=True,\n     )\n     # [START metadata_example_snowflake]\n     input_table_2 = Table(\n-        name=\"ADOPTION_CENTER_2\",\n+        name=\"ADOPTION_CENTER_2_\" + str(int(time.time())),\n         metadata=Metadata(\n             database=os.environ[\"SNOWFLAKE_DATABASE\"],\n             schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n         ),\n         conn_id=\"snowflake_conn\",\n+        temp=True,\n     )\n     # [END metadata_example_snowflake]\n \n@@ -86,16 +88,18 @@\n \n     cleaned_data = clean_data(combined_data)\n     # [START dataframe_example_2]\n+    snowflake_output_table = Table(\n+        name=\"aggregated_adoptions_\" + str(int(time.time())),\n+        metadata=Metadata(\n+            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n+            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n+        ),\n+        conn_id=\"snowflake_conn\",\n+        temp=True,\n+    )\n     aggregate_data(\n         cleaned_data,\n-        output_table=Table(\n-            name=\"aggregated_adoptions_\" + str(int(time.time())),\n-            metadata=Metadata(\n-                schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n-                database=os.environ[\"SNOWFLAKE_DATABASE\"],\n-            ),\n-            conn_id=\"snowflake_conn\",\n-        ),\n+        output_table=snowflake_output_table,\n     )\n     # [END dataframe_example_2]\n     aql.cleanup()\ndiff --git a/python-sdk/example_dags/example_snowflake_partial_table_with_append.py b/python-sdk/example_dags/example_snowflake_partial_table_with_append.py\n--- a/python-sdk/example_dags/example_snowflake_partial_table_with_append.py\n+++ b/python-sdk/example_dags/example_snowflake_partial_table_with_append.py\n@@ -9,6 +9,7 @@\n \"\"\"\n \n import os\n+import time\n from datetime import datetime\n \n import pandas as pd\n@@ -60,7 +61,7 @@\n def create_table(table: Table):\n     \"\"\"Create the reporting data which will be the target of the append method\"\"\"\n     return \"\"\"\n-    CREATE TABLE IF NOT EXISTS {{table}} (\n+      CREATE OR REPLACE TABLE {{table}} (\n       sell number,\n       list number,\n       variable varchar,\n@@ -71,7 +72,7 @@\n \n @dag(start_date=datetime(2021, 12, 1), schedule_interval=\"@daily\", catchup=False)\n def example_snowflake_partial_table_with_append():\n-    homes_reporting = Table(conn_id=SNOWFLAKE_CONN_ID)\n+    homes_reporting = Table(name=\"homes_reporting_data\", temp=True, conn_id=SNOWFLAKE_CONN_ID)\n     create_results_table = create_table(table=homes_reporting, conn_id=SNOWFLAKE_CONN_ID)\n     # [END howto_run_raw_sql_snowflake_1]\n \n@@ -102,10 +103,12 @@\n     extracted_data = extract_data(\n         homes1=homes_data1,\n         homes2=homes_data2,\n-        output_table=Table(name=\"combined_homes_data\"),\n+        output_table=Table(name=\"combined_homes_data_\" + str(int(time.time())), temp=True),\n     )\n \n-    transformed_data = transform_data(df=extracted_data, output_table=Table(name=\"homes_data_long\"))\n+    transformed_data = transform_data(\n+        df=extracted_data, output_table=Table(name=\"homes_data_long_\" + str(int(time.time())), temp=True)\n+    )\n \n     filtered_data = filter_data(\n         homes_long=transformed_data,\n", "issue": "CI: Fix Snowflake temp tables filling up\n@utkarsharma2 @sunank200 - Can one of you add more details to this -- we need to investigate and fix the tests that is filling up our Snowflake instances with tables\n", "before_files": [{"content": "import os\nimport time\nfrom datetime import datetime, timedelta\n\nimport pandas as pd\n\n# Uses data from https://www.kaggle.com/c/shelter-animal-outcomes\nfrom airflow.decorators import dag\n\nfrom astro import sql as aql\nfrom astro.files import File\nfrom astro.table import Metadata, Table\n\n\n@aql.transform()\ndef combine_data(center_1: Table, center_2: Table):\n    return \"\"\"SELECT * FROM {{center_1}}\n    UNION SELECT * FROM {{center_2}}\"\"\"\n\n\n@aql.transform()\ndef clean_data(input_table: Table):\n    return \"\"\"SELECT *\n    FROM {{input_table}} WHERE type NOT LIKE 'Guinea Pig'\n    \"\"\"\n\n\n# [START dataframe_example_1]\n@aql.dataframe(columns_names_capitalization=\"original\")\ndef aggregate_data(df: pd.DataFrame):\n    new_df = df.pivot_table(index=\"date\", values=\"name\", columns=[\"type\"], aggfunc=\"count\").reset_index()\n    new_df.columns = new_df.columns.str.lower()\n    return new_df\n\n\n# [END dataframe_example_1]\n\n\n@dag(\n    start_date=datetime(2021, 1, 1),\n    max_active_runs=1,\n    schedule_interval=\"@daily\",\n    default_args={\n        \"email_on_failure\": False,\n        \"retries\": 0,\n        \"retry_delay\": timedelta(minutes=5),\n    },\n    catchup=False,\n)\ndef example_amazon_s3_snowflake_transform():\n\n    s3_bucket = os.getenv(\"S3_BUCKET\", \"s3://tmp9\")\n\n    input_table_1 = Table(\n        name=\"ADOPTION_CENTER_1\",\n        metadata=Metadata(\n            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n        ),\n        conn_id=\"snowflake_conn\",\n    )\n    # [START metadata_example_snowflake]\n    input_table_2 = Table(\n        name=\"ADOPTION_CENTER_2\",\n        metadata=Metadata(\n            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n        ),\n        conn_id=\"snowflake_conn\",\n    )\n    # [END metadata_example_snowflake]\n\n    temp_table_1 = aql.load_file(\n        input_file=File(path=f\"{s3_bucket}/ADOPTION_CENTER_1_unquoted.csv\"),\n        output_table=input_table_1,\n    )\n    temp_table_2 = aql.load_file(\n        input_file=File(path=f\"{s3_bucket}/ADOPTION_CENTER_2_unquoted.csv\"),\n        output_table=input_table_2,\n    )\n\n    combined_data = combine_data(\n        center_1=temp_table_1,\n        center_2=temp_table_2,\n    )\n\n    cleaned_data = clean_data(combined_data)\n    # [START dataframe_example_2]\n    aggregate_data(\n        cleaned_data,\n        output_table=Table(\n            name=\"aggregated_adoptions_\" + str(int(time.time())),\n            metadata=Metadata(\n                schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n                database=os.environ[\"SNOWFLAKE_DATABASE\"],\n            ),\n            conn_id=\"snowflake_conn\",\n        ),\n    )\n    # [END dataframe_example_2]\n    aql.cleanup()\n\n\ndag = example_amazon_s3_snowflake_transform()\n", "path": "python-sdk/example_dags/example_amazon_s3_snowflake_transform.py"}, {"content": "\"\"\"\nExample ETL DAG highlighting Astro functionality\nDAG requires 2 \"Homes\" csv's (found in this repo), and a supported database\nGeneral flow of the DAG is to extract the data from csv's and combine using SQL,\nthen switch to Python for a melt transformation, then back to SQL for final\nfiltering. The data is then loaded by appending to an existing reporting table.\n\nThis example DAG creates the reporting table & truncates it by the end of the execution.\n\"\"\"\n\nimport os\nfrom datetime import datetime\n\nimport pandas as pd\nfrom airflow.decorators import dag\n\nfrom astro.files import File\nfrom astro.sql import append, cleanup, dataframe, load_file, run_raw_sql, transform\nfrom astro.table import Metadata, Table\n\nSNOWFLAKE_CONN_ID = \"snowflake_conn\"\ndir_path = os.path.dirname(os.path.realpath(__file__))\n\nFILE_PATH = dir_path + \"/data/\"\n\n\n# The first transformation combines data from the two source csv's\n@transform\ndef extract_data(homes1: Table, homes2: Table):\n    return \"\"\"\n    SELECT *\n    FROM {{homes1}}\n    UNION\n    SELECT *\n    FROM {{homes2}}\n    \"\"\"\n\n\n# Switch to Python (Pandas) for melting transformation to get data into long format\n@dataframe\ndef transform_data(df: pd.DataFrame):\n    df.columns = df.columns.str.lower()\n    melted_df = df.melt(id_vars=[\"sell\", \"list\"], value_vars=[\"living\", \"rooms\", \"beds\", \"baths\", \"age\"])\n\n    return melted_df\n\n\n# Back to SQL to filter data\n@transform\ndef filter_data(homes_long: Table):\n    return \"\"\"\n    SELECT *\n    FROM {{homes_long}}\n    WHERE SELL > 200\n    \"\"\"\n\n\n# [START howto_run_raw_sql_snowflake_1]\n@run_raw_sql\ndef create_table(table: Table):\n    \"\"\"Create the reporting data which will be the target of the append method\"\"\"\n    return \"\"\"\n    CREATE TABLE IF NOT EXISTS {{table}} (\n      sell number,\n      list number,\n      variable varchar,\n      value number\n    );\n    \"\"\"\n\n\n@dag(start_date=datetime(2021, 12, 1), schedule_interval=\"@daily\", catchup=False)\ndef example_snowflake_partial_table_with_append():\n    homes_reporting = Table(conn_id=SNOWFLAKE_CONN_ID)\n    create_results_table = create_table(table=homes_reporting, conn_id=SNOWFLAKE_CONN_ID)\n    # [END howto_run_raw_sql_snowflake_1]\n\n    # Initial load of homes data csv's into Snowflake\n    homes_data1 = load_file(\n        input_file=File(path=FILE_PATH + \"homes.csv\"),\n        output_table=Table(\n            conn_id=SNOWFLAKE_CONN_ID,\n            metadata=Metadata(\n                database=os.getenv(\"SNOWFLAKE_DATABASE\"),\n                schema=os.getenv(\"SNOWFLAKE_SCHEMA\"),\n            ),\n        ),\n    )\n\n    homes_data2 = load_file(\n        input_file=File(path=FILE_PATH + \"homes2.csv\"),\n        output_table=Table(\n            conn_id=SNOWFLAKE_CONN_ID,\n            metadata=Metadata(\n                database=os.getenv(\"SNOWFLAKE_DATABASE\"),\n                schema=os.getenv(\"SNOWFLAKE_SCHEMA\"),\n            ),\n        ),\n    )\n\n    # Define task dependencies\n    extracted_data = extract_data(\n        homes1=homes_data1,\n        homes2=homes_data2,\n        output_table=Table(name=\"combined_homes_data\"),\n    )\n\n    transformed_data = transform_data(df=extracted_data, output_table=Table(name=\"homes_data_long\"))\n\n    filtered_data = filter_data(\n        homes_long=transformed_data,\n        output_table=Table(),\n    )\n\n    # Append transformed & filtered data to reporting table\n    # Dependency is inferred by passing the previous `filtered_data` task to `append_table` param\n    # [START append_example_with_columns_list]\n    record_results = append(\n        source_table=filtered_data,\n        target_table=homes_reporting,\n        columns=[\"sell\", \"list\", \"variable\", \"value\"],\n    )\n    # [END append_example_with_columns_list]\n    record_results.set_upstream(create_results_table)\n\n    cleanup()\n\n\nexample_snowflake_partial_table_dag = example_snowflake_partial_table_with_append()\n", "path": "python-sdk/example_dags/example_snowflake_partial_table_with_append.py"}], "after_files": [{"content": "import os\nimport time\nfrom datetime import datetime, timedelta\n\nimport pandas as pd\n\n# Uses data from https://www.kaggle.com/c/shelter-animal-outcomes\nfrom airflow.decorators import dag\n\nfrom astro import sql as aql\nfrom astro.files import File\nfrom astro.table import Metadata, Table\n\n\n@aql.transform()\ndef combine_data(center_1: Table, center_2: Table):\n    return \"\"\"SELECT * FROM {{center_1}}\n    UNION SELECT * FROM {{center_2}}\"\"\"\n\n\n@aql.transform()\ndef clean_data(input_table: Table):\n    return \"\"\"SELECT *\n    FROM {{input_table}} WHERE type NOT LIKE 'Guinea Pig'\n    \"\"\"\n\n\n# [START dataframe_example_1]\n@aql.dataframe(columns_names_capitalization=\"original\")\ndef aggregate_data(df: pd.DataFrame):\n    new_df = df.pivot_table(index=\"date\", values=\"name\", columns=[\"type\"], aggfunc=\"count\").reset_index()\n    new_df.columns = new_df.columns.str.lower()\n    return new_df\n\n\n# [END dataframe_example_1]\n\n\n@dag(\n    start_date=datetime(2021, 1, 1),\n    max_active_runs=1,\n    schedule_interval=\"@daily\",\n    default_args={\n        \"email_on_failure\": False,\n        \"retries\": 0,\n        \"retry_delay\": timedelta(minutes=5),\n    },\n    catchup=False,\n)\ndef example_amazon_s3_snowflake_transform():\n\n    s3_bucket = os.getenv(\"S3_BUCKET\", \"s3://tmp9\")\n\n    input_table_1 = Table(\n        name=\"ADOPTION_CENTER_1_\" + str(int(time.time())),\n        metadata=Metadata(\n            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n        ),\n        conn_id=\"snowflake_conn\",\n        temp=True,\n    )\n    # [START metadata_example_snowflake]\n    input_table_2 = Table(\n        name=\"ADOPTION_CENTER_2_\" + str(int(time.time())),\n        metadata=Metadata(\n            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n        ),\n        conn_id=\"snowflake_conn\",\n        temp=True,\n    )\n    # [END metadata_example_snowflake]\n\n    temp_table_1 = aql.load_file(\n        input_file=File(path=f\"{s3_bucket}/ADOPTION_CENTER_1_unquoted.csv\"),\n        output_table=input_table_1,\n    )\n    temp_table_2 = aql.load_file(\n        input_file=File(path=f\"{s3_bucket}/ADOPTION_CENTER_2_unquoted.csv\"),\n        output_table=input_table_2,\n    )\n\n    combined_data = combine_data(\n        center_1=temp_table_1,\n        center_2=temp_table_2,\n    )\n\n    cleaned_data = clean_data(combined_data)\n    # [START dataframe_example_2]\n    snowflake_output_table = Table(\n        name=\"aggregated_adoptions_\" + str(int(time.time())),\n        metadata=Metadata(\n            schema=os.environ[\"SNOWFLAKE_SCHEMA\"],\n            database=os.environ[\"SNOWFLAKE_DATABASE\"],\n        ),\n        conn_id=\"snowflake_conn\",\n        temp=True,\n    )\n    aggregate_data(\n        cleaned_data,\n        output_table=snowflake_output_table,\n    )\n    # [END dataframe_example_2]\n    aql.cleanup()\n\n\ndag = example_amazon_s3_snowflake_transform()\n", "path": "python-sdk/example_dags/example_amazon_s3_snowflake_transform.py"}, {"content": "\"\"\"\nExample ETL DAG highlighting Astro functionality\nDAG requires 2 \"Homes\" csv's (found in this repo), and a supported database\nGeneral flow of the DAG is to extract the data from csv's and combine using SQL,\nthen switch to Python for a melt transformation, then back to SQL for final\nfiltering. The data is then loaded by appending to an existing reporting table.\n\nThis example DAG creates the reporting table & truncates it by the end of the execution.\n\"\"\"\n\nimport os\nimport time\nfrom datetime import datetime\n\nimport pandas as pd\nfrom airflow.decorators import dag\n\nfrom astro.files import File\nfrom astro.sql import append, cleanup, dataframe, load_file, run_raw_sql, transform\nfrom astro.table import Metadata, Table\n\nSNOWFLAKE_CONN_ID = \"snowflake_conn\"\ndir_path = os.path.dirname(os.path.realpath(__file__))\n\nFILE_PATH = dir_path + \"/data/\"\n\n\n# The first transformation combines data from the two source csv's\n@transform\ndef extract_data(homes1: Table, homes2: Table):\n    return \"\"\"\n    SELECT *\n    FROM {{homes1}}\n    UNION\n    SELECT *\n    FROM {{homes2}}\n    \"\"\"\n\n\n# Switch to Python (Pandas) for melting transformation to get data into long format\n@dataframe\ndef transform_data(df: pd.DataFrame):\n    df.columns = df.columns.str.lower()\n    melted_df = df.melt(id_vars=[\"sell\", \"list\"], value_vars=[\"living\", \"rooms\", \"beds\", \"baths\", \"age\"])\n\n    return melted_df\n\n\n# Back to SQL to filter data\n@transform\ndef filter_data(homes_long: Table):\n    return \"\"\"\n    SELECT *\n    FROM {{homes_long}}\n    WHERE SELL > 200\n    \"\"\"\n\n\n# [START howto_run_raw_sql_snowflake_1]\n@run_raw_sql\ndef create_table(table: Table):\n    \"\"\"Create the reporting data which will be the target of the append method\"\"\"\n    return \"\"\"\n      CREATE OR REPLACE TABLE {{table}} (\n      sell number,\n      list number,\n      variable varchar,\n      value number\n    );\n    \"\"\"\n\n\n@dag(start_date=datetime(2021, 12, 1), schedule_interval=\"@daily\", catchup=False)\ndef example_snowflake_partial_table_with_append():\n    homes_reporting = Table(name=\"homes_reporting_data\", temp=True, conn_id=SNOWFLAKE_CONN_ID)\n    create_results_table = create_table(table=homes_reporting, conn_id=SNOWFLAKE_CONN_ID)\n    # [END howto_run_raw_sql_snowflake_1]\n\n    # Initial load of homes data csv's into Snowflake\n    homes_data1 = load_file(\n        input_file=File(path=FILE_PATH + \"homes.csv\"),\n        output_table=Table(\n            conn_id=SNOWFLAKE_CONN_ID,\n            metadata=Metadata(\n                database=os.getenv(\"SNOWFLAKE_DATABASE\"),\n                schema=os.getenv(\"SNOWFLAKE_SCHEMA\"),\n            ),\n        ),\n    )\n\n    homes_data2 = load_file(\n        input_file=File(path=FILE_PATH + \"homes2.csv\"),\n        output_table=Table(\n            conn_id=SNOWFLAKE_CONN_ID,\n            metadata=Metadata(\n                database=os.getenv(\"SNOWFLAKE_DATABASE\"),\n                schema=os.getenv(\"SNOWFLAKE_SCHEMA\"),\n            ),\n        ),\n    )\n\n    # Define task dependencies\n    extracted_data = extract_data(\n        homes1=homes_data1,\n        homes2=homes_data2,\n        output_table=Table(name=\"combined_homes_data_\" + str(int(time.time())), temp=True),\n    )\n\n    transformed_data = transform_data(\n        df=extracted_data, output_table=Table(name=\"homes_data_long_\" + str(int(time.time())), temp=True)\n    )\n\n    filtered_data = filter_data(\n        homes_long=transformed_data,\n        output_table=Table(),\n    )\n\n    # Append transformed & filtered data to reporting table\n    # Dependency is inferred by passing the previous `filtered_data` task to `append_table` param\n    # [START append_example_with_columns_list]\n    record_results = append(\n        source_table=filtered_data,\n        target_table=homes_reporting,\n        columns=[\"sell\", \"list\", \"variable\", \"value\"],\n    )\n    # [END append_example_with_columns_list]\n    record_results.set_upstream(create_results_table)\n\n    cleanup()\n\n\nexample_snowflake_partial_table_dag = example_snowflake_partial_table_with_append()\n", "path": "python-sdk/example_dags/example_snowflake_partial_table_with_append.py"}]}