We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Can`t upload big files with waitress
* Mindsdb version you tried to install: latest
* Additional info if applicable: waitress==2.0.0

Waitress consumes too much memory when uploading a file. 16Gb of memory is not enough to upload 1.1Gb file 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mindsdb/api/http/namespaces/datasource.py`
Content:
```
1 import os
2 import threading
3 import tempfile
4 import re
5 import multipart
6 
7 from dateutil.parser import parse
8 from flask import request, send_file
9 from flask_restx import Resource, abort     # 'abort' using to return errors as json: {'message': 'error text'}
10 from flask import current_app as ca
11 
12 from mindsdb.utilities.log import log
13 from mindsdb.api.http.namespaces.configs.datasources import ns_conf
14 from mindsdb.api.http.namespaces.entitites.datasources.datasource import (
15     datasource_metadata,
16     put_datasource_params
17 )
18 from mindsdb.api.http.namespaces.entitites.datasources.datasource_data import (
19     get_datasource_rows_params,
20     datasource_rows_metadata
21 )
22 from mindsdb.api.http.namespaces.entitites.datasources.datasource_files import (
23     put_datasource_file_params
24 )
25 from mindsdb.api.http.namespaces.entitites.datasources.datasource_missed_files import (
26     datasource_missed_files_metadata,
27     get_datasource_missed_files_params
28 )
29 from mindsdb.interfaces.database.integrations import get_db_integration
30 
31 
32 def parse_filter(key, value):
33     result = re.search(r'filter(_*.*)\[(.*)\]', key)
34     operator = result.groups()[0].strip('_') or 'like'
35     field = result.groups()[1]
36     operators_map = {
37         'like': 'like',
38         'in': 'in',
39         'nin': 'not in',
40         'gt': '>',
41         'lt': '<',
42         'gte': '>=',
43         'lte': '<=',
44         'eq': '=',
45         'neq': '!='
46     }
47     if operator not in operators_map:
48         return None
49     operator = operators_map[operator]
50     return [field, operator, value]
51 
52 
53 @ns_conf.route('/')
54 class DatasourcesList(Resource):
55     @ns_conf.doc('get_datasources_list')
56     @ns_conf.marshal_list_with(datasource_metadata)
57     def get(self):
58         '''List all datasources'''
59         return request.default_store.get_datasources()
60 
61 
62 @ns_conf.route('/<name>')
63 @ns_conf.param('name', 'Datasource name')
64 class Datasource(Resource):
65     @ns_conf.doc('get_datasource')
66     @ns_conf.marshal_with(datasource_metadata)
67     def get(self, name):
68         '''return datasource metadata'''
69         ds = request.default_store.get_datasource(name)
70         if ds is not None:
71             return ds
72         return '', 404
73 
74     @ns_conf.doc('delete_datasource')
75     def delete(self, name):
76         '''delete datasource'''
77 
78         try:
79             request.default_store.delete_datasource(name)
80         except Exception as e:
81             log.error(e)
82             abort(400, str(e))
83         return '', 200
84 
85     @ns_conf.doc('put_datasource', params=put_datasource_params)
86     @ns_conf.marshal_with(datasource_metadata)
87     def put(self, name):
88         '''add new datasource'''
89         data = {}
90 
91         def on_field(field):
92             name = field.field_name.decode()
93             value = field.value.decode()
94             data[name] = value
95 
96         file_object = None
97 
98         def on_file(file):
99             nonlocal file_object
100             data['file'] = file.file_name.decode()
101             file_object = file.file_object
102 
103         temp_dir_path = tempfile.mkdtemp(prefix='datasource_file_')
104 
105         if request.headers['Content-Type'].startswith('multipart/form-data'):
106             parser = multipart.create_form_parser(
107                 headers=request.headers,
108                 on_field=on_field,
109                 on_file=on_file,
110                 config={
111                     'UPLOAD_DIR': temp_dir_path.encode(),    # bytes required
112                     'UPLOAD_KEEP_FILENAME': True,
113                     'UPLOAD_KEEP_EXTENSIONS': True,
114                     'MAX_MEMORY_FILE_SIZE': 0
115                 }
116             )
117 
118             while True:
119                 chunk = request.stream.read(8192)
120                 if not chunk:
121                     break
122                 parser.write(chunk)
123             parser.finalize()
124             parser.close()
125 
126             if file_object is not None and not file_object.closed:
127                 file_object.close()
128         else:
129             data = request.json
130 
131         if 'query' in data:
132             integration_id = request.json['integration_id']
133             integration = get_db_integration(integration_id, request.company_id)
134             if integration is None:
135                 abort(400, f"{integration_id} integration doesn't exist")
136 
137             if integration['type'] == 'mongodb':
138                 data['find'] = data['query']
139 
140             request.default_store.save_datasource(name, integration_id, data)
141             os.rmdir(temp_dir_path)
142             return request.default_store.get_datasource(name)
143 
144         ds_name = data['name'] if 'name' in data else name
145         source = data['source'] if 'source' in data else name
146         source_type = data['source_type']
147 
148         if source_type == 'file':
149             file_path = os.path.join(temp_dir_path, data['file'])
150         else:
151             file_path = None
152 
153         request.default_store.save_datasource(ds_name, source_type, source, file_path)
154         os.rmdir(temp_dir_path)
155 
156         return request.default_store.get_datasource(ds_name)
157 
158 
159 def analyzing_thread(name, default_store):
160     try:
161         from mindsdb.interfaces.storage.db import session
162         analysis = default_store.start_analysis(name)
163         session.close()
164     except Exception as e:
165         log.error(e)
166 
167 
168 @ns_conf.route('/<name>/analyze')
169 @ns_conf.param('name', 'Datasource name')
170 class Analyze(Resource):
171     @ns_conf.doc('analyse_dataset')
172     def get(self, name):
173         analysis = request.default_store.get_analysis(name)
174         if analysis is not None:
175             return analysis, 200
176 
177 
178         ds = request.default_store.get_datasource(name)
179         if ds is None:
180             log.error('No valid datasource given')
181             abort(400, 'No valid datasource given')
182 
183         x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))
184         x.start()
185         return {'status': 'analyzing'}, 200
186 
187 
188 @ns_conf.route('/<name>/analyze_refresh')
189 @ns_conf.param('name', 'Datasource name')
190 class Analyze2(Resource):
191     @ns_conf.doc('analyze_refresh_dataset')
192     def get(self, name):
193         analysis = request.default_store.get_analysis(name)
194         if analysis is not None:
195             return analysis, 200
196 
197         ds = request.default_store.get_datasource(name)
198         if ds is None:
199             log.error('No valid datasource given')
200             abort(400, 'No valid datasource given')
201 
202         x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))
203         x.start()
204         return {'status': 'analyzing'}, 200
205 
206 
207 @ns_conf.route('/<name>/data/')
208 @ns_conf.param('name', 'Datasource name')
209 class DatasourceData(Resource):
210     @ns_conf.doc('get_datasource_data', params=get_datasource_rows_params)
211     @ns_conf.marshal_with(datasource_rows_metadata)
212     def get(self, name):
213         '''return data rows'''
214         ds = request.default_store.get_datasource(name)
215         if ds is None:
216             abort(400, 'No valid datasource given')
217 
218         params = {
219             'page[size]': None,
220             'page[offset]': None
221         }
222         where = []
223         for key, value in request.args.items():
224             if key == 'page[size]':
225                 params['page[size]'] = int(value)
226             if key == 'page[offset]':
227                 params['page[offset]'] = int(value)
228             elif key.startswith('filter'):
229                 param = parse_filter(key, value)
230                 if param is None:
231                     abort(400, f'Not valid filter "{key}"')
232                 where.append(param)
233 
234         data_dict = request.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])
235         return data_dict, 200
236 
237 
238 @ns_conf.route('/<name>/download')
239 @ns_conf.param('name', 'Datasource name')
240 class DatasourceMissedFilesDownload(Resource):
241     @ns_conf.doc('get_datasource_download')
242     def get(self, name):
243         '''download uploaded file'''
244         ds = request.default_store.get_datasource(name)
245         if not ds:
246             abort(404, "{} not found".format(name))
247         # force download from s3
248         request.default_store.get_datasource_obj(name)
249         if not os.path.exists(ds['source']):
250             abort(404, "{} not found".format(name))
251 
252         return send_file(os.path.abspath(ds['source']), as_attachment=True)
253
```
--- 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/mindsdb/api/http/namespaces/datasource.py b/mindsdb/api/http/namespaces/datasource.py
--- a/mindsdb/api/http/namespaces/datasource.py
+++ b/mindsdb/api/http/namespaces/datasource.py
@@ -3,6 +3,8 @@
 import tempfile
 import re
 import multipart
+import zipfile
+import tarfile
 
 from dateutil.parser import parse
 from flask import request, send_file
@@ -10,6 +12,7 @@
 from flask import current_app as ca
 
 from mindsdb.utilities.log import log
+from mindsdb.api.http.utils import http_error
 from mindsdb.api.http.namespaces.configs.datasources import ns_conf
 from mindsdb.api.http.namespaces.entitites.datasources.datasource import (
     datasource_metadata,
@@ -147,6 +150,24 @@
 
         if source_type == 'file':
             file_path = os.path.join(temp_dir_path, data['file'])
+            lp = file_path.lower()
+            if lp.endswith(('.zip', '.tar.gz')):
+                if lp.endswith('.zip'):
+                    with zipfile.ZipFile(file_path) as f:
+                        f.extractall(temp_dir_path)
+                elif lp.endswith('.tar.gz'):
+                    with tarfile.open(file_path) as f:
+                        f.extractall(temp_dir_path)
+                os.remove(file_path)
+                files = os.listdir(temp_dir_path)
+                if len(files) != 1:
+                    os.rmdir(temp_dir_path)
+                    return http_error(400, 'Wrong content.', 'Archive must contain only one data file.')
+                file_path = os.path.join(temp_dir_path, files[0])
+                source = files[0]
+                if not os.path.isfile(file_path):
+                    os.rmdir(temp_dir_path)
+                    return http_error(400, 'Wrong content.', 'Archive must contain data file in root.')
         else:
             file_path = None
 
@@ -174,7 +195,6 @@
         if analysis is not None:
             return analysis, 200
 
-
         ds = request.default_store.get_datasource(name)
         if ds is None:
             log.error('No valid datasource given')

{"golden_diff": "diff --git a/mindsdb/api/http/namespaces/datasource.py b/mindsdb/api/http/namespaces/datasource.py\n--- a/mindsdb/api/http/namespaces/datasource.py\n+++ b/mindsdb/api/http/namespaces/datasource.py\n@@ -3,6 +3,8 @@\n import tempfile\n import re\n import multipart\n+import zipfile\n+import tarfile\n \n from dateutil.parser import parse\n from flask import request, send_file\n@@ -10,6 +12,7 @@\n from flask import current_app as ca\n \n from mindsdb.utilities.log import log\n+from mindsdb.api.http.utils import http_error\n from mindsdb.api.http.namespaces.configs.datasources import ns_conf\n from mindsdb.api.http.namespaces.entitites.datasources.datasource import (\n     datasource_metadata,\n@@ -147,6 +150,24 @@\n \n         if source_type == 'file':\n             file_path = os.path.join(temp_dir_path, data['file'])\n+            lp = file_path.lower()\n+            if lp.endswith(('.zip', '.tar.gz')):\n+                if lp.endswith('.zip'):\n+                    with zipfile.ZipFile(file_path) as f:\n+                        f.extractall(temp_dir_path)\n+                elif lp.endswith('.tar.gz'):\n+                    with tarfile.open(file_path) as f:\n+                        f.extractall(temp_dir_path)\n+                os.remove(file_path)\n+                files = os.listdir(temp_dir_path)\n+                if len(files) != 1:\n+                    os.rmdir(temp_dir_path)\n+                    return http_error(400, 'Wrong content.', 'Archive must contain only one data file.')\n+                file_path = os.path.join(temp_dir_path, files[0])\n+                source = files[0]\n+                if not os.path.isfile(file_path):\n+                    os.rmdir(temp_dir_path)\n+                    return http_error(400, 'Wrong content.', 'Archive must contain data file in root.')\n         else:\n             file_path = None\n \n@@ -174,7 +195,6 @@\n         if analysis is not None:\n             return analysis, 200\n \n-\n         ds = request.default_store.get_datasource(name)\n         if ds is None:\n             log.error('No valid datasource given')\n", "issue": "Can`t upload big files with waitress\n* Mindsdb version you tried to install: latest\r\n* Additional info if applicable: waitress==2.0.0\r\n\r\nWaitress consumes too much memory when uploading a file. 16Gb of memory is not enough to upload 1.1Gb file \n", "before_files": [{"content": "import os\nimport threading\nimport tempfile\nimport re\nimport multipart\n\nfrom dateutil.parser import parse\nfrom flask import request, send_file\nfrom flask_restx import Resource, abort     # 'abort' using to return errors as json: {'message': 'error text'}\nfrom flask import current_app as ca\n\nfrom mindsdb.utilities.log import log\nfrom mindsdb.api.http.namespaces.configs.datasources import ns_conf\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource import (\n    datasource_metadata,\n    put_datasource_params\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_data import (\n    get_datasource_rows_params,\n    datasource_rows_metadata\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_files import (\n    put_datasource_file_params\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_missed_files import (\n    datasource_missed_files_metadata,\n    get_datasource_missed_files_params\n)\nfrom mindsdb.interfaces.database.integrations import get_db_integration\n\n\ndef parse_filter(key, value):\n    result = re.search(r'filter(_*.*)\\[(.*)\\]', key)\n    operator = result.groups()[0].strip('_') or 'like'\n    field = result.groups()[1]\n    operators_map = {\n        'like': 'like',\n        'in': 'in',\n        'nin': 'not in',\n        'gt': '>',\n        'lt': '<',\n        'gte': '>=',\n        'lte': '<=',\n        'eq': '=',\n        'neq': '!='\n    }\n    if operator not in operators_map:\n        return None\n    operator = operators_map[operator]\n    return [field, operator, value]\n\n\n@ns_conf.route('/')\nclass DatasourcesList(Resource):\n    @ns_conf.doc('get_datasources_list')\n    @ns_conf.marshal_list_with(datasource_metadata)\n    def get(self):\n        '''List all datasources'''\n        return request.default_store.get_datasources()\n\n\n@ns_conf.route('/<name>')\n@ns_conf.param('name', 'Datasource name')\nclass Datasource(Resource):\n    @ns_conf.doc('get_datasource')\n    @ns_conf.marshal_with(datasource_metadata)\n    def get(self, name):\n        '''return datasource metadata'''\n        ds = request.default_store.get_datasource(name)\n        if ds is not None:\n            return ds\n        return '', 404\n\n    @ns_conf.doc('delete_datasource')\n    def delete(self, name):\n        '''delete datasource'''\n\n        try:\n            request.default_store.delete_datasource(name)\n        except Exception as e:\n            log.error(e)\n            abort(400, str(e))\n        return '', 200\n\n    @ns_conf.doc('put_datasource', params=put_datasource_params)\n    @ns_conf.marshal_with(datasource_metadata)\n    def put(self, name):\n        '''add new datasource'''\n        data = {}\n\n        def on_field(field):\n            name = field.field_name.decode()\n            value = field.value.decode()\n            data[name] = value\n\n        file_object = None\n\n        def on_file(file):\n            nonlocal file_object\n            data['file'] = file.file_name.decode()\n            file_object = file.file_object\n\n        temp_dir_path = tempfile.mkdtemp(prefix='datasource_file_')\n\n        if request.headers['Content-Type'].startswith('multipart/form-data'):\n            parser = multipart.create_form_parser(\n                headers=request.headers,\n                on_field=on_field,\n                on_file=on_file,\n                config={\n                    'UPLOAD_DIR': temp_dir_path.encode(),    # bytes required\n                    'UPLOAD_KEEP_FILENAME': True,\n                    'UPLOAD_KEEP_EXTENSIONS': True,\n                    'MAX_MEMORY_FILE_SIZE': 0\n                }\n            )\n\n            while True:\n                chunk = request.stream.read(8192)\n                if not chunk:\n                    break\n                parser.write(chunk)\n            parser.finalize()\n            parser.close()\n\n            if file_object is not None and not file_object.closed:\n                file_object.close()\n        else:\n            data = request.json\n\n        if 'query' in data:\n            integration_id = request.json['integration_id']\n            integration = get_db_integration(integration_id, request.company_id)\n            if integration is None:\n                abort(400, f\"{integration_id} integration doesn't exist\")\n\n            if integration['type'] == 'mongodb':\n                data['find'] = data['query']\n\n            request.default_store.save_datasource(name, integration_id, data)\n            os.rmdir(temp_dir_path)\n            return request.default_store.get_datasource(name)\n\n        ds_name = data['name'] if 'name' in data else name\n        source = data['source'] if 'source' in data else name\n        source_type = data['source_type']\n\n        if source_type == 'file':\n            file_path = os.path.join(temp_dir_path, data['file'])\n        else:\n            file_path = None\n\n        request.default_store.save_datasource(ds_name, source_type, source, file_path)\n        os.rmdir(temp_dir_path)\n\n        return request.default_store.get_datasource(ds_name)\n\n\ndef analyzing_thread(name, default_store):\n    try:\n        from mindsdb.interfaces.storage.db import session\n        analysis = default_store.start_analysis(name)\n        session.close()\n    except Exception as e:\n        log.error(e)\n\n\n@ns_conf.route('/<name>/analyze')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze(Resource):\n    @ns_conf.doc('analyse_dataset')\n    def get(self, name):\n        analysis = request.default_store.get_analysis(name)\n        if analysis is not None:\n            return analysis, 200\n\n\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            log.error('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/analyze_refresh')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze2(Resource):\n    @ns_conf.doc('analyze_refresh_dataset')\n    def get(self, name):\n        analysis = request.default_store.get_analysis(name)\n        if analysis is not None:\n            return analysis, 200\n\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            log.error('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/data/')\n@ns_conf.param('name', 'Datasource name')\nclass DatasourceData(Resource):\n    @ns_conf.doc('get_datasource_data', params=get_datasource_rows_params)\n    @ns_conf.marshal_with(datasource_rows_metadata)\n    def get(self, name):\n        '''return data rows'''\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            abort(400, 'No valid datasource given')\n\n        params = {\n            'page[size]': None,\n            'page[offset]': None\n        }\n        where = []\n        for key, value in request.args.items():\n            if key == 'page[size]':\n                params['page[size]'] = int(value)\n            if key == 'page[offset]':\n                params['page[offset]'] = int(value)\n            elif key.startswith('filter'):\n                param = parse_filter(key, value)\n                if param is None:\n                    abort(400, f'Not valid filter \"{key}\"')\n                where.append(param)\n\n        data_dict = request.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])\n        return data_dict, 200\n\n\n@ns_conf.route('/<name>/download')\n@ns_conf.param('name', 'Datasource name')\nclass DatasourceMissedFilesDownload(Resource):\n    @ns_conf.doc('get_datasource_download')\n    def get(self, name):\n        '''download uploaded file'''\n        ds = request.default_store.get_datasource(name)\n        if not ds:\n            abort(404, \"{} not found\".format(name))\n        # force download from s3\n        request.default_store.get_datasource_obj(name)\n        if not os.path.exists(ds['source']):\n            abort(404, \"{} not found\".format(name))\n\n        return send_file(os.path.abspath(ds['source']), as_attachment=True)\n", "path": "mindsdb/api/http/namespaces/datasource.py"}], "after_files": [{"content": "import os\nimport threading\nimport tempfile\nimport re\nimport multipart\nimport zipfile\nimport tarfile\n\nfrom dateutil.parser import parse\nfrom flask import request, send_file\nfrom flask_restx import Resource, abort     # 'abort' using to return errors as json: {'message': 'error text'}\nfrom flask import current_app as ca\n\nfrom mindsdb.utilities.log import log\nfrom mindsdb.api.http.utils import http_error\nfrom mindsdb.api.http.namespaces.configs.datasources import ns_conf\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource import (\n    datasource_metadata,\n    put_datasource_params\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_data import (\n    get_datasource_rows_params,\n    datasource_rows_metadata\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_files import (\n    put_datasource_file_params\n)\nfrom mindsdb.api.http.namespaces.entitites.datasources.datasource_missed_files import (\n    datasource_missed_files_metadata,\n    get_datasource_missed_files_params\n)\nfrom mindsdb.interfaces.database.integrations import get_db_integration\n\n\ndef parse_filter(key, value):\n    result = re.search(r'filter(_*.*)\\[(.*)\\]', key)\n    operator = result.groups()[0].strip('_') or 'like'\n    field = result.groups()[1]\n    operators_map = {\n        'like': 'like',\n        'in': 'in',\n        'nin': 'not in',\n        'gt': '>',\n        'lt': '<',\n        'gte': '>=',\n        'lte': '<=',\n        'eq': '=',\n        'neq': '!='\n    }\n    if operator not in operators_map:\n        return None\n    operator = operators_map[operator]\n    return [field, operator, value]\n\n\n@ns_conf.route('/')\nclass DatasourcesList(Resource):\n    @ns_conf.doc('get_datasources_list')\n    @ns_conf.marshal_list_with(datasource_metadata)\n    def get(self):\n        '''List all datasources'''\n        return request.default_store.get_datasources()\n\n\n@ns_conf.route('/<name>')\n@ns_conf.param('name', 'Datasource name')\nclass Datasource(Resource):\n    @ns_conf.doc('get_datasource')\n    @ns_conf.marshal_with(datasource_metadata)\n    def get(self, name):\n        '''return datasource metadata'''\n        ds = request.default_store.get_datasource(name)\n        if ds is not None:\n            return ds\n        return '', 404\n\n    @ns_conf.doc('delete_datasource')\n    def delete(self, name):\n        '''delete datasource'''\n\n        try:\n            request.default_store.delete_datasource(name)\n        except Exception as e:\n            log.error(e)\n            abort(400, str(e))\n        return '', 200\n\n    @ns_conf.doc('put_datasource', params=put_datasource_params)\n    @ns_conf.marshal_with(datasource_metadata)\n    def put(self, name):\n        '''add new datasource'''\n        data = {}\n\n        def on_field(field):\n            name = field.field_name.decode()\n            value = field.value.decode()\n            data[name] = value\n\n        file_object = None\n\n        def on_file(file):\n            nonlocal file_object\n            data['file'] = file.file_name.decode()\n            file_object = file.file_object\n\n        temp_dir_path = tempfile.mkdtemp(prefix='datasource_file_')\n\n        if request.headers['Content-Type'].startswith('multipart/form-data'):\n            parser = multipart.create_form_parser(\n                headers=request.headers,\n                on_field=on_field,\n                on_file=on_file,\n                config={\n                    'UPLOAD_DIR': temp_dir_path.encode(),    # bytes required\n                    'UPLOAD_KEEP_FILENAME': True,\n                    'UPLOAD_KEEP_EXTENSIONS': True,\n                    'MAX_MEMORY_FILE_SIZE': 0\n                }\n            )\n\n            while True:\n                chunk = request.stream.read(8192)\n                if not chunk:\n                    break\n                parser.write(chunk)\n            parser.finalize()\n            parser.close()\n\n            if file_object is not None and not file_object.closed:\n                file_object.close()\n        else:\n            data = request.json\n\n        if 'query' in data:\n            integration_id = request.json['integration_id']\n            integration = get_db_integration(integration_id, request.company_id)\n            if integration is None:\n                abort(400, f\"{integration_id} integration doesn't exist\")\n\n            if integration['type'] == 'mongodb':\n                data['find'] = data['query']\n\n            request.default_store.save_datasource(name, integration_id, data)\n            os.rmdir(temp_dir_path)\n            return request.default_store.get_datasource(name)\n\n        ds_name = data['name'] if 'name' in data else name\n        source = data['source'] if 'source' in data else name\n        source_type = data['source_type']\n\n        if source_type == 'file':\n            file_path = os.path.join(temp_dir_path, data['file'])\n            lp = file_path.lower()\n            if lp.endswith(('.zip', '.tar.gz')):\n                if lp.endswith('.zip'):\n                    with zipfile.ZipFile(file_path) as f:\n                        f.extractall(temp_dir_path)\n                elif lp.endswith('.tar.gz'):\n                    with tarfile.open(file_path) as f:\n                        f.extractall(temp_dir_path)\n                os.remove(file_path)\n                files = os.listdir(temp_dir_path)\n                if len(files) != 1:\n                    os.rmdir(temp_dir_path)\n                    return http_error(400, 'Wrong content.', 'Archive must contain only one data file.')\n                file_path = os.path.join(temp_dir_path, files[0])\n                source = files[0]\n                if not os.path.isfile(file_path):\n                    os.rmdir(temp_dir_path)\n                    return http_error(400, 'Wrong content.', 'Archive must contain data file in root.')\n        else:\n            file_path = None\n\n        request.default_store.save_datasource(ds_name, source_type, source, file_path)\n        os.rmdir(temp_dir_path)\n\n        return request.default_store.get_datasource(ds_name)\n\n\ndef analyzing_thread(name, default_store):\n    try:\n        from mindsdb.interfaces.storage.db import session\n        analysis = default_store.start_analysis(name)\n        session.close()\n    except Exception as e:\n        log.error(e)\n\n\n@ns_conf.route('/<name>/analyze')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze(Resource):\n    @ns_conf.doc('analyse_dataset')\n    def get(self, name):\n        analysis = request.default_store.get_analysis(name)\n        if analysis is not None:\n            return analysis, 200\n\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            log.error('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/analyze_refresh')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze2(Resource):\n    @ns_conf.doc('analyze_refresh_dataset')\n    def get(self, name):\n        analysis = request.default_store.get_analysis(name)\n        if analysis is not None:\n            return analysis, 200\n\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            log.error('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, request.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/data/')\n@ns_conf.param('name', 'Datasource name')\nclass DatasourceData(Resource):\n    @ns_conf.doc('get_datasource_data', params=get_datasource_rows_params)\n    @ns_conf.marshal_with(datasource_rows_metadata)\n    def get(self, name):\n        '''return data rows'''\n        ds = request.default_store.get_datasource(name)\n        if ds is None:\n            abort(400, 'No valid datasource given')\n\n        params = {\n            'page[size]': None,\n            'page[offset]': None\n        }\n        where = []\n        for key, value in request.args.items():\n            if key == 'page[size]':\n                params['page[size]'] = int(value)\n            if key == 'page[offset]':\n                params['page[offset]'] = int(value)\n            elif key.startswith('filter'):\n                param = parse_filter(key, value)\n                if param is None:\n                    abort(400, f'Not valid filter \"{key}\"')\n                where.append(param)\n\n        data_dict = request.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])\n        return data_dict, 200\n\n\n@ns_conf.route('/<name>/download')\n@ns_conf.param('name', 'Datasource name')\nclass DatasourceMissedFilesDownload(Resource):\n    @ns_conf.doc('get_datasource_download')\n    def get(self, name):\n        '''download uploaded file'''\n        ds = request.default_store.get_datasource(name)\n        if not ds:\n            abort(404, \"{} not found\".format(name))\n        # force download from s3\n        request.default_store.get_datasource_obj(name)\n        if not os.path.exists(ds['source']):\n            abort(404, \"{} not found\".format(name))\n\n        return send_file(os.path.abspath(ds['source']), as_attachment=True)\n", "path": "mindsdb/api/http/namespaces/datasource.py"}]}