We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
MongoToS3Operator failed when running with a single query (not aggregate pipeline)
**Apache Airflow version**: 2.0.2

**What happened**:

`MongoToS3Operator` failed when running with a single query (not aggregate pipeline):

```sh
Traceback (most recent call last):
  File "/home/airflow//bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow//lib/python3.8/site-packages/airflow/__main__.py", line 40, in main
    args.func(args)
  File "/home/airflow//lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/home/airflow//lib/python3.8/site-packages/airflow/utils/cli.py", line 89, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow//lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 385, in task_test
    ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True)
  File "/home/airflow//lib/python3.8/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1413, in run
    self._run_raw_task(
  File "/home/airflow//lib/python3.8/site-packages/airflow/utils/session.py", line 67, in wrapper
    return func(*args, **kwargs)
  File "/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1138, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1311, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1341, in _execute_task
    result = task_copy.execute(context=context)
  File "/home/airflow//lib/python3.8/site-packages/airflow/providers/amazon/aws/transfers/mongo_to_s3.py", line 116, in execute
    results = MongoHook(self.mongo_conn_id).find(
  File "/home/airflow//lib/python3.8/site-packages/airflow/providers/mongo/hooks/mongo.py", line 144, in find
    return collection.find(query, **kwargs)
  File "/home/airflow//lib/python3.8/site-packages/pymongo/collection.py", line 1523, in find
    return Cursor(self, *args, **kwargs)
TypeError: __init__() got an unexpected keyword argument 'allowDiskUse'
```

**What you expected to happen**:

I expect the data from MongoDB to be exported to a file in S3 with no errors.

**How to reproduce it**:

Run the following operator with a single  `mongo_query` (no aggregate pipeline):

```python
export_to_s3 = MongoToS3Operator(
    task_id='export_to_s3',
    mongo_conn_id=Variable.get('mongo_conn_id'),
    s3_conn_id=Variable.get('aws_conn_id'),
    mongo_collection='my_mongo_collection',
    mongo_query={},
    s3_bucket=Variable.get('s3_bucket'),
    s3_key="my_data.json",
    replace=True,
    dag=dag,
)
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/providers/amazon/aws/transfers/mongo_to_s3.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 import json
19 import warnings
20 from typing import Any, Iterable, Optional, Union, cast
21 
22 from bson import json_util
23 
24 from airflow.models import BaseOperator
25 from airflow.providers.amazon.aws.hooks.s3 import S3Hook
26 from airflow.providers.mongo.hooks.mongo import MongoHook
27 
28 _DEPRECATION_MSG = (
29     "The s3_conn_id parameter has been deprecated. You should pass instead the aws_conn_id parameter."
30 )
31 
32 
33 class MongoToS3Operator(BaseOperator):
34     """Operator meant to move data from mongo via pymongo to s3 via boto.
35 
36     :param mongo_conn_id: reference to a specific mongo connection
37     :type mongo_conn_id: str
38     :param aws_conn_id: reference to a specific S3 connection
39     :type aws_conn_id: str
40     :param mongo_collection: reference to a specific collection in your mongo db
41     :type mongo_collection: str
42     :param mongo_query: query to execute. A list including a dict of the query
43     :type mongo_query: list
44     :param s3_bucket: reference to a specific S3 bucket to store the data
45     :type s3_bucket: str
46     :param s3_key: in which S3 key the file will be stored
47     :type s3_key: str
48     :param mongo_db: reference to a specific mongo database
49     :type mongo_db: str
50     :param replace: whether or not to replace the file in S3 if it previously existed
51     :type replace: bool
52     :param allow_disk_use: in the case you are retrieving a lot of data, you may have
53         to use the disk to save it instead of saving all in the RAM
54     :type allow_disk_use: bool
55     :param compression: type of compression to use for output file in S3. Currently only gzip is supported.
56     :type compression: str
57     """
58 
59     template_fields = ('s3_bucket', 's3_key', 'mongo_query', 'mongo_collection')
60     ui_color = '#589636'
61     template_fields_renderers = {"mongo_query": "py"}
62     # pylint: disable=too-many-instance-attributes
63 
64     def __init__(
65         self,
66         *,
67         s3_conn_id: Optional[str] = None,
68         mongo_conn_id: str = 'mongo_default',
69         aws_conn_id: str = 'aws_default',
70         mongo_collection: str,
71         mongo_query: Union[list, dict],
72         s3_bucket: str,
73         s3_key: str,
74         mongo_db: Optional[str] = None,
75         replace: bool = False,
76         allow_disk_use: bool = False,
77         compression: Optional[str] = None,
78         **kwargs,
79     ) -> None:
80         super().__init__(**kwargs)
81         if s3_conn_id:
82             warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=3)
83             aws_conn_id = s3_conn_id
84 
85         self.mongo_conn_id = mongo_conn_id
86         self.aws_conn_id = aws_conn_id
87         self.mongo_db = mongo_db
88         self.mongo_collection = mongo_collection
89 
90         # Grab query and determine if we need to run an aggregate pipeline
91         self.mongo_query = mongo_query
92         self.is_pipeline = isinstance(self.mongo_query, list)
93 
94         self.s3_bucket = s3_bucket
95         self.s3_key = s3_key
96         self.replace = replace
97         self.allow_disk_use = allow_disk_use
98         self.compression = compression
99 
100     def execute(self, context) -> bool:
101         """Is written to depend on transform method"""
102         s3_conn = S3Hook(self.aws_conn_id)
103 
104         # Grab collection and execute query according to whether or not it is a pipeline
105         if self.is_pipeline:
106             results = MongoHook(self.mongo_conn_id).aggregate(
107                 mongo_collection=self.mongo_collection,
108                 aggregate_query=cast(list, self.mongo_query),
109                 mongo_db=self.mongo_db,
110                 allowDiskUse=self.allow_disk_use,
111             )
112 
113         else:
114             results = MongoHook(self.mongo_conn_id).find(
115                 mongo_collection=self.mongo_collection,
116                 query=cast(dict, self.mongo_query),
117                 mongo_db=self.mongo_db,
118                 allowDiskUse=self.allow_disk_use,
119             )
120 
121         # Performs transform then stringifies the docs results into json format
122         docs_str = self._stringify(self.transform(results))
123 
124         s3_conn.load_string(
125             string_data=docs_str,
126             key=self.s3_key,
127             bucket_name=self.s3_bucket,
128             replace=self.replace,
129             compression=self.compression,
130         )
131 
132     @staticmethod
133     def _stringify(iterable: Iterable, joinable: str = '\n') -> str:
134         """
135         Takes an iterable (pymongo Cursor or Array) containing dictionaries and
136         returns a stringified version using python join
137         """
138         return joinable.join([json.dumps(doc, default=json_util.default) for doc in iterable])
139 
140     @staticmethod
141     def transform(docs: Any) -> Any:
142         """This method is meant to be extended by child classes
143         to perform transformations unique to those operators needs.
144         Processes pyMongo cursor and returns an iterable with each element being
145         a JSON serializable dictionary
146 
147         Base transform() assumes no processing is needed
148         ie. docs is a pyMongo cursor of documents and cursor just
149         needs to be passed through
150 
151         Override this method for custom transformations
152         """
153         return docs
154
```
--- 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/airflow/providers/amazon/aws/transfers/mongo_to_s3.py b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
--- a/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
+++ b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py
@@ -40,7 +40,7 @@
     :param mongo_collection: reference to a specific collection in your mongo db
     :type mongo_collection: str
     :param mongo_query: query to execute. A list including a dict of the query
-    :type mongo_query: list
+    :type mongo_query: Union[list, dict]
     :param s3_bucket: reference to a specific S3 bucket to store the data
     :type s3_bucket: str
     :param s3_key: in which S3 key the file will be stored
@@ -49,8 +49,8 @@
     :type mongo_db: str
     :param replace: whether or not to replace the file in S3 if it previously existed
     :type replace: bool
-    :param allow_disk_use: in the case you are retrieving a lot of data, you may have
-        to use the disk to save it instead of saving all in the RAM
+    :param allow_disk_use: enables writing to temporary files in the case you are handling large dataset.
+        This only takes effect when `mongo_query` is a list - running an aggregate pipeline
     :type allow_disk_use: bool
     :param compression: type of compression to use for output file in S3. Currently only gzip is supported.
     :type compression: str
@@ -115,7 +115,6 @@
                 mongo_collection=self.mongo_collection,
                 query=cast(dict, self.mongo_query),
                 mongo_db=self.mongo_db,
-                allowDiskUse=self.allow_disk_use,
             )
 
         # Performs transform then stringifies the docs results into json format

{"golden_diff": "diff --git a/airflow/providers/amazon/aws/transfers/mongo_to_s3.py b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py\n--- a/airflow/providers/amazon/aws/transfers/mongo_to_s3.py\n+++ b/airflow/providers/amazon/aws/transfers/mongo_to_s3.py\n@@ -40,7 +40,7 @@\n     :param mongo_collection: reference to a specific collection in your mongo db\n     :type mongo_collection: str\n     :param mongo_query: query to execute. A list including a dict of the query\n-    :type mongo_query: list\n+    :type mongo_query: Union[list, dict]\n     :param s3_bucket: reference to a specific S3 bucket to store the data\n     :type s3_bucket: str\n     :param s3_key: in which S3 key the file will be stored\n@@ -49,8 +49,8 @@\n     :type mongo_db: str\n     :param replace: whether or not to replace the file in S3 if it previously existed\n     :type replace: bool\n-    :param allow_disk_use: in the case you are retrieving a lot of data, you may have\n-        to use the disk to save it instead of saving all in the RAM\n+    :param allow_disk_use: enables writing to temporary files in the case you are handling large dataset.\n+        This only takes effect when `mongo_query` is a list - running an aggregate pipeline\n     :type allow_disk_use: bool\n     :param compression: type of compression to use for output file in S3. Currently only gzip is supported.\n     :type compression: str\n@@ -115,7 +115,6 @@\n                 mongo_collection=self.mongo_collection,\n                 query=cast(dict, self.mongo_query),\n                 mongo_db=self.mongo_db,\n-                allowDiskUse=self.allow_disk_use,\n             )\n \n         # Performs transform then stringifies the docs results into json format\n", "issue": "MongoToS3Operator failed when running with a single query (not aggregate pipeline)\n**Apache Airflow version**: 2.0.2\r\n\r\n**What happened**:\r\n\r\n`MongoToS3Operator` failed when running with a single query (not aggregate pipeline):\r\n\r\n```sh\r\nTraceback (most recent call last):\r\n  File \"/home/airflow//bin/airflow\", line 8, in <module>\r\n    sys.exit(main())\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/__main__.py\", line 40, in main\r\n    args.func(args)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/cli/cli_parser.py\", line 48, in command\r\n    return func(*args, **kwargs)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/utils/cli.py\", line 89, in wrapper\r\n    return f(*args, **kwargs)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/cli/commands/task_command.py\", line 385, in task_test\r\n    ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/utils/session.py\", line 70, in wrapper\r\n    return func(*args, session=session, **kwargs)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1413, in run\r\n    self._run_raw_task(\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/utils/session.py\", line 67, in wrapper\r\n    return func(*args, **kwargs)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1138, in _run_raw_task\r\n    self._prepare_and_execute_task_with_callbacks(context, task)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1311, in _prepare_and_execute_task_with_callbacks\r\n    result = self._execute_task(context, task_copy)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1341, in _execute_task\r\n    result = task_copy.execute(context=context)\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/providers/amazon/aws/transfers/mongo_to_s3.py\", line 116, in execute\r\n    results = MongoHook(self.mongo_conn_id).find(\r\n  File \"/home/airflow//lib/python3.8/site-packages/airflow/providers/mongo/hooks/mongo.py\", line 144, in find\r\n    return collection.find(query, **kwargs)\r\n  File \"/home/airflow//lib/python3.8/site-packages/pymongo/collection.py\", line 1523, in find\r\n    return Cursor(self, *args, **kwargs)\r\nTypeError: __init__() got an unexpected keyword argument 'allowDiskUse'\r\n```\r\n\r\n**What you expected to happen**:\r\n\r\nI expect the data from MongoDB to be exported to a file in S3 with no errors.\r\n\r\n**How to reproduce it**:\r\n\r\nRun the following operator with a single  `mongo_query` (no aggregate pipeline):\r\n\r\n```python\r\nexport_to_s3 = MongoToS3Operator(\r\n    task_id='export_to_s3',\r\n    mongo_conn_id=Variable.get('mongo_conn_id'),\r\n    s3_conn_id=Variable.get('aws_conn_id'),\r\n    mongo_collection='my_mongo_collection',\r\n    mongo_query={},\r\n    s3_bucket=Variable.get('s3_bucket'),\r\n    s3_key=\"my_data.json\",\r\n    replace=True,\r\n    dag=dag,\r\n)\r\n```\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nimport json\nimport warnings\nfrom typing import Any, Iterable, Optional, Union, cast\n\nfrom bson import json_util\n\nfrom airflow.models import BaseOperator\nfrom airflow.providers.amazon.aws.hooks.s3 import S3Hook\nfrom airflow.providers.mongo.hooks.mongo import MongoHook\n\n_DEPRECATION_MSG = (\n    \"The s3_conn_id parameter has been deprecated. You should pass instead the aws_conn_id parameter.\"\n)\n\n\nclass MongoToS3Operator(BaseOperator):\n    \"\"\"Operator meant to move data from mongo via pymongo to s3 via boto.\n\n    :param mongo_conn_id: reference to a specific mongo connection\n    :type mongo_conn_id: str\n    :param aws_conn_id: reference to a specific S3 connection\n    :type aws_conn_id: str\n    :param mongo_collection: reference to a specific collection in your mongo db\n    :type mongo_collection: str\n    :param mongo_query: query to execute. A list including a dict of the query\n    :type mongo_query: list\n    :param s3_bucket: reference to a specific S3 bucket to store the data\n    :type s3_bucket: str\n    :param s3_key: in which S3 key the file will be stored\n    :type s3_key: str\n    :param mongo_db: reference to a specific mongo database\n    :type mongo_db: str\n    :param replace: whether or not to replace the file in S3 if it previously existed\n    :type replace: bool\n    :param allow_disk_use: in the case you are retrieving a lot of data, you may have\n        to use the disk to save it instead of saving all in the RAM\n    :type allow_disk_use: bool\n    :param compression: type of compression to use for output file in S3. Currently only gzip is supported.\n    :type compression: str\n    \"\"\"\n\n    template_fields = ('s3_bucket', 's3_key', 'mongo_query', 'mongo_collection')\n    ui_color = '#589636'\n    template_fields_renderers = {\"mongo_query\": \"py\"}\n    # pylint: disable=too-many-instance-attributes\n\n    def __init__(\n        self,\n        *,\n        s3_conn_id: Optional[str] = None,\n        mongo_conn_id: str = 'mongo_default',\n        aws_conn_id: str = 'aws_default',\n        mongo_collection: str,\n        mongo_query: Union[list, dict],\n        s3_bucket: str,\n        s3_key: str,\n        mongo_db: Optional[str] = None,\n        replace: bool = False,\n        allow_disk_use: bool = False,\n        compression: Optional[str] = None,\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        if s3_conn_id:\n            warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=3)\n            aws_conn_id = s3_conn_id\n\n        self.mongo_conn_id = mongo_conn_id\n        self.aws_conn_id = aws_conn_id\n        self.mongo_db = mongo_db\n        self.mongo_collection = mongo_collection\n\n        # Grab query and determine if we need to run an aggregate pipeline\n        self.mongo_query = mongo_query\n        self.is_pipeline = isinstance(self.mongo_query, list)\n\n        self.s3_bucket = s3_bucket\n        self.s3_key = s3_key\n        self.replace = replace\n        self.allow_disk_use = allow_disk_use\n        self.compression = compression\n\n    def execute(self, context) -> bool:\n        \"\"\"Is written to depend on transform method\"\"\"\n        s3_conn = S3Hook(self.aws_conn_id)\n\n        # Grab collection and execute query according to whether or not it is a pipeline\n        if self.is_pipeline:\n            results = MongoHook(self.mongo_conn_id).aggregate(\n                mongo_collection=self.mongo_collection,\n                aggregate_query=cast(list, self.mongo_query),\n                mongo_db=self.mongo_db,\n                allowDiskUse=self.allow_disk_use,\n            )\n\n        else:\n            results = MongoHook(self.mongo_conn_id).find(\n                mongo_collection=self.mongo_collection,\n                query=cast(dict, self.mongo_query),\n                mongo_db=self.mongo_db,\n                allowDiskUse=self.allow_disk_use,\n            )\n\n        # Performs transform then stringifies the docs results into json format\n        docs_str = self._stringify(self.transform(results))\n\n        s3_conn.load_string(\n            string_data=docs_str,\n            key=self.s3_key,\n            bucket_name=self.s3_bucket,\n            replace=self.replace,\n            compression=self.compression,\n        )\n\n    @staticmethod\n    def _stringify(iterable: Iterable, joinable: str = '\\n') -> str:\n        \"\"\"\n        Takes an iterable (pymongo Cursor or Array) containing dictionaries and\n        returns a stringified version using python join\n        \"\"\"\n        return joinable.join([json.dumps(doc, default=json_util.default) for doc in iterable])\n\n    @staticmethod\n    def transform(docs: Any) -> Any:\n        \"\"\"This method is meant to be extended by child classes\n        to perform transformations unique to those operators needs.\n        Processes pyMongo cursor and returns an iterable with each element being\n        a JSON serializable dictionary\n\n        Base transform() assumes no processing is needed\n        ie. docs is a pyMongo cursor of documents and cursor just\n        needs to be passed through\n\n        Override this method for custom transformations\n        \"\"\"\n        return docs\n", "path": "airflow/providers/amazon/aws/transfers/mongo_to_s3.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  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,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\nimport json\nimport warnings\nfrom typing import Any, Iterable, Optional, Union, cast\n\nfrom bson import json_util\n\nfrom airflow.models import BaseOperator\nfrom airflow.providers.amazon.aws.hooks.s3 import S3Hook\nfrom airflow.providers.mongo.hooks.mongo import MongoHook\n\n_DEPRECATION_MSG = (\n    \"The s3_conn_id parameter has been deprecated. You should pass instead the aws_conn_id parameter.\"\n)\n\n\nclass MongoToS3Operator(BaseOperator):\n    \"\"\"Operator meant to move data from mongo via pymongo to s3 via boto.\n\n    :param mongo_conn_id: reference to a specific mongo connection\n    :type mongo_conn_id: str\n    :param aws_conn_id: reference to a specific S3 connection\n    :type aws_conn_id: str\n    :param mongo_collection: reference to a specific collection in your mongo db\n    :type mongo_collection: str\n    :param mongo_query: query to execute. A list including a dict of the query\n    :type mongo_query: Union[list, dict]\n    :param s3_bucket: reference to a specific S3 bucket to store the data\n    :type s3_bucket: str\n    :param s3_key: in which S3 key the file will be stored\n    :type s3_key: str\n    :param mongo_db: reference to a specific mongo database\n    :type mongo_db: str\n    :param replace: whether or not to replace the file in S3 if it previously existed\n    :type replace: bool\n    :param allow_disk_use: enables writing to temporary files in the case you are handling large dataset.\n        This only takes effect when `mongo_query` is a list - running an aggregate pipeline\n    :type allow_disk_use: bool\n    :param compression: type of compression to use for output file in S3. Currently only gzip is supported.\n    :type compression: str\n    \"\"\"\n\n    template_fields = ('s3_bucket', 's3_key', 'mongo_query', 'mongo_collection')\n    ui_color = '#589636'\n    template_fields_renderers = {\"mongo_query\": \"py\"}\n    # pylint: disable=too-many-instance-attributes\n\n    def __init__(\n        self,\n        *,\n        s3_conn_id: Optional[str] = None,\n        mongo_conn_id: str = 'mongo_default',\n        aws_conn_id: str = 'aws_default',\n        mongo_collection: str,\n        mongo_query: Union[list, dict],\n        s3_bucket: str,\n        s3_key: str,\n        mongo_db: Optional[str] = None,\n        replace: bool = False,\n        allow_disk_use: bool = False,\n        compression: Optional[str] = None,\n        **kwargs,\n    ) -> None:\n        super().__init__(**kwargs)\n        if s3_conn_id:\n            warnings.warn(_DEPRECATION_MSG, DeprecationWarning, stacklevel=3)\n            aws_conn_id = s3_conn_id\n\n        self.mongo_conn_id = mongo_conn_id\n        self.aws_conn_id = aws_conn_id\n        self.mongo_db = mongo_db\n        self.mongo_collection = mongo_collection\n\n        # Grab query and determine if we need to run an aggregate pipeline\n        self.mongo_query = mongo_query\n        self.is_pipeline = isinstance(self.mongo_query, list)\n\n        self.s3_bucket = s3_bucket\n        self.s3_key = s3_key\n        self.replace = replace\n        self.allow_disk_use = allow_disk_use\n        self.compression = compression\n\n    def execute(self, context) -> bool:\n        \"\"\"Is written to depend on transform method\"\"\"\n        s3_conn = S3Hook(self.aws_conn_id)\n\n        # Grab collection and execute query according to whether or not it is a pipeline\n        if self.is_pipeline:\n            results = MongoHook(self.mongo_conn_id).aggregate(\n                mongo_collection=self.mongo_collection,\n                aggregate_query=cast(list, self.mongo_query),\n                mongo_db=self.mongo_db,\n                allowDiskUse=self.allow_disk_use,\n            )\n\n        else:\n            results = MongoHook(self.mongo_conn_id).find(\n                mongo_collection=self.mongo_collection,\n                query=cast(dict, self.mongo_query),\n                mongo_db=self.mongo_db,\n            )\n\n        # Performs transform then stringifies the docs results into json format\n        docs_str = self._stringify(self.transform(results))\n\n        s3_conn.load_string(\n            string_data=docs_str,\n            key=self.s3_key,\n            bucket_name=self.s3_bucket,\n            replace=self.replace,\n            compression=self.compression,\n        )\n\n    @staticmethod\n    def _stringify(iterable: Iterable, joinable: str = '\\n') -> str:\n        \"\"\"\n        Takes an iterable (pymongo Cursor or Array) containing dictionaries and\n        returns a stringified version using python join\n        \"\"\"\n        return joinable.join([json.dumps(doc, default=json_util.default) for doc in iterable])\n\n    @staticmethod\n    def transform(docs: Any) -> Any:\n        \"\"\"This method is meant to be extended by child classes\n        to perform transformations unique to those operators needs.\n        Processes pyMongo cursor and returns an iterable with each element being\n        a JSON serializable dictionary\n\n        Base transform() assumes no processing is needed\n        ie. docs is a pyMongo cursor of documents and cursor just\n        needs to be passed through\n\n        Override this method for custom transformations\n        \"\"\"\n        return docs\n", "path": "airflow/providers/amazon/aws/transfers/mongo_to_s3.py"}]}