We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Change `export_file` to return `File` object
**Context**

In order to allow users to perform subsequent actions on an exported file (while maintaining a functional structure), we should allow the `export_file` function to return a file object.

* Astro-SDK version: 0.9..1
* Request by: @jlaneve
* Analysed by @dimberman 

**Problem**

At the moment a user who wants to use the `output_file` object would need to explicitly set dependencies like this:

```
    output_file = File(path="/tmp/saved_df.csv")
    with sample_dag:
        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)
        export = aql.export_file(
            input_data=table,
            output_file=output_file,
            if_exists="replace",
        )
        res_df = aql.load_file(input_file=output_file)
        export >> res_df
```

**Desired behaviour**

```
    with sample_dag:
        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)
        exported_file = aql.export_file(
            input_data=table,
            output_file=File(path="/tmp/saved_df.csv"),
            if_exists="replace",
        )
        res_df = aql.load_file(input_file=exported_file)
```

**Acceptance criteria**
* Change `export_file` so it returns the `File` instance, as opposed to `None`

Since there is no documentation about this task, we don't need to update the documentation for it. To create documentation for this feature should be part of another issue.

--- 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/astro/sql/operators/export_file.py`
Content:
```
1 from typing import Optional, Union
2 
3 import pandas as pd
4 from airflow.models import BaseOperator
5 from airflow.models.xcom_arg import XComArg
6 
7 from astro.constants import ExportExistsStrategy
8 from astro.databases import create_database
9 from astro.files import File
10 from astro.sql.table import Table
11 from astro.utils.task_id_helper import get_task_id
12 
13 
14 class ExportFile(BaseOperator):
15     """Write SQL table to csv/parquet on local/S3/GCS.
16 
17     :param input_data: Table to convert to file
18     :param output_file: File object containing the path to the file and connection id.
19     :param if_exists: Overwrite file if exists. Default False.
20     """
21 
22     template_fields = ("input_data", "output_file")
23 
24     def __init__(
25         self,
26         input_data: Union[Table, pd.DataFrame],
27         output_file: File,
28         if_exists: ExportExistsStrategy = "exception",
29         **kwargs,
30     ) -> None:
31         super().__init__(**kwargs)
32         self.output_file = output_file
33         self.input_data = input_data
34         self.if_exists = if_exists
35         self.kwargs = kwargs
36 
37     def execute(self, context: dict) -> None:
38         """Write SQL table to csv/parquet on local/S3/GCS.
39 
40         Infers SQL database type based on connection.
41         """
42         # Infer db type from `input_conn_id`.
43         if isinstance(self.input_data, Table):
44             database = create_database(self.input_data.conn_id)
45             self.input_data = database.populate_table_metadata(self.input_data)
46             df = database.export_table_to_pandas_dataframe(self.input_data)
47         elif isinstance(self.input_data, pd.DataFrame):
48             df = self.input_data
49         else:
50             raise ValueError(
51                 f"Expected input_table to be Table or dataframe. Got {type(self.input_data)}"
52             )
53         # Write file if overwrite == True or if file doesn't exist.
54         if self.if_exists == "replace" or not self.output_file.exists():
55             self.output_file.create_from_dataframe(df)
56         else:
57             raise FileExistsError(f"{self.output_file.path} file already exists.")
58 
59 
60 def export_file(
61     input_data: Union[Table, pd.DataFrame],
62     output_file: File,
63     if_exists: ExportExistsStrategy = "exception",
64     task_id: Optional[str] = None,
65     **kwargs,
66 ) -> XComArg:
67     """Convert SaveFile into a function. Returns XComArg.
68 
69     Returns an XComArg object.
70 
71     :param output_file: Path and conn_id
72     :param input_data: Input table / dataframe
73     :param if_exists: Overwrite file if exists. Default "exception"
74     :param task_id: task id, optional
75     """
76 
77     task_id = (
78         task_id if task_id is not None else get_task_id("export_file", output_file.path)
79     )
80 
81     return ExportFile(
82         task_id=task_id,
83         output_file=output_file,
84         input_data=input_data,
85         if_exists=if_exists,
86     ).output
87
```
--- 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/astro/sql/operators/export_file.py b/src/astro/sql/operators/export_file.py
--- a/src/astro/sql/operators/export_file.py
+++ b/src/astro/sql/operators/export_file.py
@@ -34,7 +34,7 @@
         self.if_exists = if_exists
         self.kwargs = kwargs
 
-    def execute(self, context: dict) -> None:
+    def execute(self, context: dict) -> File:
         """Write SQL table to csv/parquet on local/S3/GCS.
 
         Infers SQL database type based on connection.
@@ -53,6 +53,7 @@
         # Write file if overwrite == True or if file doesn't exist.
         if self.if_exists == "replace" or not self.output_file.exists():
             self.output_file.create_from_dataframe(df)
+            return self.output_file
         else:
             raise FileExistsError(f"{self.output_file.path} file already exists.")
 
@@ -66,7 +67,20 @@
 ) -> XComArg:
     """Convert SaveFile into a function. Returns XComArg.
 
-    Returns an XComArg object.
+    Returns an XComArg object of type File which matches the output_file parameter.
+
+    This will allow users to perform further actions with the exported file.
+
+    e.g.
+
+        with sample_dag:
+        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)
+        exported_file = aql.export_file(
+            input_data=table,
+            output_file=File(path="/tmp/saved_df.csv"),
+            if_exists="replace",
+        )
+        res_df = aql.load_file(input_file=exported_file)
 
     :param output_file: Path and conn_id
     :param input_data: Input table / dataframe

{"golden_diff": "diff --git a/src/astro/sql/operators/export_file.py b/src/astro/sql/operators/export_file.py\n--- a/src/astro/sql/operators/export_file.py\n+++ b/src/astro/sql/operators/export_file.py\n@@ -34,7 +34,7 @@\n         self.if_exists = if_exists\n         self.kwargs = kwargs\n \n-    def execute(self, context: dict) -> None:\n+    def execute(self, context: dict) -> File:\n         \"\"\"Write SQL table to csv/parquet on local/S3/GCS.\n \n         Infers SQL database type based on connection.\n@@ -53,6 +53,7 @@\n         # Write file if overwrite == True or if file doesn't exist.\n         if self.if_exists == \"replace\" or not self.output_file.exists():\n             self.output_file.create_from_dataframe(df)\n+            return self.output_file\n         else:\n             raise FileExistsError(f\"{self.output_file.path} file already exists.\")\n \n@@ -66,7 +67,20 @@\n ) -> XComArg:\n     \"\"\"Convert SaveFile into a function. Returns XComArg.\n \n-    Returns an XComArg object.\n+    Returns an XComArg object of type File which matches the output_file parameter.\n+\n+    This will allow users to perform further actions with the exported file.\n+\n+    e.g.\n+\n+        with sample_dag:\n+        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)\n+        exported_file = aql.export_file(\n+            input_data=table,\n+            output_file=File(path=\"/tmp/saved_df.csv\"),\n+            if_exists=\"replace\",\n+        )\n+        res_df = aql.load_file(input_file=exported_file)\n \n     :param output_file: Path and conn_id\n     :param input_data: Input table / dataframe\n", "issue": "Change `export_file` to return `File` object\n**Context**\r\n\r\nIn order to allow users to perform subsequent actions on an exported file (while maintaining a functional structure), we should allow the `export_file` function to return a file object.\r\n\r\n* Astro-SDK version: 0.9..1\r\n* Request by: @jlaneve\r\n* Analysed by @dimberman \r\n\r\n**Problem**\r\n\r\nAt the moment a user who wants to use the `output_file` object would need to explicitly set dependencies like this:\r\n\r\n```\r\n    output_file = File(path=\"/tmp/saved_df.csv\")\r\n    with sample_dag:\r\n        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)\r\n        export = aql.export_file(\r\n            input_data=table,\r\n            output_file=output_file,\r\n            if_exists=\"replace\",\r\n        )\r\n        res_df = aql.load_file(input_file=output_file)\r\n        export >> res_df\r\n```\r\n\r\n**Desired behaviour**\r\n\r\n```\r\n    with sample_dag:\r\n        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)\r\n        exported_file = aql.export_file(\r\n            input_data=table,\r\n            output_file=File(path=\"/tmp/saved_df.csv\"),\r\n            if_exists=\"replace\",\r\n        )\r\n        res_df = aql.load_file(input_file=exported_file)\r\n```\r\n\r\n**Acceptance criteria**\r\n* Change `export_file` so it returns the `File` instance, as opposed to `None`\r\n\r\nSince there is no documentation about this task, we don't need to update the documentation for it. To create documentation for this feature should be part of another issue.\n", "before_files": [{"content": "from typing import Optional, Union\n\nimport pandas as pd\nfrom airflow.models import BaseOperator\nfrom airflow.models.xcom_arg import XComArg\n\nfrom astro.constants import ExportExistsStrategy\nfrom astro.databases import create_database\nfrom astro.files import File\nfrom astro.sql.table import Table\nfrom astro.utils.task_id_helper import get_task_id\n\n\nclass ExportFile(BaseOperator):\n    \"\"\"Write SQL table to csv/parquet on local/S3/GCS.\n\n    :param input_data: Table to convert to file\n    :param output_file: File object containing the path to the file and connection id.\n    :param if_exists: Overwrite file if exists. Default False.\n    \"\"\"\n\n    template_fields = (\"input_data\", \"output_file\")\n\n    def __init__(\n        self,\n        input_data: Union[Table, pd.DataFrame],\n        output_file: File,\n        if_exists: ExportExistsStrategy = \"exception\",\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        self.output_file = output_file\n        self.input_data = input_data\n        self.if_exists = if_exists\n        self.kwargs = kwargs\n\n    def execute(self, context: dict) -> None:\n        \"\"\"Write SQL table to csv/parquet on local/S3/GCS.\n\n        Infers SQL database type based on connection.\n        \"\"\"\n        # Infer db type from `input_conn_id`.\n        if isinstance(self.input_data, Table):\n            database = create_database(self.input_data.conn_id)\n            self.input_data = database.populate_table_metadata(self.input_data)\n            df = database.export_table_to_pandas_dataframe(self.input_data)\n        elif isinstance(self.input_data, pd.DataFrame):\n            df = self.input_data\n        else:\n            raise ValueError(\n                f\"Expected input_table to be Table or dataframe. Got {type(self.input_data)}\"\n            )\n        # Write file if overwrite == True or if file doesn't exist.\n        if self.if_exists == \"replace\" or not self.output_file.exists():\n            self.output_file.create_from_dataframe(df)\n        else:\n            raise FileExistsError(f\"{self.output_file.path} file already exists.\")\n\n\ndef export_file(\n    input_data: Union[Table, pd.DataFrame],\n    output_file: File,\n    if_exists: ExportExistsStrategy = \"exception\",\n    task_id: Optional[str] = None,\n    **kwargs,\n) -> XComArg:\n    \"\"\"Convert SaveFile into a function. Returns XComArg.\n\n    Returns an XComArg object.\n\n    :param output_file: Path and conn_id\n    :param input_data: Input table / dataframe\n    :param if_exists: Overwrite file if exists. Default \"exception\"\n    :param task_id: task id, optional\n    \"\"\"\n\n    task_id = (\n        task_id if task_id is not None else get_task_id(\"export_file\", output_file.path)\n    )\n\n    return ExportFile(\n        task_id=task_id,\n        output_file=output_file,\n        input_data=input_data,\n        if_exists=if_exists,\n    ).output\n", "path": "src/astro/sql/operators/export_file.py"}], "after_files": [{"content": "from typing import Optional, Union\n\nimport pandas as pd\nfrom airflow.models import BaseOperator\nfrom airflow.models.xcom_arg import XComArg\n\nfrom astro.constants import ExportExistsStrategy\nfrom astro.databases import create_database\nfrom astro.files import File\nfrom astro.sql.table import Table\nfrom astro.utils.task_id_helper import get_task_id\n\n\nclass ExportFile(BaseOperator):\n    \"\"\"Write SQL table to csv/parquet on local/S3/GCS.\n\n    :param input_data: Table to convert to file\n    :param output_file: File object containing the path to the file and connection id.\n    :param if_exists: Overwrite file if exists. Default False.\n    \"\"\"\n\n    template_fields = (\"input_data\", \"output_file\")\n\n    def __init__(\n        self,\n        input_data: Union[Table, pd.DataFrame],\n        output_file: File,\n        if_exists: ExportExistsStrategy = \"exception\",\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        self.output_file = output_file\n        self.input_data = input_data\n        self.if_exists = if_exists\n        self.kwargs = kwargs\n\n    def execute(self, context: dict) -> File:\n        \"\"\"Write SQL table to csv/parquet on local/S3/GCS.\n\n        Infers SQL database type based on connection.\n        \"\"\"\n        # Infer db type from `input_conn_id`.\n        if isinstance(self.input_data, Table):\n            database = create_database(self.input_data.conn_id)\n            self.input_data = database.populate_table_metadata(self.input_data)\n            df = database.export_table_to_pandas_dataframe(self.input_data)\n        elif isinstance(self.input_data, pd.DataFrame):\n            df = self.input_data\n        else:\n            raise ValueError(\n                f\"Expected input_table to be Table or dataframe. Got {type(self.input_data)}\"\n            )\n        # Write file if overwrite == True or if file doesn't exist.\n        if self.if_exists == \"replace\" or not self.output_file.exists():\n            self.output_file.create_from_dataframe(df)\n            return self.output_file\n        else:\n            raise FileExistsError(f\"{self.output_file.path} file already exists.\")\n\n\ndef export_file(\n    input_data: Union[Table, pd.DataFrame],\n    output_file: File,\n    if_exists: ExportExistsStrategy = \"exception\",\n    task_id: Optional[str] = None,\n    **kwargs,\n) -> XComArg:\n    \"\"\"Convert SaveFile into a function. Returns XComArg.\n\n    Returns an XComArg object of type File which matches the output_file parameter.\n\n    This will allow users to perform further actions with the exported file.\n\n    e.g.\n\n        with sample_dag:\n        table = aql.load_file(input_file=File(path=data_path), output_table=test_table)\n        exported_file = aql.export_file(\n            input_data=table,\n            output_file=File(path=\"/tmp/saved_df.csv\"),\n            if_exists=\"replace\",\n        )\n        res_df = aql.load_file(input_file=exported_file)\n\n    :param output_file: Path and conn_id\n    :param input_data: Input table / dataframe\n    :param if_exists: Overwrite file if exists. Default \"exception\"\n    :param task_id: task id, optional\n    \"\"\"\n\n    task_id = (\n        task_id if task_id is not None else get_task_id(\"export_file\", output_file.path)\n    )\n\n    return ExportFile(\n        task_id=task_id,\n        output_file=output_file,\n        input_data=input_data,\n        if_exists=if_exists,\n    ).output\n", "path": "src/astro/sql/operators/export_file.py"}]}