We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Wrong app context for CSV task
We need to run the CSV importer task in the correct context.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `timesketch/lib/tasks.py`
Content:
```
1 # Copyright 2015 Google Inc. All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Celery task for processing Plaso storage files."""
15 
16 import os
17 import logging
18 import sys
19 
20 from flask import current_app
21 # We currently don't have plaso in our Travis setup. This is a workaround
22 # for that until we fix the Travis environment.
23 # TODO: Add Plaso to our Travis environment we are running our tests in.
24 try:
25     from plaso.frontend import psort
26 except ImportError:
27     pass
28 
29 from timesketch import create_celery_app
30 from timesketch.lib.datastores.elastic import ElasticsearchDataStore
31 from timesketch.lib.utils import read_and_validate_csv
32 from timesketch.models import db_session
33 from timesketch.models.sketch import SearchIndex
34 
35 celery = create_celery_app()
36 
37 
38 def get_data_location():
39     """Path to the plaso data directory.
40 
41     Returns:
42         The path to where the plaso data directory is or None if not existing.
43     """
44     data_location = current_app.config.get(u'PLASO_DATA_LOCATION', None)
45     if not data_location:
46         data_location = os.path.join(sys.prefix, u'share', u'plaso')
47     if not os.path.exists(data_location):
48         data_location = None
49     return data_location
50 
51 
52 @celery.task(track_started=True)
53 def run_plaso(source_file_path, timeline_name, index_name, username=None):
54     """Create a Celery task for processing Plaso storage file.
55 
56     Args:
57         source_file_path: Path to plaso storage file.
58         timeline_name: Name of the Timesketch timeline.
59         index_name: Name of the datastore index.
60         username: Username of the user who will own the timeline.
61 
62     Returns:
63         Dictionary with count of processed events.
64     """
65     plaso_data_location = get_data_location()
66     flush_interval = 1000  # events to queue before bulk index
67     doc_type = u'plaso_event'  # Document type for Elasticsearch
68 
69     # Use Plaso psort frontend tool.
70     frontend = psort.PsortFrontend()
71     frontend.SetDataLocation(plaso_data_location)
72     storage_reader = frontend.CreateStorageReader(source_file_path)
73 
74     # Setup the Timesketch output module.
75     output_module = frontend.CreateOutputModule(u'timesketch')
76     output_module.SetIndexName(index_name)
77     output_module.SetTimelineName(timeline_name)
78     output_module.SetFlushInterval(flush_interval)
79     output_module.SetDocType(doc_type)
80     if username:
81         output_module.SetUserName(username)
82 
83     # Start process the Plaso storage file.
84     counter = frontend.ExportEvents(storage_reader, output_module)
85 
86     return dict(counter)
87 
88 
89 @celery.task(track_started=True)
90 def run_csv(source_file_path, timeline_name, index_name, username=None):
91     """Create a Celery task for processing a CSV file.
92 
93     Args:
94         source_file_path: Path to CSV file.
95         timeline_name: Name of the Timesketch timeline.
96         index_name: Name of the datastore index.
97 
98     Returns:
99         Dictionary with count of processed events.
100     """
101     flush_interval = 1000  # events to queue before bulk index
102     event_type = u'generic_event'  # Document type for Elasticsearch
103 
104     # Log information to Celery
105     logging.info(u'Index name: %s', index_name)
106     logging.info(u'Timeline name: %s', timeline_name)
107     logging.info(u'Flush interval: %d', flush_interval)
108     logging.info(u'Document type: %s', event_type)
109     logging.info(u'Owner: %s', username)
110 
111     es = ElasticsearchDataStore(
112         host=current_app.config[u'ELASTIC_HOST'],
113         port=current_app.config[u'ELASTIC_PORT'])
114 
115     es.create_index(index_name=index_name, doc_type=event_type)
116     for event in read_and_validate_csv(source_file_path):
117         es.import_event(
118             flush_interval, index_name, event_type, event)
119 
120     # Import the remaining events
121     total_events = es.import_event(flush_interval, index_name, event_type)
122 
123     # We are done so let's remove the processing status flag
124     search_index = SearchIndex.query.filter_by(index_name=index_name).first()
125     search_index.status.remove(search_index.status[0])
126     db_session.add(search_index)
127     db_session.commit()
128 
129     return {u'Events processed': total_events}
130
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/timesketch/lib/tasks.py b/timesketch/lib/tasks.py
--- a/timesketch/lib/tasks.py
+++ b/timesketch/lib/tasks.py
@@ -121,9 +121,10 @@
     total_events = es.import_event(flush_interval, index_name, event_type)
 
     # We are done so let's remove the processing status flag
-    search_index = SearchIndex.query.filter_by(index_name=index_name).first()
-    search_index.status.remove(search_index.status[0])
-    db_session.add(search_index)
-    db_session.commit()
+    with celery.app.app_context():
+        search_index = SearchIndex.query.filter_by(index_name=index_name).first()
+        search_index.status.remove(search_index.status[0])
+        db_session.add(search_index)
+        db_session.commit()
 
     return {u'Events processed': total_events}

{"golden_diff": "diff --git a/timesketch/lib/tasks.py b/timesketch/lib/tasks.py\n--- a/timesketch/lib/tasks.py\n+++ b/timesketch/lib/tasks.py\n@@ -121,9 +121,10 @@\n     total_events = es.import_event(flush_interval, index_name, event_type)\n \n     # We are done so let's remove the processing status flag\n-    search_index = SearchIndex.query.filter_by(index_name=index_name).first()\n-    search_index.status.remove(search_index.status[0])\n-    db_session.add(search_index)\n-    db_session.commit()\n+    with celery.app.app_context():\n+        search_index = SearchIndex.query.filter_by(index_name=index_name).first()\n+        search_index.status.remove(search_index.status[0])\n+        db_session.add(search_index)\n+        db_session.commit()\n \n     return {u'Events processed': total_events}\n", "issue": "Wrong app context for CSV task\nWe need to run the CSV importer task in the correct context.\n", "before_files": [{"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# 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, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Celery task for processing Plaso storage files.\"\"\"\n\nimport os\nimport logging\nimport sys\n\nfrom flask import current_app\n# We currently don't have plaso in our Travis setup. This is a workaround\n# for that until we fix the Travis environment.\n# TODO: Add Plaso to our Travis environment we are running our tests in.\ntry:\n    from plaso.frontend import psort\nexcept ImportError:\n    pass\n\nfrom timesketch import create_celery_app\nfrom timesketch.lib.datastores.elastic import ElasticsearchDataStore\nfrom timesketch.lib.utils import read_and_validate_csv\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\n\ncelery = create_celery_app()\n\n\ndef get_data_location():\n    \"\"\"Path to the plaso data directory.\n\n    Returns:\n        The path to where the plaso data directory is or None if not existing.\n    \"\"\"\n    data_location = current_app.config.get(u'PLASO_DATA_LOCATION', None)\n    if not data_location:\n        data_location = os.path.join(sys.prefix, u'share', u'plaso')\n    if not os.path.exists(data_location):\n        data_location = None\n    return data_location\n\n\n@celery.task(track_started=True)\ndef run_plaso(source_file_path, timeline_name, index_name, username=None):\n    \"\"\"Create a Celery task for processing Plaso storage file.\n\n    Args:\n        source_file_path: Path to plaso storage file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    plaso_data_location = get_data_location()\n    flush_interval = 1000  # events to queue before bulk index\n    doc_type = u'plaso_event'  # Document type for Elasticsearch\n\n    # Use Plaso psort frontend tool.\n    frontend = psort.PsortFrontend()\n    frontend.SetDataLocation(plaso_data_location)\n    storage_reader = frontend.CreateStorageReader(source_file_path)\n\n    # Setup the Timesketch output module.\n    output_module = frontend.CreateOutputModule(u'timesketch')\n    output_module.SetIndexName(index_name)\n    output_module.SetTimelineName(timeline_name)\n    output_module.SetFlushInterval(flush_interval)\n    output_module.SetDocType(doc_type)\n    if username:\n        output_module.SetUserName(username)\n\n    # Start process the Plaso storage file.\n    counter = frontend.ExportEvents(storage_reader, output_module)\n\n    return dict(counter)\n\n\n@celery.task(track_started=True)\ndef run_csv(source_file_path, timeline_name, index_name, username=None):\n    \"\"\"Create a Celery task for processing a CSV file.\n\n    Args:\n        source_file_path: Path to CSV file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    flush_interval = 1000  # events to queue before bulk index\n    event_type = u'generic_event'  # Document type for Elasticsearch\n\n    # Log information to Celery\n    logging.info(u'Index name: %s', index_name)\n    logging.info(u'Timeline name: %s', timeline_name)\n    logging.info(u'Flush interval: %d', flush_interval)\n    logging.info(u'Document type: %s', event_type)\n    logging.info(u'Owner: %s', username)\n\n    es = ElasticsearchDataStore(\n        host=current_app.config[u'ELASTIC_HOST'],\n        port=current_app.config[u'ELASTIC_PORT'])\n\n    es.create_index(index_name=index_name, doc_type=event_type)\n    for event in read_and_validate_csv(source_file_path):\n        es.import_event(\n            flush_interval, index_name, event_type, event)\n\n    # Import the remaining events\n    total_events = es.import_event(flush_interval, index_name, event_type)\n\n    # We are done so let's remove the processing status flag\n    search_index = SearchIndex.query.filter_by(index_name=index_name).first()\n    search_index.status.remove(search_index.status[0])\n    db_session.add(search_index)\n    db_session.commit()\n\n    return {u'Events processed': total_events}\n", "path": "timesketch/lib/tasks.py"}], "after_files": [{"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# 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, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Celery task for processing Plaso storage files.\"\"\"\n\nimport os\nimport logging\nimport sys\n\nfrom flask import current_app\n# We currently don't have plaso in our Travis setup. This is a workaround\n# for that until we fix the Travis environment.\n# TODO: Add Plaso to our Travis environment we are running our tests in.\ntry:\n    from plaso.frontend import psort\nexcept ImportError:\n    pass\n\nfrom timesketch import create_celery_app\nfrom timesketch.lib.datastores.elastic import ElasticsearchDataStore\nfrom timesketch.lib.utils import read_and_validate_csv\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\n\ncelery = create_celery_app()\n\n\ndef get_data_location():\n    \"\"\"Path to the plaso data directory.\n\n    Returns:\n        The path to where the plaso data directory is or None if not existing.\n    \"\"\"\n    data_location = current_app.config.get(u'PLASO_DATA_LOCATION', None)\n    if not data_location:\n        data_location = os.path.join(sys.prefix, u'share', u'plaso')\n    if not os.path.exists(data_location):\n        data_location = None\n    return data_location\n\n\n@celery.task(track_started=True)\ndef run_plaso(source_file_path, timeline_name, index_name, username=None):\n    \"\"\"Create a Celery task for processing Plaso storage file.\n\n    Args:\n        source_file_path: Path to plaso storage file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    plaso_data_location = get_data_location()\n    flush_interval = 1000  # events to queue before bulk index\n    doc_type = u'plaso_event'  # Document type for Elasticsearch\n\n    # Use Plaso psort frontend tool.\n    frontend = psort.PsortFrontend()\n    frontend.SetDataLocation(plaso_data_location)\n    storage_reader = frontend.CreateStorageReader(source_file_path)\n\n    # Setup the Timesketch output module.\n    output_module = frontend.CreateOutputModule(u'timesketch')\n    output_module.SetIndexName(index_name)\n    output_module.SetTimelineName(timeline_name)\n    output_module.SetFlushInterval(flush_interval)\n    output_module.SetDocType(doc_type)\n    if username:\n        output_module.SetUserName(username)\n\n    # Start process the Plaso storage file.\n    counter = frontend.ExportEvents(storage_reader, output_module)\n\n    return dict(counter)\n\n\n@celery.task(track_started=True)\ndef run_csv(source_file_path, timeline_name, index_name, username=None):\n    \"\"\"Create a Celery task for processing a CSV file.\n\n    Args:\n        source_file_path: Path to CSV file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    flush_interval = 1000  # events to queue before bulk index\n    event_type = u'generic_event'  # Document type for Elasticsearch\n\n    # Log information to Celery\n    logging.info(u'Index name: %s', index_name)\n    logging.info(u'Timeline name: %s', timeline_name)\n    logging.info(u'Flush interval: %d', flush_interval)\n    logging.info(u'Document type: %s', event_type)\n    logging.info(u'Owner: %s', username)\n\n    es = ElasticsearchDataStore(\n        host=current_app.config[u'ELASTIC_HOST'],\n        port=current_app.config[u'ELASTIC_PORT'])\n\n    es.create_index(index_name=index_name, doc_type=event_type)\n    for event in read_and_validate_csv(source_file_path):\n        es.import_event(\n            flush_interval, index_name, event_type, event)\n\n    # Import the remaining events\n    total_events = es.import_event(flush_interval, index_name, event_type)\n\n    # We are done so let's remove the processing status flag\n    with celery.app.app_context():\n        search_index = SearchIndex.query.filter_by(index_name=index_name).first()\n        search_index.status.remove(search_index.status[0])\n        db_session.add(search_index)\n        db_session.commit()\n\n    return {u'Events processed': total_events}\n", "path": "timesketch/lib/tasks.py"}]}