prompt: 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()
  
  ```
  
golden_diff: 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}
  
verification_info: {"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"}]}