We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Error if try create datasource from not existing integration
If send query `PUT api/datasources/ds_name` with data:
```
{
    "integration_id": 'unexists_integration',
    "name": 'ds_name',
    "query": f"select * from test_data.any_data limit 100;"
}
```
then in response will be error:
```
{
   "message": 'TypeError: expected str, bytes or os.PathLike object, not dict'
}
```

--- 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 datetime
2 import os
3 import threading
4 import tempfile
5 import re
6 import multipart
7 
8 import mindsdb
9 from dateutil.parser import parse
10 from flask import request, send_file
11 from flask_restx import Resource, abort     # 'abort' using to return errors as json: {'message': 'error text'}
12 from flask import current_app as ca
13 
14 from mindsdb.api.http.namespaces.configs.datasources import ns_conf
15 from mindsdb.api.http.namespaces.entitites.datasources.datasource import (
16     datasource_metadata,
17     put_datasource_params
18 )
19 from mindsdb.api.http.namespaces.entitites.datasources.datasource_data import (
20     get_datasource_rows_params,
21     datasource_rows_metadata
22 )
23 from mindsdb.api.http.namespaces.entitites.datasources.datasource_files import (
24     put_datasource_file_params
25 )
26 from mindsdb.api.http.namespaces.entitites.datasources.datasource_missed_files import (
27     datasource_missed_files_metadata,
28     get_datasource_missed_files_params
29 )
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 ca.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 = ca.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         try:
78             ca.default_store.delete_datasource(name)
79         except Exception as e:
80             print(e)
81             abort(400, str(e))
82         return '', 200
83 
84     @ns_conf.doc('put_datasource', params=put_datasource_params)
85     @ns_conf.marshal_with(datasource_metadata)
86     def put(self, name):
87         '''add new datasource'''
88         data = {}
89 
90         def on_field(field):
91             name = field.field_name.decode()
92             value = field.value.decode()
93             data[name] = value
94 
95         file_object = None
96 
97         def on_file(file):
98             nonlocal file_object
99             data['file'] = file.file_name.decode()
100             file_object = file.file_object
101 
102         temp_dir_path = tempfile.mkdtemp(prefix='datasource_file_')
103 
104         if request.headers['Content-Type'].startswith('multipart/form-data'):
105             parser = multipart.create_form_parser(
106                 headers=request.headers,
107                 on_field=on_field,
108                 on_file=on_file,
109                 config={
110                     'UPLOAD_DIR': temp_dir_path.encode(),    # bytes required
111                     'UPLOAD_KEEP_FILENAME': True,
112                     'UPLOAD_KEEP_EXTENSIONS': True,
113                     'MAX_MEMORY_FILE_SIZE': 0
114                 }
115             )
116 
117             while True:
118                 chunk = request.stream.read(8192)
119                 if not chunk:
120                     break
121                 parser.write(chunk)
122             parser.finalize()
123             parser.close()
124 
125             if file_object is not None and not file_object.closed:
126                 file_object.close()
127         else:
128             data = request.json
129 
130         if 'query' in data:
131             source_type = request.json['integration_id']
132             ca.default_store.save_datasource(name, source_type, request.json)
133             os.rmdir(temp_dir_path)
134             return ca.default_store.get_datasource(name)
135 
136         ds_name = data['name'] if 'name' in data else name
137         source = data['source'] if 'source' in data else name
138         source_type = data['source_type']
139 
140         if source_type == 'file':
141             file_path = os.path.join(temp_dir_path, data['file'])
142         else:
143             file_path = None
144 
145         ca.default_store.save_datasource(ds_name, source_type, source, file_path)
146         os.rmdir(temp_dir_path)
147 
148         return ca.default_store.get_datasource(ds_name)
149 
150 
151 ds_analysis = {}
152 
153 
154 def analyzing_thread(name, default_store):
155     global ds_analysis
156     ds_analysis[name] = None
157     ds = default_store.get_datasource(name)
158     analysis = default_store.get_analysis(ds['name'])
159     ds_analysis[name] = {
160         'created_at': datetime.datetime.utcnow(),
161         'data': analysis
162     }
163 
164 
165 @ns_conf.route('/<name>/analyze')
166 @ns_conf.param('name', 'Datasource name')
167 class Analyze(Resource):
168     @ns_conf.doc('analyse_dataset')
169     def get(self, name):
170         global ds_analysis
171         if name in ds_analysis:
172             if ds_analysis[name] is None:
173                 return {'status': 'analyzing'}, 200
174             else:
175                 analysis = ds_analysis[name]['data']
176                 return analysis, 200
177 
178         ds = ca.default_store.get_datasource(name)
179         if ds is None:
180             print('No valid datasource given')
181             abort(400, 'No valid datasource given')
182 
183         x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))
184         x.start()
185         return {'status': 'analyzing'}, 200
186 
187 @ns_conf.route('/<name>/analyze_refresh')
188 @ns_conf.param('name', 'Datasource name')
189 class Analyze(Resource):
190     @ns_conf.doc('analyze_refresh_dataset')
191     def get(self, name):
192         global ds_analysis
193         if name in ds_analysis:
194             if ds_analysis[name] is None:
195                 return {'status': 'analyzing'}, 200
196             else:
197                 del ds_analysis[name]
198 
199         ds = ca.default_store.get_datasource(name)
200         if ds is None:
201             print('No valid datasource given')
202             abort(400, 'No valid datasource given')
203 
204         x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))
205         x.start()
206         return {'status': 'analyzing'}, 200
207 
208 
209 @ns_conf.route('/<name>/analyze_subset')
210 @ns_conf.param('name', 'Datasource name')
211 class AnalyzeSubset(Resource):
212     @ns_conf.doc('analyse_datasubset')
213     def get(self, name):
214         ds = ca.default_store.get_datasource(name)
215         if ds is None:
216             print('No valid datasource given')
217             abort(400, 'No valid datasource given')
218 
219         where = []
220         for key, value in request.args.items():
221             if key.startswith('filter'):
222                 param = parse_filter(key, value)
223                 if param is None:
224                     abort(400, f'Not valid filter "{key}"')
225                 where.append(param)
226 
227         data_dict = ca.default_store.get_data(ds['name'], where)
228 
229         if data_dict['rowcount'] == 0:
230             return abort(400, 'Empty dataset after filters applying')
231 
232         return get_analysis(pd.DataFrame(data_dict['data'])), 200
233 
234 
235 @ns_conf.route('/<name>/data/')
236 @ns_conf.param('name', 'Datasource name')
237 class DatasourceData(Resource):
238     @ns_conf.doc('get_datasource_data', params=get_datasource_rows_params)
239     @ns_conf.marshal_with(datasource_rows_metadata)
240     def get(self, name):
241         '''return data rows'''
242         ds = ca.default_store.get_datasource(name)
243         if ds is None:
244             abort(400, 'No valid datasource given')
245 
246         params = {
247             'page[size]': None,
248             'page[offset]': None
249         }
250         where = []
251         for key, value in request.args.items():
252             if key == 'page[size]':
253                 params['page[size]'] = int(value)
254             if key == 'page[offset]':
255                 params['page[offset]'] = int(value)
256             elif key.startswith('filter'):
257                 param = parse_filter(key, value)
258                 if param is None:
259                     abort(400, f'Not valid filter "{key}"')
260                 where.append(param)
261 
262         data_dict = ca.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])
263 
264         return data_dict, 200
265 
266 
267 @ns_conf.route('/<name>/download')
268 @ns_conf.param('name', 'Datasource name')
269 class DatasourceMissedFilesDownload(Resource):
270     @ns_conf.doc('get_datasource_download')
271     def get(self, name):
272         '''download uploaded file'''
273         ds = ca.default_store.get_datasource(name)
274         if not ds:
275             abort(404, "{} not found".format(name))
276         if not os.path.exists(ds['source']):
277             abort(404, "{} not found".format(name))
278 
279         return send_file(os.path.abspath(ds['source']), as_attachment=True)
280
```
--- 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
@@ -129,6 +129,10 @@
 
         if 'query' in data:
             source_type = request.json['integration_id']
+            if source_type not in ca.default_store.config['integrations']:
+                # integration doens't exist
+                abort(400, f"{source_type} integration doesn't exist")
+
             ca.default_store.save_datasource(name, source_type, request.json)
             os.rmdir(temp_dir_path)
             return ca.default_store.get_datasource(name)

{"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@@ -129,6 +129,10 @@\n \n         if 'query' in data:\n             source_type = request.json['integration_id']\n+            if source_type not in ca.default_store.config['integrations']:\n+                # integration doens't exist\n+                abort(400, f\"{source_type} integration doesn't exist\")\n+\n             ca.default_store.save_datasource(name, source_type, request.json)\n             os.rmdir(temp_dir_path)\n             return ca.default_store.get_datasource(name)\n", "issue": "Error if try create datasource from not existing integration\nIf send query `PUT api/datasources/ds_name` with data:\r\n```\r\n{\r\n    \"integration_id\": 'unexists_integration',\r\n    \"name\": 'ds_name',\r\n    \"query\": f\"select * from test_data.any_data limit 100;\"\r\n}\r\n```\r\nthen in response will be error:\r\n```\r\n{\r\n   \"message\": 'TypeError: expected str, bytes or os.PathLike object, not dict'\r\n}\r\n```\n", "before_files": [{"content": "import datetime\nimport os\nimport threading\nimport tempfile\nimport re\nimport multipart\n\nimport mindsdb\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.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)\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 ca.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 = ca.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        try:\n            ca.default_store.delete_datasource(name)\n        except Exception as e:\n            print(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            source_type = request.json['integration_id']\n            ca.default_store.save_datasource(name, source_type, request.json)\n            os.rmdir(temp_dir_path)\n            return ca.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        ca.default_store.save_datasource(ds_name, source_type, source, file_path)\n        os.rmdir(temp_dir_path)\n\n        return ca.default_store.get_datasource(ds_name)\n\n\nds_analysis = {}\n\n\ndef analyzing_thread(name, default_store):\n    global ds_analysis\n    ds_analysis[name] = None\n    ds = default_store.get_datasource(name)\n    analysis = default_store.get_analysis(ds['name'])\n    ds_analysis[name] = {\n        'created_at': datetime.datetime.utcnow(),\n        'data': analysis\n    }\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        global ds_analysis\n        if name in ds_analysis:\n            if ds_analysis[name] is None:\n                return {'status': 'analyzing'}, 200\n            else:\n                analysis = ds_analysis[name]['data']\n                return analysis, 200\n\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n@ns_conf.route('/<name>/analyze_refresh')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze(Resource):\n    @ns_conf.doc('analyze_refresh_dataset')\n    def get(self, name):\n        global ds_analysis\n        if name in ds_analysis:\n            if ds_analysis[name] is None:\n                return {'status': 'analyzing'}, 200\n            else:\n                del ds_analysis[name]\n\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/analyze_subset')\n@ns_conf.param('name', 'Datasource name')\nclass AnalyzeSubset(Resource):\n    @ns_conf.doc('analyse_datasubset')\n    def get(self, name):\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        where = []\n        for key, value in request.args.items():\n            if 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 = ca.default_store.get_data(ds['name'], where)\n\n        if data_dict['rowcount'] == 0:\n            return abort(400, 'Empty dataset after filters applying')\n\n        return get_analysis(pd.DataFrame(data_dict['data'])), 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 = ca.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 = ca.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])\n\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 = ca.default_store.get_datasource(name)\n        if not ds:\n            abort(404, \"{} not found\".format(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 datetime\nimport os\nimport threading\nimport tempfile\nimport re\nimport multipart\n\nimport mindsdb\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.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)\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 ca.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 = ca.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        try:\n            ca.default_store.delete_datasource(name)\n        except Exception as e:\n            print(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            source_type = request.json['integration_id']\n            if source_type not in ca.default_store.config['integrations']:\n                # integration doens't exist\n                abort(400, f\"{source_type} integration doesn't exist\")\n\n            ca.default_store.save_datasource(name, source_type, request.json)\n            os.rmdir(temp_dir_path)\n            return ca.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        ca.default_store.save_datasource(ds_name, source_type, source, file_path)\n        os.rmdir(temp_dir_path)\n\n        return ca.default_store.get_datasource(ds_name)\n\n\nds_analysis = {}\n\n\ndef analyzing_thread(name, default_store):\n    global ds_analysis\n    ds_analysis[name] = None\n    ds = default_store.get_datasource(name)\n    analysis = default_store.get_analysis(ds['name'])\n    ds_analysis[name] = {\n        'created_at': datetime.datetime.utcnow(),\n        'data': analysis\n    }\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        global ds_analysis\n        if name in ds_analysis:\n            if ds_analysis[name] is None:\n                return {'status': 'analyzing'}, 200\n            else:\n                analysis = ds_analysis[name]['data']\n                return analysis, 200\n\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n@ns_conf.route('/<name>/analyze_refresh')\n@ns_conf.param('name', 'Datasource name')\nclass Analyze(Resource):\n    @ns_conf.doc('analyze_refresh_dataset')\n    def get(self, name):\n        global ds_analysis\n        if name in ds_analysis:\n            if ds_analysis[name] is None:\n                return {'status': 'analyzing'}, 200\n            else:\n                del ds_analysis[name]\n\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        x = threading.Thread(target=analyzing_thread, args=(name, ca.default_store))\n        x.start()\n        return {'status': 'analyzing'}, 200\n\n\n@ns_conf.route('/<name>/analyze_subset')\n@ns_conf.param('name', 'Datasource name')\nclass AnalyzeSubset(Resource):\n    @ns_conf.doc('analyse_datasubset')\n    def get(self, name):\n        ds = ca.default_store.get_datasource(name)\n        if ds is None:\n            print('No valid datasource given')\n            abort(400, 'No valid datasource given')\n\n        where = []\n        for key, value in request.args.items():\n            if 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 = ca.default_store.get_data(ds['name'], where)\n\n        if data_dict['rowcount'] == 0:\n            return abort(400, 'Empty dataset after filters applying')\n\n        return get_analysis(pd.DataFrame(data_dict['data'])), 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 = ca.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 = ca.default_store.get_data(name, where, params['page[size]'], params['page[offset]'])\n\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 = ca.default_store.get_datasource(name)\n        if not ds:\n            abort(404, \"{} not found\".format(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"}]}