Repository: google/turbinia (language: python)

Issue:
Final state_manager update from worker happens too late
The state_manager is already removed (probably because of serialization) so the worker can't update Datastore/Redis after the Task has been completed.  This shows up in the logs as:

`No state_manager initialized, not updating Task info`

We should move this update to happen earlier.


Discussion / hints:

Fix patch:
diff --git a/.pylintrc b/.pylintrc
index 0913992d9..8a00c4fca 100644
--- a/.pylintrc
+++ b/.pylintrc
@@ -102,7 +102,6 @@ disable=abstract-method,
         no-init,  # added
         no-member,
         no-name-in-module,
-        no-self-use,
         nonzero-method,
         oct-method,
         old-division,
diff --git a/.style.yapf b/.style.yapf
index f2027a283..d411357b9 100644
--- a/.style.yapf
+++ b/.style.yapf
@@ -4,6 +4,7 @@
 #
 [style]
 based_on_style = yapf
+COLUMN_LIMIT = 80
 COALESCE_BRACKETS = True
 SPLIT_BEFORE_FIRST_ARGUMENT = True
 SPLIT_PENALTY_AFTER_OPENING_BRACKET = 0
diff --git a/docs/conf.py b/docs/conf.py
index 98b8b5b4d..8ba991e9e 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -14,8 +14,6 @@
 # import sys
 # sys.path.insert(0, os.path.abspath('.'))
 
-from __future__ import unicode_literals
-
 from recommonmark.transform import AutoStructify
 from docutils import nodes, transforms
 
diff --git a/poetry.lock b/poetry.lock
index d4cd8782f..4f287192c 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -3974,13 +3974,13 @@ ocsp = ["cryptography (>=36.0.1)", "pyopenssl (==20.0.1)", "requests (>=2.26.0)"
 
 [[package]]
 name = "requests"
-version = "2.31.0"
+version = "2.32.3"
 description = "Python HTTP for Humans."
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "requests-2.31.0-py3-none-any.whl", hash = "sha256:58cd2187c01e70e6e26505bca751777aa9f2ee0b7f4300988b709f44e013003f"},
-    {file = "requests-2.31.0.tar.gz", hash = "sha256:942c5a758f98d790eaed1a29cb6eefc7ffb0d1cf7af05c3d2791656dbd6ad1e1"},
+    {file = "requests-2.32.3-py3-none-any.whl", hash = "sha256:70761cfe03c773ceb22aa2f671b4757976145175cdfca038c02654d061d6dcc6"},
+    {file = "requests-2.32.3.tar.gz", hash = "sha256:55365417734eb18255590a9ff9eb97e9e1da868d4ccd6402399eaf68af20a760"},
 ]
 
 [package.dependencies]
diff --git a/pyproject.toml b/pyproject.toml
index 3a82a0dc1..ea29dbc12 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -67,4 +67,4 @@ ignore_patterns = [
 
 [build-system]
 requires = ["poetry-core"]
-build-backend = "poetry.core.masonry.api"
+build-backend = "poetry.core.masonry.api"
\ No newline at end of file
diff --git a/tools/turbinia_config_parser.py b/tools/turbinia_config_parser.py
index 15cd0d2be..8abe0dad9 100644
--- a/tools/turbinia_config_parser.py
+++ b/tools/turbinia_config_parser.py
@@ -14,7 +14,6 @@
 # # limitations under the License.
 """Turbinia Config Parser."""
 
-from __future__ import unicode_literals
 from __future__ import print_function
 
 import imp
diff --git a/tools/turbinia_job_graph.py b/tools/turbinia_job_graph.py
index dc577b96a..cd226dd97 100755
--- a/tools/turbinia_job_graph.py
+++ b/tools/turbinia_job_graph.py
@@ -16,7 +16,6 @@
 """Graph to visualise job/evidence relationships."""
 
 from __future__ import print_function
-from __future__ import unicode_literals
 
 import argparse
 import graphviz
diff --git a/turbinia/api/cli/poetry.lock b/turbinia/api/cli/poetry.lock
index 1da7c2bc6..27e2bfa6b 100644
--- a/turbinia/api/cli/poetry.lock
+++ b/turbinia/api/cli/poetry.lock
@@ -448,13 +448,13 @@ files = [
 
 [[package]]
 name = "requests"
-version = "2.31.0"
+version = "2.32.3"
 description = "Python HTTP for Humans."
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "requests-2.31.0-py3-none-any.whl", hash = "sha256:58cd2187c01e70e6e26505bca751777aa9f2ee0b7f4300988b709f44e013003f"},
-    {file = "requests-2.31.0.tar.gz", hash = "sha256:942c5a758f98d790eaed1a29cb6eefc7ffb0d1cf7af05c3d2791656dbd6ad1e1"},
+    {file = "requests-2.32.3-py3-none-any.whl", hash = "sha256:70761cfe03c773ceb22aa2f671b4757976145175cdfca038c02654d061d6dcc6"},
+    {file = "requests-2.32.3.tar.gz", hash = "sha256:55365417734eb18255590a9ff9eb97e9e1da868d4ccd6402399eaf68af20a760"},
 ]
 
 [package.dependencies]
diff --git a/turbinia/api/cli/turbinia_client/core/groups.py b/turbinia/api/cli/turbinia_client/core/groups.py
index bd55dce55..e7f9a11c6 100644
--- a/turbinia/api/cli/turbinia_client/core/groups.py
+++ b/turbinia/api/cli/turbinia_client/core/groups.py
@@ -77,4 +77,3 @@ def submit_group(ctx: click.Context):
   of available evidence types.
   """
   ctx.invoke(setup_submit)
-  click.echo(submit_group.get_help(ctx))
diff --git a/turbinia/api/cli/turbinia_client/helpers/auth_helper.py b/turbinia/api/cli/turbinia_client/helpers/auth_helper.py
index 105069d24..c8fc9287e 100644
--- a/turbinia/api/cli/turbinia_client/helpers/auth_helper.py
+++ b/turbinia/api/cli/turbinia_client/helpers/auth_helper.py
@@ -40,7 +40,7 @@ def get_oauth2_token_id(credentials_path, client_secrets_path):
       log.error(f'Error loading credentials: {exception!s}')
     # Refresh credentials using existing refresh_token
     if credentials and credentials.refresh_token:
-      log.debug('Found a refresh token. Requesting new id_token...')
+      log.debug('Found a refresh token. Requesting new id_token..')
       try:
         credentials.refresh(Request())
       except google_exceptions.RefreshError as exception:
diff --git a/turbinia/api/cli/turbinia_client/turbiniacli_tool.py b/turbinia/api/cli/turbinia_client/turbiniacli_tool.py
index 9453ea4c3..d0157b7ff 100644
--- a/turbinia/api/cli/turbinia_client/turbiniacli_tool.py
+++ b/turbinia/api/cli/turbinia_client/turbiniacli_tool.py
@@ -38,7 +38,7 @@
     show_default=True, show_envvar=True, type=str,
     default=lambda: os.environ.get('TURBINIA_CONFIG_INSTANCE', 'default'))
 @click.option(
-    '--config_path', '-p', help='Path to the .turbinia_api_config.json file..',
+    '--config_path', '-p', help='Path to the .turbinia_api_config.json file.',
     show_default=True, show_envvar=True, type=str,
     default=lambda: os.environ.get('TURBINIA_API_CONFIG_PATH', '~'))
 @click.pass_context
diff --git a/turbinia/api/client/turbinia_api_lib/api/turbinia_configuration_api.py b/turbinia/api/client/turbinia_api_lib/api/turbinia_configuration_api.py
index 87a850460..7e7aeb5f9 100644
--- a/turbinia/api/client/turbinia_api_lib/api/turbinia_configuration_api.py
+++ b/turbinia/api/client/turbinia_api_lib/api/turbinia_configuration_api.py
@@ -11,7 +11,6 @@
     Do not edit the class manually.
 """  # noqa: E501
 
-
 import re  # noqa: F401
 import io
 import warnings
@@ -20,7 +19,6 @@
 
 from typing import Any
 
-
 from turbinia_api_lib.api_client import ApiClient
 from turbinia_api_lib.api_response import ApiResponse
 from turbinia_api_lib.exceptions import (  # noqa: F401
@@ -30,20 +28,20 @@
 
 
 class TurbiniaConfigurationApi:
-    """NOTE: This class is auto generated by OpenAPI Generator
+  """NOTE: This class is auto generated by OpenAPI Generator
     Ref: https://openapi-generator.tech
 
     Do not edit the class manually.
     """
 
-    def __init__(self, api_client=None) -> None:
-        if api_client is None:
-            api_client = ApiClient.get_default()
-        self.api_client = api_client
+  def __init__(self, api_client=None) -> None:
+    if api_client is None:
+      api_client = ApiClient.get_default()
+    self.api_client = api_client
 
-    @validate_arguments
-    def get_request_options(self, **kwargs) -> object:  # noqa: E501
-        """Get Request Options  # noqa: E501
+  @validate_arguments
+  def get_request_options(self, **kwargs) -> object:  # noqa: E501
+    """Get Request Options  # noqa: E501
 
         Returns a list BaseRequestOptions attributes.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -63,15 +61,16 @@ def get_request_options(self, **kwargs) -> object:  # noqa: E501
                  returns the request thread.
         :rtype: object
         """
-        kwargs['_return_http_data_only'] = True
-        if '_preload_content' in kwargs:
-            message = "Error! Please call the get_request_options_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
-            raise ValueError(message)
-        return self.get_request_options_with_http_info(**kwargs)  # noqa: E501
+    kwargs['_return_http_data_only'] = True
+    if '_preload_content' in kwargs:
+      message = "Error! Please call the get_request_options_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
+      raise ValueError(message)
+    return self.get_request_options_with_http_info(**kwargs)  # noqa: E501
 
-    @validate_arguments
-    def get_request_options_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
-        """Get Request Options  # noqa: E501
+  @validate_arguments
+  def get_request_options_with_http_info(
+      self, **kwargs) -> ApiResponse:  # noqa: E501
+    """Get Request Options  # noqa: E501
 
         Returns a list BaseRequestOptions attributes.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -105,77 +104,70 @@ def get_request_options_with_http_info(self, **kwargs) -> ApiResponse:  # noqa:
         :rtype: tuple(object, status_code(int), headers(HTTPHeaderDict))
         """
 
-        _params = locals()
-
-        _all_params = [
-        ]
-        _all_params.extend(
-            [
-                'async_req',
-                '_return_http_data_only',
-                '_preload_content',
-                '_request_timeout',
-                '_request_auth',
-                '_content_type',
-                '_headers'
-            ]
-        )
-
-        # validate the arguments
-        for _key, _val in _params['kwargs'].items():
-            if _key not in _all_params:
-                raise ApiTypeError(
-                    "Got an unexpected keyword argument '%s'"
-                    " to method get_request_options" % _key
-                )
-            _params[_key] = _val
-        del _params['kwargs']
-
-        _collection_formats = {}
-
-        # process the path parameters
-        _path_params = {}
-
-        # process the query parameters
-        _query_params = []
-        # process the header parameters
-        _header_params = dict(_params.get('_headers', {}))
-        # process the form parameters
-        _form_params = []
-        _files = {}
-        # process the body parameter
-        _body_params = None
-        # set the HTTP header `Accept`
-        _header_params['Accept'] = self.api_client.select_header_accept(
-            ['application/json'])  # noqa: E501
-
-        # authentication setting
-        _auth_settings = ['oAuth2']  # noqa: E501
-
-        _response_types_map = {
-            '200': "object",
-        }
-
-        return self.api_client.call_api(
-            '/api/config/request_options', 'GET',
-            _path_params,
-            _query_params,
-            _header_params,
-            body=_body_params,
-            post_params=_form_params,
-            files=_files,
-            response_types_map=_response_types_map,
-            auth_settings=_auth_settings,
-            async_req=_params.get('async_req'),
-            _return_http_data_only=_params.get('_return_http_data_only'),  # noqa: E501
-            _preload_content=_params.get('_preload_content', True),
-            _request_timeout=_params.get('_request_timeout'),
-            collection_formats=_collection_formats,
-            _request_auth=_params.get('_request_auth'))
-
-    @validate_arguments
-    def get_version(self, **kwargs) -> object:  # noqa: E501
-        """Get Version  # noqa: E501
+    _params = locals()
+
+    _all_params = []
+    _all_params.extend([
+        'async_req', '_return_http_data_only', '_preload_content',
+        '_request_timeout', '_request_auth', '_content_type', '_headers'
+    ])
+
+    # validate the arguments
+    for _key, _val in _params['kwargs'].items():
+      if _key not in _all_params:
+        raise ApiTypeError(
+            "Got an unexpected keyword argument '%s'"
+            " to method get_request_options" % _key)
+      _params[_key] = _val
+    del _params['kwargs']
+
+    _collection_formats = {}
+
+    # process the path parameters
+    _path_params = {}
+
+    # process the query parameters
+    _query_params = []
+    # process the header parameters
+    _header_params = dict(_params.get('_headers', {}))
+    # process the form parameters
+    _form_params = []
+    _files = {}
+    # process the body parameter
+    _body_params = None
+    # set the HTTP header `Accept`
+    _header_params['Accept'] = self.api_client.select_header_accept(
+        ['application/json'])  # noqa: E501
+
+    # authentication setting
+    _auth_settings = ['oAuth2']  # noqa: E501
+
+    _response_types_map = {
+        '200': "object",
+    }
+
+    return self.api_client.call_api(
+        '/api/config/request_options',
+        'GET',
+        _path_params,
+        _query_params,
+        _header_params,
+        body=_body_params,
+        post_params=_form_params,
+        files=_files,
+        response_types_map=_response_types_map,
+        auth_settings=_auth_settings,
+        async_req=_params.get('async_req'),
+        _return_http_data_only=_params.get(
+            '_return_http_data_only'),  # noqa: E501
+        _preload_content=_params.get('_preload_content', True),
+        _request_timeout=_params.get('_request_timeout'),
+        collection_formats=_collection_formats,
+        _request_auth=_params.get('_request_auth'))
+
+  @validate_arguments
+  def get_version(self, **kwargs) -> object:  # noqa: E501
+    """Get Version  # noqa: E501
 
         Returns the Turbinia version.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -195,15 +187,15 @@ def get_version(self, **kwargs) -> object:  # noqa: E501
                  returns the request thread.
         :rtype: object
         """
-        kwargs['_return_http_data_only'] = True
-        if '_preload_content' in kwargs:
-            message = "Error! Please call the get_version_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
-            raise ValueError(message)
-        return self.get_version_with_http_info(**kwargs)  # noqa: E501
+    kwargs['_return_http_data_only'] = True
+    if '_preload_content' in kwargs:
+      message = "Error! Please call the get_version_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
+      raise ValueError(message)
+    return self.get_version_with_http_info(**kwargs)  # noqa: E501
 
-    @validate_arguments
-    def get_version_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
-        """Get Version  # noqa: E501
+  @validate_arguments
+  def get_version_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
+    """Get Version  # noqa: E501
 
         Returns the Turbinia version.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -237,77 +229,70 @@ def get_version_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
         :rtype: tuple(object, status_code(int), headers(HTTPHeaderDict))
         """
 
-        _params = locals()
-
-        _all_params = [
-        ]
-        _all_params.extend(
-            [
-                'async_req',
-                '_return_http_data_only',
-                '_preload_content',
-                '_request_timeout',
-                '_request_auth',
-                '_content_type',
-                '_headers'
-            ]
-        )
-
-        # validate the arguments
-        for _key, _val in _params['kwargs'].items():
-            if _key not in _all_params:
-                raise ApiTypeError(
-                    "Got an unexpected keyword argument '%s'"
-                    " to method get_version" % _key
-                )
-            _params[_key] = _val
-        del _params['kwargs']
-
-        _collection_formats = {}
-
-        # process the path parameters
-        _path_params = {}
-
-        # process the query parameters
-        _query_params = []
-        # process the header parameters
-        _header_params = dict(_params.get('_headers', {}))
-        # process the form parameters
-        _form_params = []
-        _files = {}
-        # process the body parameter
-        _body_params = None
-        # set the HTTP header `Accept`
-        _header_params['Accept'] = self.api_client.select_header_accept(
-            ['application/json'])  # noqa: E501
-
-        # authentication setting
-        _auth_settings = ['oAuth2']  # noqa: E501
-
-        _response_types_map = {
-            '200': "object",
-        }
-
-        return self.api_client.call_api(
-            '/api/config/version', 'GET',
-            _path_params,
-            _query_params,
-            _header_params,
-            body=_body_params,
-            post_params=_form_params,
-            files=_files,
-            response_types_map=_response_types_map,
-            auth_settings=_auth_settings,
-            async_req=_params.get('async_req'),
-            _return_http_data_only=_params.get('_return_http_data_only'),  # noqa: E501
-            _preload_content=_params.get('_preload_content', True),
-            _request_timeout=_params.get('_request_timeout'),
-            collection_formats=_collection_formats,
-            _request_auth=_params.get('_request_auth'))
-
-    @validate_arguments
-    def read_config(self, **kwargs) -> object:  # noqa: E501
-        """Read Config  # noqa: E501
+    _params = locals()
+
+    _all_params = []
+    _all_params.extend([
+        'async_req', '_return_http_data_only', '_preload_content',
+        '_request_timeout', '_request_auth', '_content_type', '_headers'
+    ])
+
+    # validate the arguments
+    for _key, _val in _params['kwargs'].items():
+      if _key not in _all_params:
+        raise ApiTypeError(
+            "Got an unexpected keyword argument '%s'"
+            " to method get_version" % _key)
+      _params[_key] = _val
+    del _params['kwargs']
+
+    _collection_formats = {}
+
+    # process the path parameters
+    _path_params = {}
+
+    # process the query parameters
+    _query_params = []
+    # process the header parameters
+    _header_params = dict(_params.get('_headers', {}))
+    # process the form parameters
+    _form_params = []
+    _files = {}
+    # process the body parameter
+    _body_params = None
+    # set the HTTP header `Accept`
+    _header_params['Accept'] = self.api_client.select_header_accept(
+        ['application/json'])  # noqa: E501
+
+    # authentication setting
+    _auth_settings = ['oAuth2']  # noqa: E501
+
+    _response_types_map = {
+        '200': "object",
+    }
+
+    return self.api_client.call_api(
+        '/api/config/version',
+        'GET',
+        _path_params,
+        _query_params,
+        _header_params,
+        body=_body_params,
+        post_params=_form_params,
+        files=_files,
+        response_types_map=_response_types_map,
+        auth_settings=_auth_settings,
+        async_req=_params.get('async_req'),
+        _return_http_data_only=_params.get(
+            '_return_http_data_only'),  # noqa: E501
+        _preload_content=_params.get('_preload_content', True),
+        _request_timeout=_params.get('_request_timeout'),
+        collection_formats=_collection_formats,
+        _request_auth=_params.get('_request_auth'))
+
+  @validate_arguments
+  def read_config(self, **kwargs) -> object:  # noqa: E501
+    """Read Config  # noqa: E501
 
         Retrieve turbinia config.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -327,15 +312,15 @@ def read_config(self, **kwargs) -> object:  # noqa: E501
                  returns the request thread.
         :rtype: object
         """
-        kwargs['_return_http_data_only'] = True
-        if '_preload_content' in kwargs:
-            message = "Error! Please call the read_config_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
-            raise ValueError(message)
-        return self.read_config_with_http_info(**kwargs)  # noqa: E501
+    kwargs['_return_http_data_only'] = True
+    if '_preload_content' in kwargs:
+      message = "Error! Please call the read_config_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
+      raise ValueError(message)
+    return self.read_config_with_http_info(**kwargs)  # noqa: E501
 
-    @validate_arguments
-    def read_config_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
-        """Read Config  # noqa: E501
+  @validate_arguments
+  def read_config_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
+    """Read Config  # noqa: E501
 
         Retrieve turbinia config.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -369,70 +354,63 @@ def read_config_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
         :rtype: tuple(object, status_code(int), headers(HTTPHeaderDict))
         """
 
-        _params = locals()
-
-        _all_params = [
-        ]
-        _all_params.extend(
-            [
-                'async_req',
-                '_return_http_data_only',
-                '_preload_content',
-                '_request_timeout',
-                '_request_auth',
-                '_content_type',
-                '_headers'
-            ]
-        )
-
-        # validate the arguments
-        for _key, _val in _params['kwargs'].items():
-            if _key not in _all_params:
-                raise ApiTypeError(
-                    "Got an unexpected keyword argument '%s'"
-                    " to method read_config" % _key
-                )
-            _params[_key] = _val
-        del _params['kwargs']
-
-        _collection_formats = {}
-
-        # process the path parameters
-        _path_params = {}
-
-        # process the query parameters
-        _query_params = []
-        # process the header parameters
-        _header_params = dict(_params.get('_headers', {}))
-        # process the form parameters
-        _form_params = []
-        _files = {}
-        # process the body parameter
-        _body_params = None
-        # set the HTTP header `Accept`
-        _header_params['Accept'] = self.api_client.select_header_accept(
-            ['application/json'])  # noqa: E501
-
-        # authentication setting
-        _auth_settings = ['oAuth2']  # noqa: E501
-
-        _response_types_map = {
-            '200': "object",
-        }
-
-        return self.api_client.call_api(
-            '/api/config/', 'GET',
-            _path_params,
-            _query_params,
-            _header_params,
-            body=_body_params,
-            post_params=_form_params,
-            files=_files,
-            response_types_map=_response_types_map,
-            auth_settings=_auth_settings,
-            async_req=_params.get('async_req'),
-            _return_http_data_only=_params.get('_return_http_data_only'),  # noqa: E501
-            _preload_content=_params.get('_preload_content', True),
-            _request_timeout=_params.get('_request_timeout'),
-            collection_formats=_collection_formats,
-            _request_auth=_params.get('_request_auth'))
+    _params = locals()
+
+    _all_params = []
+    _all_params.extend([
+        'async_req', '_return_http_data_only', '_preload_content',
+        '_request_timeout', '_request_auth', '_content_type', '_headers'
+    ])
+
+    # validate the arguments
+    for _key, _val in _params['kwargs'].items():
+      if _key not in _all_params:
+        raise ApiTypeError(
+            "Got an unexpected keyword argument '%s'"
+            " to method read_config" % _key)
+      _params[_key] = _val
+    del _params['kwargs']
+
+    _collection_formats = {}
+
+    # process the path parameters
+    _path_params = {}
+
+    # process the query parameters
+    _query_params = []
+    # process the header parameters
+    _header_params = dict(_params.get('_headers', {}))
+    # process the form parameters
+    _form_params = []
+    _files = {}
+    # process the body parameter
+    _body_params = None
+    # set the HTTP header `Accept`
+    _header_params['Accept'] = self.api_client.select_header_accept(
+        ['application/json'])  # noqa: E501
+
+    # authentication setting
+    _auth_settings = ['oAuth2']  # noqa: E501
+
+    _response_types_map = {
+        '200': "object",
+    }
+
+    return self.api_client.call_api(
+        '/api/config/',
+        'GET',
+        _path_params,
+        _query_params,
+        _header_params,
+        body=_body_params,
+        post_params=_form_params,
+        files=_files,
+        response_types_map=_response_types_map,
+        auth_settings=_auth_settings,
+        async_req=_params.get('async_req'),
+        _return_http_data_only=_params.get(
+            '_return_http_data_only'),  # noqa: E501
+        _preload_content=_params.get('_preload_content', True),
+        _request_timeout=_params.get('_request_timeout'),
+        collection_formats=_collection_formats,
+        _request_auth=_params.get('_request_auth'))
diff --git a/turbinia/api/client/turbinia_api_lib/api/turbinia_jobs_api.py b/turbinia/api/client/turbinia_api_lib/api/turbinia_jobs_api.py
index 0d205f640..58078351b 100644
--- a/turbinia/api/client/turbinia_api_lib/api/turbinia_jobs_api.py
+++ b/turbinia/api/client/turbinia_api_lib/api/turbinia_jobs_api.py
@@ -11,7 +11,6 @@
     Do not edit the class manually.
 """  # noqa: E501
 
-
 import re  # noqa: F401
 import io
 import warnings
@@ -20,7 +19,6 @@
 
 from typing import Any
 
-
 from turbinia_api_lib.api_client import ApiClient
 from turbinia_api_lib.api_response import ApiResponse
 from turbinia_api_lib.exceptions import (  # noqa: F401
@@ -30,20 +28,20 @@
 
 
 class TurbiniaJobsApi:
-    """NOTE: This class is auto generated by OpenAPI Generator
+  """NOTE: This class is auto generated by OpenAPI Generator
     Ref: https://openapi-generator.tech
 
     Do not edit the class manually.
     """
 
-    def __init__(self, api_client=None) -> None:
-        if api_client is None:
-            api_client = ApiClient.get_default()
-        self.api_client = api_client
+  def __init__(self, api_client=None) -> None:
+    if api_client is None:
+      api_client = ApiClient.get_default()
+    self.api_client = api_client
 
-    @validate_arguments
-    def read_jobs(self, **kwargs) -> object:  # noqa: E501
-        """Read Jobs  # noqa: E501
+  @validate_arguments
+  def read_jobs(self, **kwargs) -> object:  # noqa: E501
+    """Read Jobs  # noqa: E501
 
         Return enabled jobs from the current Turbinia config.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -63,15 +61,15 @@ def read_jobs(self, **kwargs) -> object:  # noqa: E501
                  returns the request thread.
         :rtype: object
         """
-        kwargs['_return_http_data_only'] = True
-        if '_preload_content' in kwargs:
-            message = "Error! Please call the read_jobs_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
-            raise ValueError(message)
-        return self.read_jobs_with_http_info(**kwargs)  # noqa: E501
+    kwargs['_return_http_data_only'] = True
+    if '_preload_content' in kwargs:
+      message = "Error! Please call the read_jobs_with_http_info method with `_preload_content` instead and obtain raw data from ApiResponse.raw_data"  # noqa: E501
+      raise ValueError(message)
+    return self.read_jobs_with_http_info(**kwargs)  # noqa: E501
 
-    @validate_arguments
-    def read_jobs_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
-        """Read Jobs  # noqa: E501
+  @validate_arguments
+  def read_jobs_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
+    """Read Jobs  # noqa: E501
 
         Return enabled jobs from the current Turbinia config.  # noqa: E501
         This method makes a synchronous HTTP request by default. To make an
@@ -105,70 +103,63 @@ def read_jobs_with_http_info(self, **kwargs) -> ApiResponse:  # noqa: E501
         :rtype: tuple(object, status_code(int), headers(HTTPHeaderDict))
         """
 
-        _params = locals()
-
-        _all_params = [
-        ]
-        _all_params.extend(
-            [
-                'async_req',
-                '_return_http_data_only',
-                '_preload_content',
-                '_request_timeout',
-                '_request_auth',
-                '_content_type',
-                '_headers'
-            ]
-        )
-
-        # validate the arguments
-        for _key, _val in _params['kwargs'].items():
-            if _key not in _all_params:
-                raise ApiTypeError(
-                    "Got an unexpected keyword argument '%s'"
-                    " to method read_jobs" % _key
-                )
-            _params[_key] = _val
-        del _params['kwargs']
-
-        _collection_formats = {}
-
-        # process the path parameters
-        _path_params = {}
-
-        # process the query parameters
-        _query_params = []
-        # process the header parameters
-        _header_params = dict(_params.get('_headers', {}))
-        # process the form parameters
-        _form_params = []
-        _files = {}
-        # process the body parameter
-        _body_params = None
-        # set the HTTP header `Accept`
-        _header_params['Accept'] = self.api_client.select_header_accept(
-            ['application/json'])  # noqa: E501
-
-        # authentication setting
-        _auth_settings = ['oAuth2']  # noqa: E501
-
-        _response_types_map = {
-            '200': "object",
-        }
-
-        return self.api_client.call_api(
-            '/api/jobs/', 'GET',
-            _path_params,
-            _query_params,
-            _header_params,
-            body=_body_params,
-            post_params=_form_params,
-            files=_files,
-            response_types_map=_response_types_map,
-            auth_settings=_auth_settings,
-            async_req=_params.get('async_req'),
-            _return_http_data_only=_params.get('_return_http_data_only'),  # noqa: E501
-            _preload_content=_params.get('_preload_content', True),
-            _request_timeout=_params.get('_request_timeout'),
-            collection_formats=_collection_formats,
-            _request_auth=_params.get('_request_auth'))
+    _params = locals()
+
+    _all_params = []
+    _all_params.extend([
+        'async_req', '_return_http_data_only', '_preload_content',
+        '_request_timeout', '_request_auth', '_content_type', '_headers'
+    ])
+
+    # validate the arguments
+    for _key, _val in _params['kwargs'].items():
+      if _key not in _all_params:
+        raise ApiTypeError(
+            "Got an unexpected keyword argument '%s'"
+            " to method read_jobs" % _key)
+      _params[_key] = _val
+    del _params['kwargs']
+
+    _collection_formats = {}
+
+    # process the path parameters
+    _path_params = {}
+
+    # process the query parameters
+    _query_params = []
+    # process the header parameters
+    _header_params = dict(_params.get('_headers', {}))
+    # process the form parameters
+    _form_params = []
+    _files = {}
+    # process the body parameter
+    _body_params = None
+    # set the HTTP header `Accept`
+    _header_params['Accept'] = self.api_client.select_header_accept(
+        ['application/json'])  # noqa: E501
+
+    # authentication setting
+    _auth_settings = ['oAuth2']  # noqa: E501
+
+    _response_types_map = {
+        '200': "object",
+    }
+
+    return self.api_client.call_api(
+        '/api/jobs/',
+        'GET',
+        _path_params,
+        _query_params,
+        _header_params,
+        body=_body_params,
+        post_params=_form_params,
+        files=_files,
+        response_types_map=_response_types_map,
+        auth_settings=_auth_settings,
+        async_req=_params.get('async_req'),
+        _return_http_data_only=_params.get(
+            '_return_http_data_only'),  # noqa: E501
+        _preload_content=_params.get('_preload_content', True),
+        _request_timeout=_params.get('_request_timeout'),
+        collection_formats=_collection_formats,
+        _request_auth=_params.get('_request_auth'))
diff --git a/turbinia/api/client/turbinia_api_lib/api_client.py b/turbinia/api/client/turbinia_api_lib/api_client.py
index 14a0f5d93..c68d7a48c 100644
--- a/turbinia/api/client/turbinia_api_lib/api_client.py
+++ b/turbinia/api/client/turbinia_api_lib/api_client.py
@@ -11,7 +11,6 @@
     Do not edit the class manually.
 """  # noqa: E501
 
-
 import atexit
 import datetime
 from dateutil.parser import parse
@@ -32,7 +31,7 @@
 
 
 class ApiClient:
-    """Generic API client for OpenAPI client library builds.
+  """Generic API client for OpenAPI client library builds.
 
     OpenAPI generic API client. This client handles the client-
     server communication, and is invariant across implementations. Specifics of
@@ -49,78 +48,78 @@ class ApiClient:
         to the API. More threads means more concurrent API requests.
     """
 
-    PRIMITIVE_TYPES = (float, bool, bytes, str, int)
-    NATIVE_TYPES_MAPPING = {
-        'int': int,
-        'long': int, # TODO remove as only py3 is supported?
-        'float': float,
-        'str': str,
-        'bool': bool,
-        'date': datetime.date,
-        'datetime': datetime.datetime,
-        'object': object,
-    }
-    _pool = None
-
-    def __init__(self, configuration=None, header_name=None, header_value=None,
-                 cookie=None, pool_threads=1) -> None:
-        # use default configuration if none is provided
-        if configuration is None:
-            configuration = Configuration.get_default()
-        self.configuration = configuration
-        self.pool_threads = pool_threads
-
-        self.rest_client = rest.RESTClientObject(configuration)
-        self.default_headers = {}
-        if header_name is not None:
-            self.default_headers[header_name] = header_value
-        self.cookie = cookie
-        # Set default User-Agent.
-        self.user_agent = 'OpenAPI-Generator/1.0.0/python'
-        self.client_side_validation = configuration.client_side_validation
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.close()
-
-    def close(self):
-        if self._pool:
-            self._pool.close()
-            self._pool.join()
-            self._pool = None
-            if hasattr(atexit, 'unregister'):
-                atexit.unregister(self.close)
-
-    @property
-    def pool(self):
-        """Create thread pool on first request
+  PRIMITIVE_TYPES = (float, bool, bytes, str, int)
+  NATIVE_TYPES_MAPPING = {
+      'int': int,
+      'long': int,  # TODO remove as only py3 is supported?
+      'float': float,
+      'str': str,
+      'bool': bool,
+      'date': datetime.date,
+      'datetime': datetime.datetime,
+      'object': object,
+  }
+  _pool = None
+
+  def __init__(
+      self, configuration=None, header_name=None, header_value=None,
+      cookie=None, pool_threads=1) -> None:
+    # use default configuration if none is provided
+    if configuration is None:
+      configuration = Configuration.get_default()
+    self.configuration = configuration
+    self.pool_threads = pool_threads
+
+    self.rest_client = rest.RESTClientObject(configuration)
+    self.default_headers = {}
+    if header_name is not None:
+      self.default_headers[header_name] = header_value
+    self.cookie = cookie
+    # Set default User-Agent.
+    self.user_agent = 'OpenAPI-Generator/1.0.0/python'
+    self.client_side_validation = configuration.client_side_validation
+
+  def __enter__(self):
+    return self
+
+  def __exit__(self, exc_type, exc_value, traceback):
+    self.close()
+
+  def close(self):
+    if self._pool:
+      self._pool.close()
+      self._pool.join()
+      self._pool = None
+      if hasattr(atexit, 'unregister'):
+        atexit.unregister(self.close)
+
+  @property
+  def pool(self):
+    """Create thread pool on first request
          avoids instantiating unused threadpool for blocking clients.
         """
-        if self._pool is None:
-            atexit.register(self.close)
-            self._pool = ThreadPool(self.pool_threads)
-        return self._pool
-
-    @property
-    def user_agent(self):
-        """User agent for this API client"""
-        return self.default_headers['User-Agent']
+    if self._pool is None:
+      atexit.register(self.close)
+      self._pool = ThreadPool(self.pool_threads)
+    return self._pool
 
-    @user_agent.setter
-    def user_agent(self, value):
-        self.default_headers['User-Agent'] = value
+  @property
+  def user_agent(self):
+    """User agent for this API client"""
+    return self.default_headers['User-Agent']
 
-    def set_default_header(self, header_name, header_value):
-        self.default_headers[header_name] = header_value
+  @user_agent.setter
+  def user_agent(self, value):
+    self.default_headers['User-Agent'] = value
 
+  def set_default_header(self, header_name, header_value):
+    self.default_headers[header_name] = header_value
 
-    _default = None
+  _default = None
 
-    @classmethod
-    def get_default(cls):
-        """Return new instance of ApiClient.
+  @classmethod
+  def get_default(cls):
+    """Return new instance of ApiClient.
 
         This method returns newly created, based on default constructor,
         object of ApiClient class or returns a copy of default
@@ -128,136 +127,127 @@ def get_default(cls):
 
         :return: The ApiClient object.
         """
-        if cls._default is None:
-            cls._default = ApiClient()
-        return cls._default
+    if cls._default is None:
+      cls._default = ApiClient()
+    return cls._default
 
-    @classmethod
-    def set_default(cls, default):
-        """Set default instance of ApiClient.
+  @classmethod
+  def set_default(cls, default):
+    """Set default instance of ApiClient.
 
         It stores default ApiClient.
 
         :param default: object of ApiClient.
         """
-        cls._default = default
-
-    def __call_api(
-            self, resource_path, method, path_params=None,
-            query_params=None, header_params=None, body=None, post_params=None,
-            files=None, response_types_map=None, auth_settings=None,
-            _return_http_data_only=None, collection_formats=None,
-            _preload_content=True, _request_timeout=None, _host=None,
-            _request_auth=None):
-
-        config = self.configuration
-
-        # header parameters
-        header_params = header_params or {}
-        header_params.update(self.default_headers)
-        if self.cookie:
-            header_params['Cookie'] = self.cookie
-        if header_params:
-            header_params = self.sanitize_for_serialization(header_params)
-            header_params = dict(self.parameters_to_tuples(header_params,
-                                                           collection_formats))
-
-        # path parameters
-        if path_params:
-            path_params = self.sanitize_for_serialization(path_params)
-            path_params = self.parameters_to_tuples(path_params,
-                                                    collection_formats)
-            for k, v in path_params:
-                # specified safe chars, encode everything
-                resource_path = resource_path.replace(
-                    '{%s}' % k,
-                    quote(str(v), safe=config.safe_chars_for_path_param)
-                )
-
-        # post parameters
-        if post_params or files:
-            post_params = post_params if post_params else []
-            post_params = self.sanitize_for_serialization(post_params)
-            post_params = self.parameters_to_tuples(post_params,
-                                                    collection_formats)
-            post_params.extend(self.files_parameters(files))
-
-        # auth setting
-        self.update_params_for_auth(
-            header_params, query_params, auth_settings,
-            resource_path, method, body,
-            request_auth=_request_auth)
-
-        # body
-        if body:
-            body = self.sanitize_for_serialization(body)
-
-        # request url
-        if _host is None:
-            url = self.configuration.host + resource_path
-        else:
-            # use server/host defined in path or operation instead
-            url = _host + resource_path
-
-        # query parameters
-        if query_params:
-            query_params = self.sanitize_for_serialization(query_params)
-            url_query = self.parameters_to_url_query(query_params,
-                                                     collection_formats)
-            url += "?" + url_query
-
-        try:
-            # perform request and return response
-            response_data = self.request(
-                method, url,
-                query_params=query_params,
-                headers=header_params,
-                post_params=post_params, body=body,
-                _preload_content=_preload_content,
-                _request_timeout=_request_timeout)
-        except ApiException as e:
-            if e.body:
-                e.body = e.body.decode('utf-8')
-            raise e
-
-        self.last_response = response_data
-
-        return_data = None # assuming deserialization is not needed
-        # data needs deserialization or returns HTTP data (deserialized) only
-        if _preload_content or _return_http_data_only:
-          response_type = response_types_map.get(str(response_data.status), None)
-          if not response_type and isinstance(response_data.status, int) and 100 <= response_data.status <= 599:
-              # if not found, look for '1XX', '2XX', etc.
-              response_type = response_types_map.get(str(response_data.status)[0] + "XX", None)
-
-          if response_type == "bytearray":
-              response_data.data = response_data.data
-          else:
-              match = None
-              content_type = response_data.getheader('content-type')
-              if content_type is not None:
-                  match = re.search(r"charset=([a-zA-Z\-\d]+)[\s;]?", content_type)
-              encoding = match.group(1) if match else "utf-8"
-              response_data.data = response_data.data.decode(encoding)
-
-          # deserialize response data
-          if response_type == "bytearray":
-              return_data = response_data.data
-          elif response_type:
-              return_data = self.deserialize(response_data, response_type)
-          else:
-              return_data = None
-
-        if _return_http_data_only:
-            return return_data
-        else:
-            return ApiResponse(status_code = response_data.status,
-                           data = return_data,
-                           headers = response_data.getheaders(),
-                           raw_data = response_data.data)
-
-    def sanitize_for_serialization(self, obj):
-        """Builds a JSON POST object.
+    cls._default = default
+
+  def __call_api(
+      self, resource_path, method, path_params=None, query_params=None,
+      header_params=None, body=None, post_params=None, files=None,
+      response_types_map=None, auth_settings=None, _return_http_data_only=None,
+      collection_formats=None, _preload_content=True, _request_timeout=None,
+      _host=None, _request_auth=None):
+
+    config = self.configuration
+
+    # header parameters
+    header_params = header_params or {}
+    header_params.update(self.default_headers)
+    if self.cookie:
+      header_params['Cookie'] = self.cookie
+    if header_params:
+      header_params = self.sanitize_for_serialization(header_params)
+      header_params = dict(
+          self.parameters_to_tuples(header_params, collection_formats))
+
+    # path parameters
+    if path_params:
+      path_params = self.sanitize_for_serialization(path_params)
+      path_params = self.parameters_to_tuples(path_params, collection_formats)
+      for k, v in path_params:
+        # specified safe chars, encode everything
+        resource_path = resource_path.replace(
+            '{%s}' % k, quote(str(v), safe=config.safe_chars_for_path_param))
+
+    # post parameters
+    if post_params or files:
+      post_params = post_params if post_params else []
+      post_params = self.sanitize_for_serialization(post_params)
+      post_params = self.parameters_to_tuples(post_params, collection_formats)
+      post_params.extend(self.files_parameters(files))
+
+    # auth setting
+    self.update_params_for_auth(
+        header_params, query_params, auth_settings, resource_path, method, body,
+        request_auth=_request_auth)
+
+    # body
+    if body:
+      body = self.sanitize_for_serialization(body)
+
+    # request url
+    if _host is None:
+      url = self.configuration.host + resource_path
+    else:
+      # use server/host defined in path or operation instead
+      url = _host + resource_path
+
+    # query parameters
+    if query_params:
+      query_params = self.sanitize_for_serialization(query_params)
+      url_query = self.parameters_to_url_query(query_params, collection_formats)
+      url += "?" + url_query
+
+    try:
+      # perform request and return response
+      response_data = self.request(
+          method, url, query_params=query_params, headers=header_params,
+          post_params=post_params, body=body, _preload_content=_preload_content,
+          _request_timeout=_request_timeout)
+    except ApiException as e:
+      if e.body:
+        e.body = e.body.decode('utf-8')
+      raise e
+
+    self.last_response = response_data
+
+    return_data = None  # assuming deserialization is not needed
+    # data needs deserialization or returns HTTP data (deserialized) only
+    if _preload_content or _return_http_data_only:
+      response_type = response_types_map.get(str(response_data.status), None)
+      if not response_type and isinstance(
+          response_data.status, int) and 100 <= response_data.status <= 599:
+        # if not found, look for '1XX', '2XX', etc.
+        response_type = response_types_map.get(
+            str(response_data.status)[0] + "XX", None)
+
+      if response_type == "bytearray":
+        response_data.data = response_data.data
+      else:
+        match = None
+        content_type = response_data.getheader('content-type')
+        if content_type is not None:
+          match = re.search(r"charset=([a-zA-Z\-\d]+)[\s;]?", content_type)
+        encoding = match.group(1) if match else "utf-8"
+        response_data.data = response_data.data.decode(encoding)
+
+      # deserialize response data
+      if response_type == "bytearray":
+        return_data = response_data.data
+      elif response_type:
+        return_data = self.deserialize(response_data, response_type)
+      else:
+        return_data = None
+
+    if _return_http_data_only:
+      return return_data
+    else:
+      return ApiResponse(
+          status_code=response_data.status, data=return_data,
+          headers=response_data.getheaders(), raw_data=response_data.data)
+
+  def sanitize_for_serialization(self, obj):
+    """Builds a JSON POST object.
 
         If obj is None, return None.
         If obj is str, int, long, float, bool, return directly.
@@ -270,34 +260,34 @@ def sanitize_for_serialization(self, obj):
         :param obj: The data to serialize.
         :return: The serialized form of data.
         """
-        if obj is None:
-            return None
-        elif isinstance(obj, self.PRIMITIVE_TYPES):
-            return obj
-        elif isinstance(obj, list):
-            return [self.sanitize_for_serialization(sub_obj)
-                    for sub_obj in obj]
-        elif isinstance(obj, tuple):
-            return tuple(self.sanitize_for_serialization(sub_obj)
-                         for sub_obj in obj)
-        elif isinstance(obj, (datetime.datetime, datetime.date)):
-            return obj.isoformat()
-
-        if isinstance(obj, dict):
-            obj_dict = obj
-        else:
-            # Convert model obj to dict except
-            # attributes `openapi_types`, `attribute_map`
-            # and attributes which value is not None.
-            # Convert attribute name to json key in
-            # model definition for request.
-            obj_dict = obj.to_dict()
-
-        return {key: self.sanitize_for_serialization(val)
-                for key, val in obj_dict.items()}
+    if obj is None:
+      return None
+    elif isinstance(obj, self.PRIMITIVE_TYPES):
+      return obj
+    elif isinstance(obj, list):
+      return [self.sanitize_for_serialization(sub_obj) for sub_obj in obj]
+    elif isinstance(obj, tuple):
+      return tuple(self.sanitize_for_serialization(sub_obj) for sub_obj in obj)
+    elif isinstance(obj, (datetime.datetime, datetime.date)):
+      return obj.isoformat()
+
+    if isinstance(obj, dict):
+      obj_dict = obj
+    else:
+      # Convert model obj to dict except
+      # attributes `openapi_types`, `attribute_map`
+      # and attributes which value is not None.
+      # Convert attribute name to json key in
+      # model definition for request.
+      obj_dict = obj.to_dict()
+
+    return {
+        key: self.sanitize_for_serialization(val)
+        for key, val in obj_dict.items()
+    }
 
-    def deserialize(self, response, response_type):
-        """Deserializes response into an object.
+  def deserialize(self, response, response_type):
+    """Deserializes response into an object.
 
         :param response: RESTResponse object to be deserialized.
         :param response_type: class literal for
@@ -305,66 +295,64 @@ def deserialize(self, response, response_type):
 
         :return: deserialized object.
         """
-        # handle file downloading
-        # save response body into a tmp file and return the instance
-        if response_type == "file":
-            return self.__deserialize_file(response)
+    # handle file downloading
+    # save response body into a tmp file and return the instance
+    if response_type == "file":
+      return self.__deserialize_file(response)
 
-        # fetch data from response object
-        try:
-            data = json.loads(response.data)
-        except ValueError:
-            data = response.data
+    # fetch data from response object
+    try:
+      data = json.loads(response.data)
+    except ValueError:
+      data = response.data
 
-        return self.__deserialize(data, response_type)
+    return self.__deserialize(data, response_type)
 
-    def __deserialize(self, data, klass):
-        """Deserializes dict, list, str into an object.
+  def __deserialize(self, data, klass):
+    """Deserializes dict, list, str into an object.
 
         :param data: dict, list or str.
         :param klass: class literal, or string of class name.
 
         :return: object.
         """
-        if data is None:
-            return None
-
-        if isinstance(klass, str):
-            if klass.startswith('List['):
-                sub_kls = re.match(r'List\[(.*)]', klass).group(1)
-                return [self.__deserialize(sub_data, sub_kls)
-                        for sub_data in data]
-
-            if klass.startswith('Dict['):
-                sub_kls = re.match(r'Dict\[([^,]*), (.*)]', klass).group(2)
-                return {k: self.__deserialize(v, sub_kls)
-                        for k, v in data.items()}
-
-            # convert str to class
-            if klass in self.NATIVE_TYPES_MAPPING:
-                klass = self.NATIVE_TYPES_MAPPING[klass]
-            else:
-                klass = getattr(turbinia_api_lib.models, klass)
-
-        if klass in self.PRIMITIVE_TYPES:
-            return self.__deserialize_primitive(data, klass)
-        elif klass == object:
-            return self.__deserialize_object(data)
-        elif klass == datetime.date:
-            return self.__deserialize_date(data)
-        elif klass == datetime.datetime:
-            return self.__deserialize_datetime(data)
-        else:
-            return self.__deserialize_model(data, klass)
-
-    def call_api(self, resource_path, method,
-                 path_params=None, query_params=None, header_params=None,
-                 body=None, post_params=None, files=None,
-                 response_types_map=None, auth_settings=None,
-                 async_req=None, _return_http_data_only=None,
-                 collection_formats=None, _preload_content=True,
-                 _request_timeout=None, _host=None, _request_auth=None):
-        """Makes the HTTP request (synchronous) and returns deserialized data.
+    if data is None:
+      return None
+
+    if isinstance(klass, str):
+      if klass.startswith('List['):
+        sub_kls = re.match(r'List\[(.*)]', klass).group(1)
+        return [self.__deserialize(sub_data, sub_kls) for sub_data in data]
+
+      if klass.startswith('Dict['):
+        sub_kls = re.match(r'Dict\[([^,]*), (.*)]', klass).group(2)
+        return {k: self.__deserialize(v, sub_kls) for k, v in data.items()}
+
+      # convert str to class
+      if klass in self.NATIVE_TYPES_MAPPING:
+        klass = self.NATIVE_TYPES_MAPPING[klass]
+      else:
+        klass = getattr(turbinia_api_lib.models, klass)
+
+    if klass in self.PRIMITIVE_TYPES:
+      return self.__deserialize_primitive(data, klass)
+    elif klass == object:
+      return self.__deserialize_object(data)
+    elif klass == datetime.date:
+      return self.__deserialize_date(data)
+    elif klass == datetime.datetime:
+      return self.__deserialize_datetime(data)
+    else:
+      return self.__deserialize_model(data, klass)
+
+  def call_api(
+      self, resource_path, method, path_params=None, query_params=None,
+      header_params=None, body=None, post_params=None, files=None,
+      response_types_map=None, auth_settings=None, async_req=None,
+      _return_http_data_only=None, collection_formats=None,
+      _preload_content=True, _request_timeout=None, _host=None,
+      _request_auth=None):
+    """Makes the HTTP request (synchronous) and returns deserialized data.
 
         To make an async_req request, set the async_req parameter.
 
@@ -405,214 +393,196 @@ def call_api(self, resource_path, method,
             If parameter async_req is False or missing,
             then the method will return the response directly.
         """
-        args = (
-            resource_path,
-            method,
-            path_params,
-            query_params,
-            header_params,
-            body,
-            post_params,
-            files,
-            response_types_map,
-            auth_settings,
-            _return_http_data_only,
-            collection_formats,
-            _preload_content,
-            _request_timeout,
-            _host,
-            _request_auth,
-        )
-        if not async_req:
-            return self.__call_api(*args)
-
-        return self.pool.apply_async(self.__call_api, args)
-
-    def request(self, method, url, query_params=None, headers=None,
-                post_params=None, body=None, _preload_content=True,
-                _request_timeout=None):
-        """Makes the HTTP request using RESTClient."""
-        if method == "GET":
-            return self.rest_client.get_request(url,
-                                        query_params=query_params,
-                                        _preload_content=_preload_content,
-                                        _request_timeout=_request_timeout,
-                                        headers=headers)
-        elif method == "HEAD":
-            return self.rest_client.head_request(url,
-                                         query_params=query_params,
-                                         _preload_content=_preload_content,
-                                         _request_timeout=_request_timeout,
-                                         headers=headers)
-        elif method == "OPTIONS":
-            return self.rest_client.options_request(url,
-                                            query_params=query_params,
-                                            headers=headers,
-                                            _preload_content=_preload_content,
-                                            _request_timeout=_request_timeout)
-        elif method == "POST":
-            return self.rest_client.post_request(url,
-                                         query_params=query_params,
-                                         headers=headers,
-                                         post_params=post_params,
-                                         _preload_content=_preload_content,
-                                         _request_timeout=_request_timeout,
-                                         body=body)
-        elif method == "PUT":
-            return self.rest_client.put_request(url,
-                                        query_params=query_params,
-                                        headers=headers,
-                                        post_params=post_params,
-                                        _preload_content=_preload_content,
-                                        _request_timeout=_request_timeout,
-                                        body=body)
-        elif method == "PATCH":
-            return self.rest_client.patch_request(url,
-                                          query_params=query_params,
-                                          headers=headers,
-                                          post_params=post_params,
-                                          _preload_content=_preload_content,
-                                          _request_timeout=_request_timeout,
-                                          body=body)
-        elif method == "DELETE":
-            return self.rest_client.delete_request(url,
-                                           query_params=query_params,
-                                           headers=headers,
-                                           _preload_content=_preload_content,
-                                           _request_timeout=_request_timeout,
-                                           body=body)
-        else:
-            raise ApiValueError(
-                "http method must be `GET`, `HEAD`, `OPTIONS`,"
-                " `POST`, `PATCH`, `PUT` or `DELETE`."
-            )
-
-    def parameters_to_tuples(self, params, collection_formats):
-        """Get parameters as list of tuples, formatting collections.
+    args = (
+        resource_path,
+        method,
+        path_params,
+        query_params,
+        header_params,
+        body,
+        post_params,
+        files,
+        response_types_map,
+        auth_settings,
+        _return_http_data_only,
+        collection_formats,
+        _preload_content,
+        _request_timeout,
+        _host,
+        _request_auth,
+    )
+    if not async_req:
+      return self.__call_api(*args)
+
+    return self.pool.apply_async(self.__call_api, args)
+
+  def request(
+      self, method, url, query_params=None, headers=None, post_params=None,
+      body=None, _preload_content=True, _request_timeout=None):
+    """Makes the HTTP request using RESTClient."""
+    if method == "GET":
+      return self.rest_client.get_request(
+          url, query_params=query_params, _preload_content=_preload_content,
+          _request_timeout=_request_timeout, headers=headers)
+    elif method == "HEAD":
+      return self.rest_client.head_request(
+          url, query_params=query_params, _preload_content=_preload_content,
+          _request_timeout=_request_timeout, headers=headers)
+    elif method == "OPTIONS":
+      return self.rest_client.options_request(
+          url, query_params=query_params, headers=headers,
+          _preload_content=_preload_content, _request_timeout=_request_timeout)
+    elif method == "POST":
+      return self.rest_client.post_request(
+          url, query_params=query_params, headers=headers,
+          post_params=post_params, _preload_content=_preload_content,
+          _request_timeout=_request_timeout, body=body)
+    elif method == "PUT":
+      return self.rest_client.put_request(
+          url, query_params=query_params, headers=headers,
+          post_params=post_params, _preload_content=_preload_content,
+          _request_timeout=_request_timeout, body=body)
+    elif method == "PATCH":
+      return self.rest_client.patch_request(
+          url, query_params=query_params, headers=headers,
+          post_params=post_params, _preload_content=_preload_content,
+          _request_timeout=_request_timeout, body=body)
+    elif method == "DELETE":
+      return self.rest_client.delete_request(
+          url, query_params=query_params, headers=headers,
+          _preload_content=_preload_content, _request_timeout=_request_timeout,
+          body=body)
+    else:
+      raise ApiValueError(
+          "http method must be `GET`, `HEAD`, `OPTIONS`,"
+          " `POST`, `PATCH`, `PUT` or `DELETE`.")
+
+  def parameters_to_tuples(self, params, collection_formats):
+    """Get parameters as list of tuples, formatting collections.
 
         :param params: Parameters as dict or list of two-tuples
         :param dict collection_formats: Parameter collection formats
         :return: Parameters as list of tuples, collections formatted
         """
-        new_params = []
-        if collection_formats is None:
-            collection_formats = {}
-        for k, v in params.items() if isinstance(params, dict) else params:  # noqa: E501
-            if k in collection_formats:
-                collection_format = collection_formats[k]
-                if collection_format == 'multi':
-                    new_params.extend((k, value) for value in v)
-                else:
-                    if collection_format == 'ssv':
-                        delimiter = ' '
-                    elif collection_format == 'tsv':
-                        delimiter = '\t'
-                    elif collection_format == 'pipes':
-                        delimiter = '|'
-                    else:  # csv is the default
-                        delimiter = ','
-                    new_params.append(
-                        (k, delimiter.join(str(value) for value in v)))
-            else:
-                new_params.append((k, v))
-        return new_params
-
-    def parameters_to_url_query(self, params, collection_formats):
-        """Get parameters as list of tuples, formatting collections.
+    new_params = []
+    if collection_formats is None:
+      collection_formats = {}
+    for k, v in params.items() if isinstance(params,
+                                             dict) else params:  # noqa: E501
+      if k in collection_formats:
+        collection_format = collection_formats[k]
+        if collection_format == 'multi':
+          new_params.extend((k, value) for value in v)
+        else:
+          if collection_format == 'ssv':
+            delimiter = ' '
+          elif collection_format == 'tsv':
+            delimiter = '\t'
+          elif collection_format == 'pipes':
+            delimiter = '|'
+          else:  # csv is the default
+            delimiter = ','
+          new_params.append((k, delimiter.join(str(value) for value in v)))
+      else:
+        new_params.append((k, v))
+    return new_params
+
+  def parameters_to_url_query(self, params, collection_formats):
+    """Get parameters as list of tuples, formatting collections.
 
         :param params: Parameters as dict or list of two-tuples
         :param dict collection_formats: Parameter collection formats
         :return: URL query string (e.g. a=Hello%20World&b=123)
         """
-        new_params = []
-        if collection_formats is None:
-            collection_formats = {}
-        for k, v in params.items() if isinstance(params, dict) else params:  # noqa: E501
-            if isinstance(v, bool):
-                v = str(v).lower()
-            if isinstance(v, (int, float)):
-                v = str(v)
-            if isinstance(v, dict):
-                v = json.dumps(v)
-
-            if k in collection_formats:
-                collection_format = collection_formats[k]
-                if collection_format == 'multi':
-                    new_params.extend((k, str(value)) for value in v)
-                else:
-                    if collection_format == 'ssv':
-                        delimiter = ' '
-                    elif collection_format == 'tsv':
-                        delimiter = '\t'
-                    elif collection_format == 'pipes':
-                        delimiter = '|'
-                    else:  # csv is the default
-                        delimiter = ','
-                    new_params.append(
-                        (k, delimiter.join(quote(str(value)) for value in v)))
-            else:
-                new_params.append((k, quote(str(v))))
-
-        return "&".join(["=".join(map(str, item)) for item in new_params])
-
-    def files_parameters(self, files=None):
-        """Builds form parameters.
+    new_params = []
+    if collection_formats is None:
+      collection_formats = {}
+    for k, v in params.items() if isinstance(params,
+                                             dict) else params:  # noqa: E501
+      if isinstance(v, bool):
+        v = str(v).lower()
+      if isinstance(v, (int, float)):
+        v = str(v)
+      if isinstance(v, dict):
+        v = json.dumps(v)
+
+      if k in collection_formats:
+        collection_format = collection_formats[k]
+        if collection_format == 'multi':
+          new_params.extend((k, str(value)) for value in v)
+        else:
+          if collection_format == 'ssv':
+            delimiter = ' '
+          elif collection_format == 'tsv':
+            delimiter = '\t'
+          elif collection_format == 'pipes':
+            delimiter = '|'
+          else:  # csv is the default
+            delimiter = ','
+          new_params.append(
+              (k, delimiter.join(quote(str(value)) for value in v)))
+      else:
+        new_params.append((k, quote(str(v))))
+
+    return "&".join(["=".join(map(str, item)) for item in new_params])
+
+  def files_parameters(self, files=None):
+    """Builds form parameters.
 
         :param files: File parameters.
         :return: Form parameters with files.
         """
-        params = []
-
-        if files:
-            for k, v in files.items():
-                if not v:
-                    continue
-                file_names = v if type(v) is list else [v]
-                for n in file_names:
-                    with open(n, 'rb') as f:
-                        filename = os.path.basename(f.name)
-                        filedata = f.read()
-                        mimetype = (mimetypes.guess_type(filename)[0] or
-                                    'application/octet-stream')
-                        params.append(
-                            tuple([k, tuple([filename, filedata, mimetype])]))
-
-        return params
-
-    def select_header_accept(self, accepts):
-        """Returns `Accept` based on an array of accepts provided.
+    params = []
+
+    if files:
+      for k, v in files.items():
+        if not v:
+          continue
+        file_names = v if type(v) is list else [v]
+        for n in file_names:
+          with open(n, 'rb') as f:
+            filename = os.path.basename(f.name)
+            filedata = f.read()
+            mimetype = (
+                mimetypes.guess_type(filename)[0] or 'application/octet-stream')
+            params.append(tuple([k, tuple([filename, filedata, mimetype])]))
+
+    return params
+
+  def select_header_accept(self, accepts):
+    """Returns `Accept` based on an array of accepts provided.
 
         :param accepts: List of headers.
         :return: Accept (e.g. application/json).
         """
-        if not accepts:
-            return
+    if not accepts:
+      return
 
-        for accept in accepts:
-            if re.search('json', accept, re.IGNORECASE):
-                return accept
+    for accept in accepts:
+      if re.search('json', accept, re.IGNORECASE):
+        return accept
 
-        return accepts[0]
+    return accepts[0]
 
-    def select_header_content_type(self, content_types):
-        """Returns `Content-Type` based on an array of content_types provided.
+  def select_header_content_type(self, content_types):
+    """Returns `Content-Type` based on an array of content_types provided.
 
         :param content_types: List of content-types.
         :return: Content-Type (e.g. application/json).
         """
-        if not content_types:
-            return None
+    if not content_types:
+      return None
 
-        for content_type in content_types:
-            if re.search('json', content_type, re.IGNORECASE):
-                return content_type
+    for content_type in content_types:
+      if re.search('json', content_type, re.IGNORECASE):
+        return content_type
 
-        return content_types[0]
+    return content_types[0]
 
-    def update_params_for_auth(self, headers, queries, auth_settings,
-                               resource_path, method, body,
-                               request_auth=None):
-        """Updates header and query params based on authentication setting.
+  def update_params_for_auth(
+      self, headers, queries, auth_settings, resource_path, method, body,
+      request_auth=None):
+    """Updates header and query params based on authentication setting.
 
         :param headers: Header parameters dict to be updated.
         :param queries: Query parameters tuple list to be updated.
@@ -624,26 +594,23 @@ def update_params_for_auth(self, headers, queries, auth_settings,
         :param request_auth: if set, the provided settings will
                              override the token in the configuration.
         """
-        if not auth_settings:
-            return
-
-        if request_auth:
-            self._apply_auth_params(headers, queries,
-                                    resource_path, method, body,
-                                    request_auth)
-            return
-
-        for auth in auth_settings:
-            auth_setting = self.configuration.auth_settings().get(auth)
-            if auth_setting:
-                self._apply_auth_params(headers, queries,
-                                        resource_path, method, body,
-                                        auth_setting)
-
-    def _apply_auth_params(self, headers, queries,
-                           resource_path, method, body,
-                           auth_setting):
-        """Updates the request parameters based on a single auth_setting
+    if not auth_settings:
+      return
+
+    if request_auth:
+      self._apply_auth_params(
+          headers, queries, resource_path, method, body, request_auth)
+      return
+
+    for auth in auth_settings:
+      auth_setting = self.configuration.auth_settings().get(auth)
+      if auth_setting:
+        self._apply_auth_params(
+            headers, queries, resource_path, method, body, auth_setting)
+
+  def _apply_auth_params(
+      self, headers, queries, resource_path, method, body, auth_setting):
+    """Updates the request parameters based on a single auth_setting
 
         :param headers: Header parameters dict to be updated.
         :param queries: Query parameters tuple list to be updated.
@@ -653,20 +620,18 @@ def _apply_auth_params(self, headers, queries,
         The object type is the return value of sanitize_for_serialization().
         :param auth_setting: auth settings for the endpoint
         """
-        if auth_setting['in'] == 'cookie':
-            headers['Cookie'] = auth_setting['value']
-        elif auth_setting['in'] == 'header':
-            if auth_setting['type'] != 'http-signature':
-                headers[auth_setting['key']] = auth_setting['value']
-        elif auth_setting['in'] == 'query':
-            queries.append((auth_setting['key'], auth_setting['value']))
-        else:
-            raise ApiValueError(
-                'Authentication token must be in `query` or `header`'
-            )
-
-    def __deserialize_file(self, response):
-        """Deserializes body to file
+    if auth_setting['in'] == 'cookie':
+      headers['Cookie'] = auth_setting['value']
+    elif auth_setting['in'] == 'header':
+      if auth_setting['type'] != 'http-signature':
+        headers[auth_setting['key']] = auth_setting['value']
+    elif auth_setting['in'] == 'query':
+      queries.append((auth_setting['key'], auth_setting['value']))
+    else:
+      raise ApiValueError('Authentication token must be in `query` or `header`')
+
+  def __deserialize_file(self, response):
+    """Deserializes body to file
 
         Saves response body into a file in a temporary folder,
         using the filename from the `Content-Disposition` header if provided.
@@ -674,86 +639,81 @@ def __deserialize_file(self, response):
         :param response:  RESTResponse.
         :return: file path.
         """
-        fd, path = tempfile.mkstemp(dir=self.configuration.temp_folder_path)
-        os.close(fd)
-        os.remove(path)
+    fd, path = tempfile.mkstemp(dir=self.configuration.temp_folder_path)
+    os.close(fd)
+    os.remove(path)
 
-        content_disposition = response.getheader("Content-Disposition")
-        if content_disposition:
-            filename = re.search(r'filename=[\'"]?([^\'"\s]+)[\'"]?',
-                                 content_disposition).group(1)
-            path = os.path.join(os.path.dirname(path), filename)
+    content_disposition = response.getheader("Content-Disposition")
+    if content_disposition:
+      filename = re.search(
+          r'filename=[\'"]?([^\'"\s]+)[\'"]?', content_disposition).group(1)
+      path = os.path.join(os.path.dirname(path), filename)
 
-        with open(path, "wb") as f:
-            f.write(response.data)
+    with open(path, "wb") as f:
+      f.write(response.data)
 
-        return path
+    return path
 
-    def __deserialize_primitive(self, data, klass):
-        """Deserializes string to primitive type.
+  def __deserialize_primitive(self, data, klass):
+    """Deserializes string to primitive type.
 
         :param data: str.
         :param klass: class literal.
 
         :return: int, long, float, str, bool.
         """
-        try:
-            return klass(data)
-        except UnicodeEncodeError:
-            return str(data)
-        except TypeError:
-            return data
+    try:
+      return klass(data)
+    except UnicodeEncodeError:
+      return str(data)
+    except TypeError:
+      return data
 
-    def __deserialize_object(self, value):
-        """Return an original value.
+  def __deserialize_object(self, value):
+    """Return an original value.
 
         :return: object.
         """
-        return value
+    return value
 
-    def __deserialize_date(self, string):
-        """Deserializes string to date.
+  def __deserialize_date(self, string):
+    """Deserializes string to date.
 
         :param string: str.
         :return: date.
         """
-        try:
-            return parse(string).date()
-        except ImportError:
-            return string
-        except ValueError:
-            raise rest.ApiException(
-                status=0,
-                reason="Failed to parse `{0}` as date object".format(string)
-            )
-
-    def __deserialize_datetime(self, string):
-        """Deserializes string to datetime.
+    try:
+      return parse(string).date()
+    except ImportError:
+      return string
+    except ValueError:
+      raise rest.ApiException(
+          status=0,
+          reason="Failed to parse `{0}` as date object".format(string))
+
+  def __deserialize_datetime(self, string):
+    """Deserializes string to datetime.
 
         The string should be in iso8601 datetime format.
 
         :param string: str.
         :return: datetime.
         """
-        try:
-            return parse(string)
-        except ImportError:
-            return string
-        except ValueError:
-            raise rest.ApiException(
-                status=0,
-                reason=(
-                    "Failed to parse `{0}` as datetime object"
-                    .format(string)
-                )
-            )
-
-    def __deserialize_model(self, data, klass):
-        """Deserializes list or dict to model.
+    try:
+      return parse(string)
+    except ImportError:
+      return string
+    except ValueError:
+      raise rest.ApiException(
+          status=0,
+          reason=("Failed to parse `{0}` as datetime object".format(string)))
+
+  def __deserialize_model(self, data, klass):
+    """Deserializes list or dict to model.
 
         :param data: dict, list.
         :param klass: class literal.
         :return: model object.
         """
 
-        return klass.from_dict(data)
+    return klass.from_dict(data)
diff --git a/turbinia/api/client/turbinia_api_lib/models/validation_error.py b/turbinia/api/client/turbinia_api_lib/models/validation_error.py
index 9c513201e..f7cf8158c 100644
--- a/turbinia/api/client/turbinia_api_lib/models/validation_error.py
+++ b/turbinia/api/client/turbinia_api_lib/models/validation_error.py
@@ -11,73 +11,70 @@
     Do not edit the class manually.
 """  # noqa: E501
 
-
 from __future__ import annotations
 import pprint
 import re  # noqa: F401
 import json
 
-
 from typing import List
 from pydantic import BaseModel, Field, StrictStr, conlist
 from turbinia_api_lib.models.location_inner import LocationInner
 
+
 class ValidationError(BaseModel):
-    """
+  """
     ValidationError
     """
-    loc: conlist(LocationInner) = Field(...)
-    msg: StrictStr = Field(...)
-    type: StrictStr = Field(...)
-    __properties = ["loc", "msg", "type"]
-
-    class Config:
-        """Pydantic configuration"""
-        allow_population_by_field_name = True
-        validate_assignment = True
-
-    def to_str(self) -> str:
-        """Returns the string representation of the model using alias"""
-        return pprint.pformat(self.dict(by_alias=True))
-
-    def to_json(self) -> str:
-        """Returns the JSON representation of the model using alias"""
-        return json.dumps(self.to_dict())
-
-    @classmethod
-    def from_json(cls, json_str: str) -> ValidationError:
-        """Create an instance of ValidationError from a JSON string"""
-        return cls.from_dict(json.loads(json_str))
-
-    def to_dict(self):
-        """Returns the dictionary representation of the model using alias"""
-        _dict = self.dict(by_alias=True,
-                          exclude={
-                          },
-                          exclude_none=True)
-        # override the default output from pydantic by calling `to_dict()` of each item in loc (list)
-        _items = []
-        if self.loc:
-            for _item in self.loc:
-                if _item:
-                    _items.append(_item.to_dict())
-            _dict['loc'] = _items
-        return _dict
-
-    @classmethod
-    def from_dict(cls, obj: dict) -> ValidationError:
-        """Create an instance of ValidationError from a dict"""
-        if obj is None:
-            return None
-
-        if not isinstance(obj, dict):
-            return ValidationError.parse_obj(obj)
-
-        _obj = ValidationError.parse_obj({
-            "loc": [LocationInner.from_dict(_item) for _item in obj.get("loc")] if obj.get("loc") is not None else None,
-            "msg": obj.get("msg"),
-            "type": obj.get("type")
-        })
-        return _obj
-
-
+  loc: conlist(LocationInner) = Field(...)
+  msg: StrictStr = Field(...)
+  type: StrictStr = Field(...)
+  __properties = ["loc", "msg", "type"]
+
+  class Config:
+    """Pydantic configuration"""
+    allow_population_by_field_name = True
+    validate_assignment = True
+
+  def to_str(self) -> str:
+    """Returns the string representation of the model using alias"""
+    return pprint.pformat(self.dict(by_alias=True))
+
+  def to_json(self) -> str:
+    """Returns the JSON representation of the model using alias"""
+    return json.dumps(self.to_dict())
+
+  @classmethod
+  def from_json(cls, json_str: str) -> ValidationError:
+    """Create an instance of ValidationError from a JSON string"""
+    return cls.from_dict(json.loads(json_str))
+
+  def to_dict(self):
+    """Returns the dictionary representation of the model using alias"""
+    _dict = self.dict(by_alias=True, exclude={}, exclude_none=True)
+    # override the default output from pydantic by calling `to_dict()` of each item in loc (list)
+    _items = []
+    if self.loc:
+      for _item in self.loc:
+        if _item:
+          _items.append(_item.to_dict())
+      _dict['loc'] = _items
+    return _dict
+
+  @classmethod
+  def from_dict(cls, obj: dict) -> ValidationError:
+    """Create an instance of ValidationError from a dict"""
+    if obj is None:
+      return None
+
+    if not isinstance(obj, dict):
+      return ValidationError.parse_obj(obj)
+
+    _obj = ValidationError.parse_obj({
+        "loc": [LocationInner.from_dict(_item) for _item in obj.get("loc")]
+               if obj.get("loc") is not None else None,
+        "msg":
+            obj.get("msg"),
+        "type":
+            obj.get("type")
+    })
+    return _obj
diff --git a/turbinia/api/models/request_status.py b/turbinia/api/models/request_status.py
index 7aab0d9d1..9daf1dbea 100644
--- a/turbinia/api/models/request_status.py
+++ b/turbinia/api/models/request_status.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 """Turbinia API server Request models."""
 
-import datetime
 import logging
 
 from typing import Optional, List, Dict
@@ -22,24 +21,24 @@
 from turbinia import state_manager
 from turbinia import config as turbinia_config
 
-log = logging.getLogger('turbinia:api_server:routes:request')
+log = logging.getLogger(__name__)
 
 
 class RequestStatus(BaseModel):
   """Represents a Turbinia request status object."""
-  request_id: str = None
-  evidence_name: str = None
   evidence_id: str = None
-  tasks: List[Dict] = []
+  evidence_name: str = None
+  failed_tasks: int = 0
+  last_task_update_time: str = None
+  queued_tasks: int = 0
   reason: str = None
+  request_id: str = None
   requester: str = None
-  last_task_update_time: str = None
+  running_tasks: int = 0
   status: str = None
-  task_count: int = 0
   successful_tasks: int = 0
-  running_tasks: int = 0
-  failed_tasks: int = 0
-  queued_tasks: int = 0
+  task_count: int = 0
+  tasks: List[Dict] = []
 
   def get_request_data(
       self, request_id: str, tasks: Optional[List[Dict]] = None,
@@ -54,111 +53,31 @@ def get_request_data(
     Returns:
       bool: True if the request has at least one task associated with it.
     """
-    if not tasks:
-      _state_manager = state_manager.get_state_manager()
-      tasks = _state_manager.get_task_data(
-          instance=turbinia_config.INSTANCE_ID, request_id=request_id)
+    state_client = state_manager.get_state_manager()
+    self.request_id = request_id
 
     if not summary:
-      for task in tasks:
-        current_request_id = task.get('request_id')
-        if current_request_id == request_id:
-          self.tasks.append(task)
-
-    # Tries to get the evidence_name from the -l argument of the first task,
-    # which is the argument passed in the terminal to determine the evidence.
-    # If successful, sets the initial_start_time to None, if not, to the
-    # current time, so that it can be used later to determine the first started
-    # task and then get the evidence_name, as later tasks may have a different
-    # evidence name. There is a small chance of the first task having a
-    # different evidence_name, so getting it from arguments is preferred when
-    # they exist.
-    # todo(igormr): Save request information in redis to get the evidence_name
-    name_from_args = False
-    if tasks:
-      if tasks[0].get('all_args'):
-        arguments = tasks[0].get('all_args', 0).split()
-        for i in range(len(arguments) - 1):
-          if arguments[i] == '-l':
-            self.evidence_name = arguments[i + 1]
-            name_from_args = True
-            break
-
-    initial_start_time = datetime.datetime.now().strftime(
-        turbinia_config.DATETIME_FORMAT)
-    for task in tasks:
-      self.request_id = task.get('request_id')
-      self.requester = task.get('requester')
-      self.reason = task.get('reason')
-      self.task_count = len(tasks)
-      task_status = task.get('status')
-      # Gets the evidence_name from the first started task.
-      if name_from_args and task.get('evidence_name') == self.evidence_name:
-        self.evidence_id = task.get('evidence_id')
-      elif not name_from_args and task.get('start_time') and task.get(
-          'start_time') < initial_start_time:
-        initial_start_time = task.get('start_time')
-        self.evidence_name = task.get('evidence_name')
-        self.evidence_id = task.get('evidence_id')
-      if isinstance(task.get('last_update'), datetime.datetime):
-        task_last_update = datetime.datetime.timestamp(task.get('last_update'))
-      else:
-        task_last_update = task.get('last_update')
-      if not self.last_task_update_time:
-        self.last_task_update_time = task_last_update
-      else:
-        self.last_task_update_time = max(
-            self.last_task_update_time, task_last_update)
-
-      if task.get('successful'):
-        self.successful_tasks += 1
-      # 'successful' could be None or False, which means different things.
-      # If False, the task has failed, If None, could be queued or running.
-      elif task.get('successful') is False:
-        self.failed_tasks += 1
-      elif task.get('successful') is None:
-        if task_status:
-          if 'running' in task_status:
-            self.running_tasks += 1
-        else:
-          # 'successful' is None and 'status' is None
-          self.queued_tasks += 1
-      if isinstance(task['last_update'], datetime.datetime):
-        task['last_update'] = task['last_update'].strftime(
-            turbinia_config.DATETIME_FORMAT)
-
-    if self.last_task_update_time:
-      if isinstance(self.last_task_update_time, float):
-        self.last_task_update_time = datetime.datetime.fromtimestamp(
-            self.last_task_update_time).strftime(
-                turbinia_config.DATETIME_FORMAT)
-
-    completed_tasks = self.successful_tasks + self.failed_tasks
-
-    if completed_tasks == self.task_count and self.failed_tasks > 0:
-      self.status = 'completed_with_errors'
-    elif self.failed_tasks == self.task_count:
-      self.status = 'failed'
-    elif self.successful_tasks == self.task_count:
-      self.status = 'successful'
-    else:
-      # TODO(leaniz): Add a 'pending' state to tasks for cases 2 and 3.
-      # ref: https://github.com/google/turbinia/issues/1239
-      #
-      # A 'running' status for a request covers multiple cases:
-      #  1) One or more tasks are still in a running status.
-      #  2) Zero tasks are running, zero or more tasks are queued
-      #    and none have failed/succeeded.
-      #  (e.g. all tasks scheduled on the Turbinia server and none picked
-      #    up by any worker yet.)
-      #  3) Zero tasks are running, one or more tasks are queued
-      #    and some have failed/succeeded.
-      #  (e.g. some tasks have completed, others are scheduled on the
-      #    Turbinia server but not picked up by a worker yet.)
-      #
-      # Note that this method is concerned with a Turbiania request's status
-      # which is different than the status of an individual task.
-      self.status = 'running'
+      self.tasks = tasks if tasks else state_client.get_task_data(
+          instance=turbinia_config.INSTANCE_ID, request_id=request_id)
+
+    # Gets the information from the request if it is stored in Redis
+    request_key = state_client.redis_client.build_key_name(
+        'request', request_id)
+    if state_client.redis_client.key_exists(request_key):
+      saved_request = state_client.get_request_data(request_id)
+      self.evidence_name = saved_request.get('original_evidence').get('name')
+      self.evidence_id = saved_request.get('original_evidence').get('id')
+      self.requester = saved_request.get('requester')
+      self.reason = saved_request.get('reason')
+      self.status = saved_request.get('status')
+      self.last_task_update_time = saved_request.get('last_update')
+      self.successful_tasks = len(saved_request.get('successful_tasks', []))
+      self.failed_tasks = len(saved_request.get('failed_tasks', []))
+      self.queued_tasks = len(saved_request.get('queued_tasks', []))
+      self.running_tasks = len(saved_request.get('running_tasks', []))
+      task_ids = saved_request.get('task_ids', [])
+      self.task_count = len(task_ids)
+      self.status = saved_request.get('status', 'pending')
 
     return bool(self.tasks)
 
@@ -169,22 +88,12 @@ class RequestsSummary(BaseModel):
 
   def get_requests_summary(self) -> bool:
     """Generates a status summary for each Turbinia request."""
-    _state_manager = state_manager.get_state_manager()
-    instance_id = turbinia_config.INSTANCE_ID
-    tasks = _state_manager.get_task_data(instance=instance_id)
-    request_ids = set()
-
-    for task in tasks:
-      request_id = task.get('request_id')
-      if not request_id in request_ids:
-        request_ids.add(request_id)
-
-    for request_id in request_ids:
-      filtered_tasks = [
-          task for task in tasks if task.get('request_id') == request_id
-      ]
+    state_client = state_manager.get_state_manager()
+
+    for request_key in state_client.redis_client.iterate_keys('Request'):
+      request_id = request_key.split(':')[1]
       request_status = RequestStatus()
-      request_status.get_request_data(request_id, filtered_tasks, summary=True)
+      request_status.get_request_data(request_id, summary=True)
       self.requests_status.append(request_status)
 
     return bool(self.requests_status)
diff --git a/turbinia/api/models/tasks_statistics.py b/turbinia/api/models/tasks_statistics.py
index 8c7a82231..326ed48d8 100644
--- a/turbinia/api/models/tasks_statistics.py
+++ b/turbinia/api/models/tasks_statistics.py
@@ -23,7 +23,7 @@
 from turbinia import state_manager
 from turbinia import config as turbinia_config
 
-log = logging.getLogger('turbinia:api_server:models:tasks_statistics')
+log = logging.getLogger(__name__)
 
 
 class SingleTurbiniaStats:
@@ -87,7 +87,7 @@ def format_stats(self):
     """
     return (
         f'{self.description}: Count: {self.count}, Min: {self.min}, '
-        f'Mean: {self.mean}, Max: {self.maxs}')
+        f'Mean: {self.mean}, Max: {self.max}')
 
   def format_stats_csv(self):
     """Formats statistics data into CSV output.
diff --git a/turbinia/api/models/workers_status.py b/turbinia/api/models/workers_status.py
index 0009f5bcc..6476d4e0e 100644
--- a/turbinia/api/models/workers_status.py
+++ b/turbinia/api/models/workers_status.py
@@ -20,16 +20,16 @@
 from datetime import datetime
 from datetime import timedelta
 from operator import itemgetter
-from pydantic import BaseModel
+from pydantic import BaseModel, Extra
 from typing import ClassVar
 
 from turbinia import state_manager
 from turbinia import config as turbinia_config
 
-log = logging.getLogger('turbinia:api_server:models:workers_status')
+log = logging.getLogger(__name__)
 
 
-class WorkersInfo(BaseModel):
+class WorkersInfo(BaseModel, extra=Extra.allow):
   """Information about Workers."""
 
   workers_dict: ClassVar[dict] = {}
diff --git a/turbinia/api/routes/config.py b/turbinia/api/routes/config.py
index 623b3a4ff..e0579525f 100644
--- a/turbinia/api/routes/config.py
+++ b/turbinia/api/routes/config.py
@@ -32,7 +32,7 @@
 
 @router.get('/')
 async def read_config(request: Request):
-  """Retrieve turbinia config."""
+  """Retrieves Turbinia configuration."""
   try:
     current_config = turbinia_config.toDict()
     if current_config:
diff --git a/turbinia/api/routes/evidence.py b/turbinia/api/routes/evidence.py
index ea1a3908d..4ea0398e9 100644
--- a/turbinia/api/routes/evidence.py
+++ b/turbinia/api/routes/evidence.py
@@ -24,7 +24,6 @@
 from fastapi.responses import JSONResponse
 from typing import List, Annotated
 
-from turbinia.api.schemas import request_options
 from turbinia import evidence
 from turbinia import config as turbinia_config
 from turbinia import state_manager
@@ -97,9 +96,10 @@ async def upload_file(
         sha_hash.update(chunk)
       size += len(chunk)
       if size >= turbinia_config.API_MAX_UPLOAD_SIZE:
+        msg_size = turbinia_config.API_MAX_UPLOAD_SIZE / (1024**3)
         error_message = (
             f'Unable to upload file {file.filename} greater',
-            f'than {turbinia_config.API_MAX_UPLOAD_SIZE / (1024 ** 3)} GB')
+            f'than {msg_size} GB')
         log.error(error_message)
         raise TurbiniaException(error_message)
     file_info = {
@@ -217,7 +217,7 @@ async def get_evidence_by_id(request: Request, evidence_id):
     return JSONResponse(content=stored_evidence, status_code=200)
   raise HTTPException(
       status_code=404,
-      detail=f'UUID {evidence_id} not found or it had no associated evidences.')
+      detail=f'UUID {evidence_id} not found or it had no associated evidence.')
 
 
 @router.post('/upload')
@@ -251,6 +251,7 @@ async def upload_evidence(
           file_info.get('hash')):
         warning_message = (
             f'File {file.filename} was uploaded before, check {evidence_key}')
+
     if warning_message:
       evidences.append(warning_message)
       log.error(warning_message)
diff --git a/turbinia/api/routes/jobs.py b/turbinia/api/routes/jobs.py
index fbc08c961..5a2210f64 100644
--- a/turbinia/api/routes/jobs.py
+++ b/turbinia/api/routes/jobs.py
@@ -31,11 +31,11 @@
 
 @router.get('/')
 async def read_jobs(request: Request):
-  """Return enabled jobs from the current Turbinia config."""
+  """Return enabled jobs from the current Turbiniaconfig."""
   try:
     _jobs_manager: jobs_manager.JobsManager = jobs_manager.JobsManager()
     registered_jobs: set = set(_jobs_manager.GetJobNames())
-    disabled_jobs: set = set(turbinia_config.CONFIG.DISABLED_JOBS)
+    disabled_jobs: set = set(turbinia_config.DISABLED_JOBS)
     enabled_jobs: set = registered_jobs.difference(disabled_jobs)
 
     if not registered_jobs:
diff --git a/turbinia/api/routes/request.py b/turbinia/api/routes/request.py
index b18267369..edc0a4369 100644
--- a/turbinia/api/routes/request.py
+++ b/turbinia/api/routes/request.py
@@ -17,7 +17,6 @@
 import logging
 import uuid
 import json
-from collections import OrderedDict
 
 from fastapi import HTTPException, APIRouter
 from fastapi.responses import JSONResponse
@@ -48,10 +47,9 @@ async def get_requests_summary(request: Request):
     if not requests_summary.get_requests_summary():
       return JSONResponse(
           content={'detail': 'Request summary is empty'}, status_code=200)
-    response_json = requests_summary.json()
-    response = OrderedDict(sorted(json.loads(response_json).items()))
+    response_json = json.loads(requests_summary.json())
     return JSONResponse(
-        status_code=200, content=response, media_type='application/json')
+        status_code=200, content=response_json, media_type='application/json')
   except (json.JSONDecodeError, TypeError, ValueError, AttributeError,
           ValidationError) as exception:
     log.error(
@@ -78,13 +76,13 @@ async def get_request_status(request: Request, request_id: str):
       raise HTTPException(
           status_code=404,
           detail='Request ID not found or the request had no associated tasks.')
-    response_json = request_out.json()
-    response = OrderedDict(sorted(json.loads(response_json).items()))
+    response_json = json.loads(request_out.json())
     return JSONResponse(
-        status_code=200, content=response, media_type='application/json')
+        status_code=200, content=response_json, media_type='application/json')
   except (json.JSONDecodeError, TypeError, ValueError, AttributeError,
-          ValidationError) as exception:
-    log.error(f'Error retrieving request information: {exception!s}')
+          TurbiniaException) as exception:
+    log.error(
+        f'Error retrieving request information: {exception!s}', exc_info=True)
     raise HTTPException(
         status_code=500,
         detail='Error retrieving request information') from exception
diff --git a/turbinia/api/routes/task.py b/turbinia/api/routes/task.py
index 2dcefe735..c6bbbd2e5 100644
--- a/turbinia/api/routes/task.py
+++ b/turbinia/api/routes/task.py
@@ -30,7 +30,7 @@
 from turbinia.api.models import workers_status
 from turbinia.api.models import tasks_statistics
 
-log = logging.getLogger('turbinia:api_server:task')
+log = logging.getLogger(__name__)
 
 router = APIRouter(prefix='/task', tags=['Turbinia Tasks'])
 
@@ -83,8 +83,8 @@ async def get_workers_status(
 async def get_task_status(request: Request, task_id: str):
   """Retrieve task information."""
   try:
-    _state_manager = state_manager.get_state_manager()
-    tasks = _state_manager.get_task_data(
+    state_client = state_manager.get_state_manager()
+    tasks = state_client.get_task_data(
         instance=turbinia_config.INSTANCE_ID, task_id=task_id)
     if tasks:
       task = tasks[0]
diff --git a/turbinia/client.py b/turbinia/client.py
index 5b9a6f755..3083a5932 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Client objects for Turbinia."""
 
-from __future__ import unicode_literals
-
 from collections import defaultdict
 from datetime import datetime
 from datetime import timedelta
@@ -164,7 +162,7 @@ def create_recipe(
       self, debug_tasks=False, filter_patterns=None, group_id='',
       jobs_allowlist=None, jobs_denylist=None, recipe_name=None, sketch_id=None,
       skip_recipe_validation=False, yara_rules=None, group_name=None,
-      reason=None, all_args=None):
+      reason=None):
     """Creates a Turbinia recipe.
 
     If no recipe_name is specified, this  method returns a default recipe.
@@ -186,7 +184,6 @@ def create_recipe(
       yara_rules (str): a string containing yara rules.
       group_name (str): Name for grouping evidence.
       reason (str): Reason or justification for Turbinia requests.
-      all_args (str): a string of commandline arguments provided to run client.
 
     Returns:
       dict: a Turbinia recipe dictionary.
@@ -240,8 +237,6 @@ def create_recipe(
       recipe['globals']['group_name'] = group_name
     if reason:
       recipe['globals']['reason'] = reason
-    if all_args:
-      recipe['globals']['all_args'] = all_args
     if yara_rules:
       recipe['globals']['yara_rules'] = yara_rules
 
@@ -249,15 +244,13 @@ def create_recipe(
 
   def create_request(
       self, request_id=None, group_id=None, requester=None, recipe=None,
-      context=None, evidence_=None, group_name=None, reason=None,
-      all_args=None):
+      evidence_=None, group_name=None, reason=None):
     """Wrapper method to create a Turbinia request."""
     default_recipe = self.create_recipe()
     request = TurbiniaRequest(
         request_id=request_id, group_id=group_id, requester=requester,
-        recipe=recipe if recipe else default_recipe, context=context,
-        evidence=evidence_, group_name=group_name, reason=reason,
-        all_args=all_args)
+        recipe=recipe if recipe else default_recipe, evidence=evidence_,
+        group_name=group_name, reason=reason)
     return request
 
   def list_jobs(self):
diff --git a/turbinia/config/__init__.py b/turbinia/config/__init__.py
index 403981fbd..d6711fa90 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -12,9 +12,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Basic Turbinia config."""
-
-from __future__ import unicode_literals
+"""Basic Turbiniaconfig."""
 
 import importlib.util
 import importlib.machinery
@@ -56,7 +54,6 @@
     'OUTPUT_DIR',
     'TMP_DIR',
     'SLEEP_TIME',
-    'SINGLE_RUN',
     'MOUNT_DIR_PREFIX',
     'SHARED_FILESYSTEM',
     'DEBUG_TASKS',
@@ -170,7 +167,7 @@ def LoadConfig(config_file=None):
   # Warn about using fallback source config, but it's currently necessary for
   # tests. See issue #446.
   if 'turbinia_config_tmpl' in config_file:
-    log.warning(f'Using fallback source config. {CONFIG_MSG:s}')
+    log.warning(f'Using fallback source config {CONFIG_MSG:s}')
   try:
     config_loader = importlib.machinery.SourceFileLoader('config', config_file)
     config_spec = importlib.util.spec_from_loader(
@@ -191,7 +188,7 @@ def LoadConfig(config_file=None):
     os.environ['GOOGLE_CLOUD_PROJECT'] = _config.TURBINIA_PROJECT
 
   CONFIG = _config
-  log.debug(f'Returning parsed config loaded from {CONFIG.configSource:s}')
+  log.debug(f'Returning parsed config loaded from {_config.configSource:s}')
   return _config
 
 
diff --git a/turbinia/config/logger.py b/turbinia/config/logger.py
index 8fa38621e..4e9677bbf 100644
--- a/turbinia/config/logger.py
+++ b/turbinia/config/logger.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 """Sets up logging."""
 
-from __future__ import unicode_literals
 import logging
 import logging.handlers
 import warnings
diff --git a/turbinia/config/turbinia_config_tmpl.py b/turbinia/config/turbinia_config_tmpl.py
index 3a6bac8fc..488314e41 100644
--- a/turbinia/config/turbinia_config_tmpl.py
+++ b/turbinia/config/turbinia_config_tmpl.py
@@ -74,9 +74,6 @@
 # Time in seconds to sleep in task management loops
 SLEEP_TIME = 10
 
-# Whether to run as a single run, or to keep server running indefinitely
-SINGLE_RUN = False
-
 # Local directory in the worker to put other mount directories for locally
 # mounting images/disks
 MOUNT_DIR_PREFIX = '/mnt/turbinia-mounts'
@@ -155,7 +152,7 @@
 # still be enabled with the --jobs_allowlist flag on the server, but the client
 # will not be able to allowlist jobs that have been disabled or denylisted on
 # the server.
-DISABLED_JOBS = ['BinaryExtractorJob', 'BulkExtractorJob', 'DfdeweyJob', 'HindsightJob', 'PhotorecJob']  # yapf: disable
+DISABLED_JOBS = ['VolatilityJob', 'BinaryExtractorJob', 'BulkExtractorJob', 'DfdeweyJob', 'HindsightJob', 'PhotorecJob']  # yapf: disable
 
 # Configure additional job dependency checks below.
 DEPENDENCIES = [{
@@ -396,4 +393,4 @@
 # in llm_client.py module with the provider name (as key) and
 # implementationclass (as value).
 # possible values ["vertexai"]
-LLM_PROVIDER = "vertexai"
+LLM_PROVIDER = 'vertexai'
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 56a831b28..5a8b8a4bc 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Turbinia Evidence objects."""
 
-from __future__ import unicode_literals
-
 from enum import IntEnum
 from collections import defaultdict
 from typing import Any
@@ -196,8 +194,10 @@ class Evidence:
     copyable (bool): Whether this evidence can be copied.  This will be set to
         True for object types that we want to copy to/from storage (e.g.
         PlasoFile, but not RawDisk).
+    creation_time (datetime): The time the evidence was created.
     name (str): Name of evidence.
     description (str): Description of evidence.
+    hash (str): The hash value of the file/device associated with this evidence.
     size (int): The evidence size in bytes where available (Used for metric
         tracking).
     saved_path (str): Path to secondary location evidence is saved for later
@@ -227,6 +227,7 @@ class Evidence:
     mount_path (str): Path to a mounted file system (if relevant).
     credentials (list): Decryption keys for encrypted evidence.
     tags (dict): Extra tags associated with this evidence.
+    tasks (list): List of tasks that have processed this evidence.
     request_id (str): The id of the request this evidence came from, if any.
     has_child_evidence (bool): This property indicates the evidence object has
         child evidence.
@@ -245,6 +246,17 @@ class Evidence:
         in a state file to allow for access amongst multiple workers.
     resource_id (str): The unique id used to track the state of a given Evidence
         type for stateful tracking.
+
+  Note:
+    Evidence objects will be serialized by the state manager through the
+    `to_json()` method. The state manager will store JSON serializable
+    evidence objects in the datastore (e.g Redis) as TurbiniaEvidence keys
+    by calling the `write_evidence()` method. The `TurbiniaEvidence` keys
+    will contain all the attributes except the `config` attribute because
+    it can be quite large and is currently not necessary to store it. If
+    the evidence object contains the `hash` attribute, the state manager
+    will update a `TurbiniaEvidenceHashes` key with the corresponding hash
+    value and evidence id.
   """
 
   # The list of attributes a given piece of Evidence requires to be set
@@ -276,8 +288,6 @@ def __init__(self, *args, **kwargs):
     self.mount_path = kwargs.get('mount_path', None)
     self._name = kwargs.get('name')
     self.parent_evidence = kwargs.get('parent_evidence', None)
-    # List of jobs that have processed this evidence
-    self.processed_by = kwargs.get('processed_by', [])
     self.request_id = kwargs.get('request_id', None)
     self.resource_id = kwargs.get('resource_id', None)
     self.resource_tracked = kwargs.get('resource_tracked', False)
@@ -290,7 +300,6 @@ def __init__(self, *args, **kwargs):
     self.tags = kwargs.get('tags', {})
     self.tasks = kwargs.get('tasks', [])
     self.type = self.__class__.__name__
-
     self.local_path = self.source_path
 
     if 'state' in kwargs:
@@ -502,7 +511,7 @@ def preprocess(self, task_id, tmp_dir=None, required_states=None):
     Raises:
       TurbiniaException: If the required evidence state cannot be met by the
           possible states of the Evidence or if the parent evidence object does
-          not exist when it is required by the Evidence type..
+          not exist when it is required by the Evidence type.
     """
     self.local_path = self.source_path
     if not required_states:
diff --git a/turbinia/job_utils.py b/turbinia/job_utils.py
index 31110ce32..e0e10a68e 100644
--- a/turbinia/job_utils.py
+++ b/turbinia/job_utils.py
@@ -41,7 +41,7 @@ def register_job_timeouts(dependencies):
     timeout = values.get('timeout')
     if not isinstance(timeout, int):
       log.warning(
-          'No timeout found for job: {0:s}. Setting default timeout of {1:d} seconds.'
-          .format(job, timeout_default))
+          f'No timeout found for job: {job}. '
+          'Setting default timeout of {timeout_default} seconds.')
       timeout = timeout_default
-    job_manager.JobsManager.RegisterTimeout(job, timeout)
\ No newline at end of file
+    job_manager.JobsManager.RegisterTimeout(job, timeout)
diff --git a/turbinia/jobs/binary_extractor.py b/turbinia/jobs/binary_extractor.py
index 9bce054c9..7868979fe 100644
--- a/turbinia/jobs/binary_extractor.py
+++ b/turbinia/jobs/binary_extractor.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute binary_extractor task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import Directory
 from turbinia.evidence import GoogleCloudDisk
 from turbinia.evidence import GoogleCloudDiskRawEmbedded
diff --git a/turbinia/jobs/bulk_extractor.py b/turbinia/jobs/bulk_extractor.py
index 89e777230..0bbed5a97 100644
--- a/turbinia/jobs/bulk_extractor.py
+++ b/turbinia/jobs/bulk_extractor.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Bulk Extractor task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import GoogleCloudDisk
 from turbinia.evidence import Directory
 from turbinia.evidence import CompressedDirectory
diff --git a/turbinia/jobs/containerd.py b/turbinia/jobs/containerd.py
index 7d5be8461..af0fa17da 100644
--- a/turbinia/jobs/containerd.py
+++ b/turbinia/jobs/containerd.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to enumerate and mount containerd containers."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import DiskPartition
 from turbinia.evidence import ContainerdContainer
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/dfdewey.py b/turbinia/jobs/dfdewey.py
index a7fc7c637..fe5e9a796 100644
--- a/turbinia/jobs/dfdewey.py
+++ b/turbinia/jobs/dfdewey.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute dfDewey task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import GoogleCloudDisk
 from turbinia.evidence import GoogleCloudDiskRawEmbedded
 from turbinia.evidence import RawDisk
diff --git a/turbinia/jobs/docker.py b/turbinia/jobs/docker.py
index 54a16a613..37d7ac0d6 100644
--- a/turbinia/jobs/docker.py
+++ b/turbinia/jobs/docker.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute docker-explorer, and generate new Evidence per container"""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import DiskPartition
 from turbinia.evidence import DockerContainer
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/file_system_timeline.py b/turbinia/jobs/file_system_timeline.py
index c610465e4..c197ffe65 100644
--- a/turbinia/jobs/file_system_timeline.py
+++ b/turbinia/jobs/file_system_timeline.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to create a file system timeline task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import GoogleCloudDisk
 from turbinia.evidence import GoogleCloudDiskRawEmbedded
 from turbinia.evidence import RawDisk
diff --git a/turbinia/jobs/finalize_request.py b/turbinia/jobs/finalize_request.py
index 6c446e196..514c2993c 100644
--- a/turbinia/jobs/finalize_request.py
+++ b/turbinia/jobs/finalize_request.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to finalize a request."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import FinalReport
 from turbinia.jobs import interface
 from turbinia.jobs import manager
diff --git a/turbinia/jobs/fsstat.py b/turbinia/jobs/fsstat.py
index a13e696b7..ae75ec7a4 100644
--- a/turbinia/jobs/fsstat.py
+++ b/turbinia/jobs/fsstat.py
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Job to run fsstat Task."""
-from __future__ import unicode_literals
+
 from turbinia.evidence import DiskPartition
 from turbinia.jobs import interface
 from turbinia.jobs import manager
@@ -40,4 +40,4 @@ def create_tasks(self, evidence):
     return [FsstatTask() for _ in evidence]
 
 
-manager.JobsManager.RegisterJob(FsstatJob)
\ No newline at end of file
+manager.JobsManager.RegisterJob(FsstatJob)
diff --git a/turbinia/jobs/grep.py b/turbinia/jobs/grep.py
index b2ee22f1f..6b624e283 100644
--- a/turbinia/jobs/grep.py
+++ b/turbinia/jobs/grep.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute grep task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import TextFile
 from turbinia.evidence import FilteredTextFile
 from turbinia.evidence import PlasoCsvFile
diff --git a/turbinia/jobs/hindsight.py b/turbinia/jobs/hindsight.py
index dce33e7e6..91f981b71 100644
--- a/turbinia/jobs/hindsight.py
+++ b/turbinia/jobs/hindsight.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Hindsight tasks."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import ChromiumProfile
 from turbinia.evidence import TextFile
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/http_access_logs.py b/turbinia/jobs/http_access_logs.py
index 4795e74d4..cffa5b5a7 100644
--- a/turbinia/jobs/http_access_logs.py
+++ b/turbinia/jobs/http_access_logs.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Job to execute HTTP Access logs analysis task."""
-from __future__ import unicode_literals
 
 from turbinia.evidence import Directory
 from turbinia.evidence import ContainerdContainer
diff --git a/turbinia/jobs/jenkins.py b/turbinia/jobs/jenkins.py
index dd3483072..a01b4d915 100644
--- a/turbinia/jobs/jenkins.py
+++ b/turbinia/jobs/jenkins.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Jenkins analysis task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import ContainerdContainer
 from turbinia.evidence import Directory
 from turbinia.evidence import DockerContainer
diff --git a/turbinia/jobs/jupyter.py b/turbinia/jobs/jupyter.py
index 541e42c8b..7fc99668b 100644
--- a/turbinia/jobs/jupyter.py
+++ b/turbinia/jobs/jupyter.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Jupyter Notebook config analysis."""
 
-from __future__ import unicode_literals
-
 from turbinia.workers import artifact
 from turbinia.evidence import ContainerdContainer
 from turbinia.evidence import Directory
diff --git a/turbinia/jobs/llm_artifacts_analyzer.py b/turbinia/jobs/llm_artifacts_analyzer.py
index a0b2f5571..3961ae8de 100644
--- a/turbinia/jobs/llm_artifacts_analyzer.py
+++ b/turbinia/jobs/llm_artifacts_analyzer.py
@@ -1,7 +1,5 @@
 """Job to execute LLM analysis on log, config and history files."""
 
-from __future__ import unicode_literals
-
 import logging
 
 from turbinia import config as turbinia_config
diff --git a/turbinia/jobs/manager.py b/turbinia/jobs/manager.py
index 42f1cf17d..2a3e21b52 100644
--- a/turbinia/jobs/manager.py
+++ b/turbinia/jobs/manager.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """This file contains a class for managing jobs."""
 
-from __future__ import unicode_literals
-
 from turbinia import TurbiniaException
 
 DEFAULT_TIMEOUT = 7200
diff --git a/turbinia/jobs/photorec.py b/turbinia/jobs/photorec.py
index 8e8541ea7..e25bc5fbb 100644
--- a/turbinia/jobs/photorec.py
+++ b/turbinia/jobs/photorec.py
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Job to run photorec Task."""
-from __future__ import unicode_literals
+
 from turbinia.evidence import DiskPartition
 from turbinia.jobs import interface
 from turbinia.jobs import manager
diff --git a/turbinia/jobs/plaso.py b/turbinia/jobs/plaso.py
index 54807a05b..ef995772a 100644
--- a/turbinia/jobs/plaso.py
+++ b/turbinia/jobs/plaso.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Plaso task."""
 
-from __future__ import unicode_literals
-
 import logging
 
 from turbinia.evidence import EwfDisk
diff --git a/turbinia/jobs/psort.py b/turbinia/jobs/psort.py
index 8a5db6dc9..da11b019e 100644
--- a/turbinia/jobs/psort.py
+++ b/turbinia/jobs/psort.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute Psort task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import PlasoFile
 from turbinia.evidence import PlasoCsvFile
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/redis.py b/turbinia/jobs/redis.py
index 6707fe50f..2ca1980e8 100644
--- a/turbinia/jobs/redis.py
+++ b/turbinia/jobs/redis.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute redis_config analysis task."""
 
-from __future__ import unicode_literals
-
 from turbinia.workers import artifact
 from turbinia.workers.analysis import redis
 from turbinia.evidence import ContainerdContainer
diff --git a/turbinia/jobs/ssh_analyzer.py b/turbinia/jobs/ssh_analyzer.py
index db7984418..02fcdf82c 100644
--- a/turbinia/jobs/ssh_analyzer.py
+++ b/turbinia/jobs/ssh_analyzer.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute SSH analyzer task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import ContainerdContainer
 from turbinia.evidence import Directory
 from turbinia.evidence import DockerContainer
diff --git a/turbinia/jobs/sshd.py b/turbinia/jobs/sshd.py
index ac3e64af4..15a16534b 100644
--- a/turbinia/jobs/sshd.py
+++ b/turbinia/jobs/sshd.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute sshd_config analysis task."""
 
-from __future__ import unicode_literals
-
 from turbinia.workers import artifact
 from turbinia.workers.analysis import sshd
 from turbinia.evidence import ContainerdContainer
diff --git a/turbinia/jobs/strings.py b/turbinia/jobs/strings.py
index 2b026b8a8..570627ae8 100644
--- a/turbinia/jobs/strings.py
+++ b/turbinia/jobs/strings.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute strings task."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import DiskPartition
 from turbinia.evidence import TextFile
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/tomcat.py b/turbinia/jobs/tomcat.py
index 36e7abf0c..8b1e94e08 100644
--- a/turbinia/jobs/tomcat.py
+++ b/turbinia/jobs/tomcat.py
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Job to execute Apache Tomcat analysis task."""
-from __future__ import unicode_literals
+
 from turbinia.workers import artifact
 from turbinia.workers.analysis import tomcat
 from turbinia.evidence import ContainerdContainer
diff --git a/turbinia/jobs/volatility.py b/turbinia/jobs/volatility.py
index f6615501a..f51cf0314 100644
--- a/turbinia/jobs/volatility.py
+++ b/turbinia/jobs/volatility.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job to execute volatility tasks."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import RawMemory
 from turbinia.evidence import VolatilityReport
 from turbinia.jobs import interface
diff --git a/turbinia/jobs/worker_stat.py b/turbinia/jobs/worker_stat.py
index 660e10ab8..cef00270d 100644
--- a/turbinia/jobs/worker_stat.py
+++ b/turbinia/jobs/worker_stat.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Job for running a stat call on the supplied evidence."""
 
-from __future__ import unicode_literals
-
 from turbinia.evidence import EwfDisk
 from turbinia.evidence import Directory
 from turbinia.evidence import RawDisk
diff --git a/turbinia/lib/docker_manager.py b/turbinia/lib/docker_manager.py
index e004e9ce2..52e301eee 100644
--- a/turbinia/lib/docker_manager.py
+++ b/turbinia/lib/docker_manager.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Library to handle docker related queries."""
 
-from __future__ import unicode_literals
-
 import logging
 import codecs
 import os
diff --git a/turbinia/lib/llm_libs/llm_lib_base.py b/turbinia/lib/llm_libs/llm_lib_base.py
index f3e59fc39..97ef3a23c 100644
--- a/turbinia/lib/llm_libs/llm_lib_base.py
+++ b/turbinia/lib/llm_libs/llm_lib_base.py
@@ -1,7 +1,5 @@
 """Base class for Turbinia LLM libraries."""
 
-from __future__ import unicode_literals
-
 import abc
 import typing
 
diff --git a/turbinia/lib/llm_libs/vertex_ai_lib.py b/turbinia/lib/llm_libs/vertex_ai_lib.py
index be4f78c77..5badd44de 100644
--- a/turbinia/lib/llm_libs/vertex_ai_lib.py
+++ b/turbinia/lib/llm_libs/vertex_ai_lib.py
@@ -1,7 +1,5 @@
 """Turbinia LLM library that uses VertexAI APIs."""
 
-from __future__ import unicode_literals
-
 import logging
 
 import backoff
diff --git a/turbinia/lib/recipe_helpers.py b/turbinia/lib/recipe_helpers.py
index 9036775e0..c14cdc3e1 100644
--- a/turbinia/lib/recipe_helpers.py
+++ b/turbinia/lib/recipe_helpers.py
@@ -42,7 +42,6 @@
     'sketch_id': None,
     'group_name': '',
     'reason': '',
-    'all_args': '',
     'group_id': ''
 }
 
diff --git a/turbinia/lib/text_formatter.py b/turbinia/lib/text_formatter.py
index 0d550a6d7..babb8449d 100644
--- a/turbinia/lib/text_formatter.py
+++ b/turbinia/lib/text_formatter.py
@@ -16,7 +16,6 @@
 """Methods for formatting text."""
 
 from __future__ import print_function
-from __future__ import unicode_literals
 
 
 def bold(text):
diff --git a/turbinia/lib/utils.py b/turbinia/lib/utils.py
index 36adde6c6..9aaa117ea 100644
--- a/turbinia/lib/utils.py
+++ b/turbinia/lib/utils.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Common utils."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import subprocess
diff --git a/turbinia/message.py b/turbinia/message.py
index 5af90f01a..cc2293971 100644
--- a/turbinia/message.py
+++ b/turbinia/message.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Core classes for Turbinia Requests and Messaging components."""
 
-from __future__ import unicode_literals
-
 import codecs
 import copy
 import json
@@ -23,8 +21,11 @@
 import logging
 import six
 
-from turbinia import evidence
+from datetime import datetime
+
+from turbinia import evidence as turbinia_evidence
 from turbinia import TurbiniaException
+from turbinia.config import DATETIME_FORMAT
 
 log = logging.getLogger(__name__)
 
@@ -33,49 +34,97 @@ class TurbiniaRequest:
   """An object to request evidence to be processed.
 
   Attributes:
-    request_id(str): A client specified ID for this request.
+    evidence (list): A list of Evidence objects.
+    failed_tasks (list): List of failed tasks.
     group_id(str): A client specified group id for this request.
-    requester(str): The username of who made the request.
-    recipe(dict): Recipe to use when processing this request.
-    context(dict): A Dict of context data to be passed around with this request.
-    evidence(list): A list of Evidence objects.
     group_name (str): Name for grouping evidence.
+    last_update (datetime.datetime): Last modification timestmap. 
+    queued_tasks (list): List of queued tasks.
     reason (str): Reason or justification for Turbinia requests.
-    all_args (str): a string of commandline arguments provided to run client.
+    recipe(dict): Recipe to use when processing this request.
+    request_id(str): A client specified ID for this request.
+    requester(str): The username of who made the request.
+    running_tasks (list): List of running tasks.
+    start_time (datetime.datetime): Task start timestamp.
+    status (str): The status of the request.
+    successful_tasks (list): List of successful tasks.
+    task_ids (list): List of all tasks associated with the request.
+    type (str): 'TurbiniaRequest' or class name.
+
+  Note:
+    Objects of this class will be stored by the state manager. The state
+    manager will persist all attributes that are serializable to JSON by
+    calling to_json(). Evidence objects are not serializable, but the
+    evidence identifiers (IDs) are. Evidence IDs are stored in the
+    evidence_ids key of the JSON object.
   """
 
   def __init__(
       self, request_id=None, group_id=None, requester=None, recipe=None,
-      context=None, evidence=None, group_name=None, reason=None, all_args=None):
+      evidence=None, group_name=None, reason=None):
     """Initialization for TurbiniaRequest."""
-    self.request_id = request_id if request_id else uuid.uuid4().hex
-    self.group_id = group_id if group_id else uuid.uuid4().hex
-    self.requester = requester if requester else 'user_unspecified'
-    self.recipe = recipe if recipe else {'globals': {}}
-    self.context = context if context else {}
     self.evidence = evidence if evidence else []
+    if evidence and len(evidence) > 0:
+      self.original_evidence = {'id': evidence[0].id, 'name': evidence[0].name}
+    else:
+      self.original_evidence = {}
+    self.failed_tasks = []
+    self.group_id = group_id if group_id else uuid.uuid4().hex
     self.group_name = group_name if group_name else ''
+    self.last_update = datetime.now().strftime(DATETIME_FORMAT)
+    self.queued_tasks = []
     self.reason = reason if reason else ''
-    self.all_args = all_args if all_args else ''
+    self.recipe = recipe if recipe else {'globals': {}}
+    self.request_id = request_id if request_id else uuid.uuid4().hex
+    self.requester = requester if requester else 'user_unspecified'
+    self.running_tasks = []
+    self.start_time = datetime.now().strftime(DATETIME_FORMAT)
+    self.status = 'pending'
+    self.successful_tasks = []
+    self.task_ids = []
     self.type = self.__class__.__name__
 
-  def to_json(self):
+  def to_json(self, json_values=False):
     """Convert object to JSON.
 
+    Args:
+      json_values (bool): Returns only values of the dictionary as json strings
+        instead of the entire dictionary.
+
     Returns:
       A JSON serialized object.
     """
     serializable = copy.deepcopy(self.__dict__)
-    serializable['evidence'] = [x.serialize() for x in serializable['evidence']]
-
-    try:
-      serialized = json.dumps(serializable)
-    except TypeError as exception:
-      msg = (
-          'JSON serialization of TurbiniaRequest object {0:s} failed: '
-          '{1:s}'.format(self.type, str(exception)))
-      raise TurbiniaException(msg)
-
+    if json_values:
+      if evidence_list := serializable.pop('evidence'):
+        if not serializable.get('original_evidence') and len(evidence_list) > 0:
+          serializable['original_evidence'] = {
+              'name': evidence_list[0].name,
+              'id': evidence_list[0].id
+          }
+        serializable['evidence_ids'] = [
+            evidence.id for evidence in evidence_list
+        ]
+      serialized = {}
+      try:
+        for attribute_name, attribute_value in serializable.items():
+          serialized[attribute_name] = json.dumps(attribute_value)
+      except TypeError as exception:
+        msg = (
+            f'JSON serialization of TurbiniaRequest object {self.type} '
+            f'failed: {str(exception)}')
+        raise TurbiniaException(msg) from exception
+    else:
+      serializable['evidence'] = [
+          x.serialize() for x in serializable['evidence']
+      ]
+      try:
+        serialized = json.dumps(serializable)
+      except TypeError as exception:
+        msg = (
+            f'JSON serialization of TurbiniaRequest object {self.type} '
+            f'failed: {str(exception)}')
+        raise TurbiniaException(msg) from exception
     return serialized
 
   def from_json(self, json_str):
@@ -94,13 +143,15 @@ def from_json(self, json_str):
       obj = json.loads(json_str)
     except ValueError as exception:
       raise TurbiniaException(
-          f'Can not load json from string {str(exception):s}')
+          f'Can not load json from string {str(exception):s}') from exception
 
     if obj.get('type', None) != self.type:
       raise TurbiniaException(
           f'Deserialized object does not have type of {self.type:s}')
 
-    obj['evidence'] = [evidence.evidence_decode(e) for e in obj['evidence']]
+    obj['evidence'] = [
+        turbinia_evidence.evidence_decode(e) for e in obj['evidence']
+    ]
     # pylint: disable=attribute-defined-outside-init
     self.__dict__ = obj
 
diff --git a/turbinia/output_manager.py b/turbinia/output_manager.py
index 9bf146922..886bb4349 100644
--- a/turbinia/output_manager.py
+++ b/turbinia/output_manager.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Classes to write output to various location types."""
 
-from __future__ import unicode_literals
-
 import errno
 import json
 import logging
diff --git a/turbinia/processors/archive.py b/turbinia/processors/archive.py
index 0cf257a10..d8d76e778 100644
--- a/turbinia/processors/archive.py
+++ b/turbinia/processors/archive.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """File archiving processor"""
 
-from __future__ import unicode_literals
-
 import os
 import tarfile
 import logging
diff --git a/turbinia/processors/containerd.py b/turbinia/processors/containerd.py
index 1361c71ab..242ee1d20 100644
--- a/turbinia/processors/containerd.py
+++ b/turbinia/processors/containerd.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Evidence processor to mount local containerd containers."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import subprocess
diff --git a/turbinia/processors/docker.py b/turbinia/processors/docker.py
index 9141e63ee..66fc1829c 100644
--- a/turbinia/processors/docker.py
+++ b/turbinia/processors/docker.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Evidence processor to mount local Docker containers."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import subprocess
diff --git a/turbinia/processors/google_cloud.py b/turbinia/processors/google_cloud.py
index 2f7c26046..e8a49e7c9 100644
--- a/turbinia/processors/google_cloud.py
+++ b/turbinia/processors/google_cloud.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Evidence processor for Google Cloud resources."""
 
-from __future__ import unicode_literals
-
 import glob
 import logging
 import os
diff --git a/turbinia/processors/mount_local.py b/turbinia/processors/mount_local.py
index cb132954a..a4e67f486 100644
--- a/turbinia/processors/mount_local.py
+++ b/turbinia/processors/mount_local.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Evidence processor to mount local images or disks."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import subprocess
@@ -65,17 +63,16 @@ def GetDiskSize(source_path):
     size = int(cmd_output[0].decode('utf-8'))
   except subprocess.CalledProcessError:
     log.debug(
-        'blockdev failed, attempting to get file size using stat() instead')
+        'blockdev failed, attempting to get evidence size using stat() instead')
   except ValueError:
     log.warning(
         f"Unexpected output from blockdev: {cmd_output[0].decode('utf-8'):s}")
-
   if not size:
     # evidence is not a block device, check image file size
     try:
       size = os.stat(source_path).st_size
     except OSError as exception:
-      log.warning(f'Checking disk size failed: {exception!s}')
+      log.warning(f'Checking evidence size failed: {exception!s}')
 
   return size
 
diff --git a/turbinia/processors/resource_manager.py b/turbinia/processors/resource_manager.py
index 41be625d2..8092dc763 100644
--- a/turbinia/processors/resource_manager.py
+++ b/turbinia/processors/resource_manager.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Handles management of the resource state file."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import json
diff --git a/turbinia/redis_client.py b/turbinia/redis_client.py
new file mode 100644
index 000000000..a280a9884
--- /dev/null
+++ b/turbinia/redis_client.py
@@ -0,0 +1,322 @@
+# -*- coding: utf-8 -*-
+# Copyright 2024 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""State manager helper"""
+
+import logging
+import json
+import redis
+
+from typing import Any, Iterator, Dict
+
+from turbinia import config
+
+log = logging.getLogger(__name__)
+
+
+class RedisClientError(Exception):
+  """This class handles Redis client errors."""
+  pass
+
+
+class RedisClient:
+  """This class handles Redis operations on keys/values."""
+
+  def __init__(self, redis_client=None):
+    self.client = redis_client
+    if not self.client:
+      self.client = redis.StrictRedis(
+          host=config.REDIS_HOST, port=config.REDIS_PORT, db=config.REDIS_DB,
+          socket_timeout=10, socket_keepalive=True, socket_connect_timeout=10)
+
+  def set_attribute(
+      self, redis_key: str, attribute_name: str, json_value: str) -> bool:
+    """Sets the attribute of a Turbinia hash object in redis.
+
+    Args:
+      redis_key (str): The key of the Turbinia hash object in redis.
+      attribute_name (str): The name of the attribute to be set.
+      json_value (str): The json-serialized value to be set
+
+    Returns:
+      (bool): Boolean specifying whether the function call was successful. 
+
+    Raises:
+      RedisClientError: if there was an error setting the attribute value.
+    """
+    try:
+      values_set = self.client.hset(redis_key, attribute_name, json_value)
+      return bool(values_set)
+    except redis.RedisError as exception:
+      error_message = f'Error setting {attribute_name} for key {redis_key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def iterate_keys(self, key_type: str) -> Iterator[str]:
+    """Iterates over the Turbinia keys of a specific type.
+
+    Args:
+      key_type (str): The type of the Turbinia key (e.g. Task, Evidence)
+
+    Yields:
+      key (str): Decoded key of stored Turbinia object. 
+
+    Raises:
+      RedisClientError: If Redis fails in getting the keys or if
+        decode fails.
+    """
+    valid_key_types = ('evidence', 'task', 'request')
+    if key_type.lower() not in valid_key_types:
+      raise RedisClientError(f'Invalid key type: {key_type}')
+
+    try:
+      keys = self.client.scan_iter(f'Turbinia{key_type.title()}:*', count=1000)
+    except redis.RedisError as exception:
+      error_message = f'Error getting {key_type} keys in Redis'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+    try:
+      for data in keys:
+        yield data.decode()
+    except ValueError as exception:
+      error_message = 'Error decoding key in Redis'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def get_attribute(
+      self, redis_key: str, attribute_name: str,
+      decode_json: bool = True) -> Any:
+    """Gets the attribute of a Turbinia hash object in redis.
+
+    Args:
+      redis_key (str): The key of the Turbinia hash object in redis.
+      attribute_name (str): The name of the attribute to be get.
+      decode_json (bool): Boolean specifying if the value should be loaded.
+
+    Returns:
+      attribute_value (any): successful. 
+
+    Raises:
+      RedisClientError: If Redis fails in getting the attribute or if
+        json loads fails.
+    """
+    try:
+      attribute_value = self.client.hget(redis_key, attribute_name)
+      if not attribute_value:
+        message = f'Attribute {attribute_name} for key {redis_key} not found.'
+        log.warning(message)
+    except redis.RedisError as exception:
+      error_message = (
+          f'Error getting {attribute_name} from {redis_key} in Redis')
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+    if decode_json and attribute_value:
+      try:
+        return json.loads(attribute_value)
+      except (TypeError, ValueError) as exception:
+        error_message = (
+            f'Error decoding JSON {attribute_name} on {redis_key} '
+            f'in Redis')
+        log.error(f'{error_message}: {exception}')
+        raise RedisClientError(error_message) from exception
+    else:
+      return attribute_value
+
+  def iterate_attributes(self, key: str) -> Iterator[tuple]:
+    """Iterates over the attribute names of the Redis hash object.
+
+    Args:
+      key (str): The key of the stored hash object.
+
+    Yields:
+      attribute_name (tuple): Decoded name of object attribute.
+
+    Raises:
+      RedisClientError: If Redis fails in getting the attributes or if
+        decode or json loads fails. 
+    """
+    try:
+      for attribute in self.client.hscan_iter(key, count=100):
+        if attribute:
+          try:
+            attribute_name = attribute[0].decode()
+            attribute_value = json.loads(attribute[1])
+            yield (attribute_name, attribute_value)
+          except (TypeError, ValueError) as exception:
+            error_message = (
+                f'Error decoding JSON value for {attribute_name} on {key} '
+                f'in Redis')
+            log.error(f'{error_message}: {exception}')
+            raise RedisClientError(error_message) from exception
+    except (redis.exceptions.ResponseError, redis.RedisError) as exception:
+      error_message = f'Error getting attributes from {key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def build_key_name(self, key_type: str, identifier: str) -> str:
+    """Returns a valid redis key name.
+
+    Args:
+      key_type (str):  The type of key (e.g. request, task, evidence).
+      identifier (str): The ID used to build the key name.
+    
+    Returns:
+      str: A valid Redis key name.
+
+    Raises:
+      ValueError: If an invalid key type was provided.
+    """
+    redis_key = ''
+    if key_type == 'request':
+      redis_key = f'TurbiniaRequest:{identifier}'
+    elif key_type == 'task':
+      redis_key = f'TurbiniaTask:{identifier}'
+    elif key_type == 'evidence':
+      redis_key = f'TurbiniaEvidence:{identifier}'
+    else:
+      raise ValueError('{key_type} is not a valid type of key.')
+    return redis_key
+
+  def key_exists(self, redis_key: str) -> bool:
+    """Checks if the key is saved in Redis.
+
+    Args:
+      redis_key (str): The redis key name to be checked.
+
+    Returns:
+      exists (bool): Boolean indicating if key is saved. 
+
+    Raises:
+      RedisClientError: If Redis fails in checking the existence of the key.
+    """
+    try:
+      return self.client.exists(redis_key)
+    except redis.RedisError as exception:
+      error_message = f'Error checking existence of {redis_key} in Redis'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def attribute_exists(self, redis_key: str, attribute_name: str) -> bool:
+    """Checks if the attribute of the hashed key is saved in Redis.
+
+    Args:
+      redis_key (str): The key to be checked.
+      attribute_name (str): The attribute to be checked.
+
+    Returns:
+      exists (bool): Boolean indicating if attribute is saved. 
+
+    Raises:
+      RedisClientError: If Redis fails in checking the existence.
+    """
+    try:
+      return self.client.hexists(redis_key, attribute_name)
+    except redis.RedisError as exception:
+      error_message = (
+          f'Error checking existence of attribute {attribute_name}'
+          f'for key {redis_key}')
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def get_key_type(self, redis_key: str) -> bool:
+    """Gets the type of the Redis key.
+
+    Args:
+      redis_key (str): The key to be checked.
+
+    Returns:
+      type (str): Type of the Redis key. 
+
+    Raises:
+      RedisClientError: If Redis fails in getting the type of the key.
+    """
+    try:
+      return self.client.type(redis_key)
+    except redis.RedisError as exception:
+      error_message = f'Error getting type for key {redis_key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def add_to_list(
+      self, redis_key: str, list_name: str, new_item: Any,
+      allow_repeated: bool = False) -> None:
+    """Appends new item to a list attribute in a hashed Redis object.
+
+    Args:
+      redis_key: Key of the Redis object.
+      list_name: Name of the list attribute.
+      new_item: Item to be saved.
+      repeated: Allows repeated items to be saved.
+
+    Raises:
+      RedisClientError: If there was an error writing to Redis.
+    """
+    if not self.attribute_exists(redis_key, list_name):
+      list_attribute = [new_item]
+    else:
+      list_attribute = self.get_attribute(redis_key, list_name)
+    if new_item not in list_attribute and not allow_repeated:
+      list_attribute.append(new_item)
+    try:
+      self.set_attribute(redis_key, list_name, json.dumps(list_attribute))
+    except (TypeError, ValueError) as exception:
+      error_message = f'Error encoding list {list_attribute} from {redis_key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def remove_from_list(self, redis_key: str, list_name: str, item: Any) -> None:
+    """Removes an item from a list attribute in a hashed Redis object.
+
+    Args:
+      redis_key: Key of the Redis object.
+      list_name: Name of the list attribute.
+      item: Item to be removed.
+
+    Raises:
+      RedisClientError: If there was an error writing to Redis.
+    """
+    if not self.attribute_exists(redis_key, list_name):
+      return
+    list_attribute = self.get_attribute(redis_key, list_name)
+    if item in list_attribute:
+      list_attribute.remove(item)
+    try:
+      self.set_attribute(redis_key, list_name, json.dumps(list_attribute))
+    except (TypeError, ValueError) as exception:
+      error_message = f'Error encoding list {list_attribute} from {redis_key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+
+  def write_hash_object(
+      self, redis_key: str, object_dict: Dict[str, Any]) -> str:
+    """Writes new hash object into redis. 
+
+    Args:
+      object_dict: A dictionary containing the serialized
+        attributes that will be saved.
+
+    Returns:
+      redis_key: The key corresponding to the object in Redis
+    
+    Raises:
+      RedisClientError: if there was an error writing the key.
+    """
+    log.debug(f'Updating key {redis_key}')
+    try:
+      self.client.hset(redis_key, mapping=object_dict)
+    except redis.RedisError as exception:
+      error_message = f'Error writing {redis_key}'
+      log.error(f'{error_message}: {exception}')
+      raise RedisClientError(error_message) from exception
+    return redis_key
diff --git a/turbinia/state_manager.py b/turbinia/state_manager.py
index e4b5b56ac..838cee580 100644
--- a/turbinia/state_manager.py
+++ b/turbinia/state_manager.py
@@ -18,21 +18,20 @@
 storage.
 """
 
-from __future__ import unicode_literals
-
 import codecs
 import json
 import logging
 import sys
 from datetime import datetime
 from datetime import timedelta
-from typing import Any, Iterator
+from typing import Any, List, Dict, Optional
 
 import six
 
 from turbinia import config
 from turbinia.config import DATETIME_FORMAT
 from turbinia import TurbiniaException
+from turbinia.redis_client import RedisClient, RedisClientError
 
 config.LoadConfig()
 if 'unittest' in sys.modules.keys():
@@ -45,7 +44,6 @@
   msg = f'State Manager type "{config.STATE_MANAGER:s}" not implemented'
   raise TurbiniaException(msg)
 
-EMPTY_JSON_VALUES = ('null', '{}', '[]')
 MAX_DATASTORE_STRLEN = 1500
 log = logging.getLogger(__name__)
 
@@ -64,14 +62,14 @@ def get_state_manager():
   if config.STATE_MANAGER.lower() == 'redis':
     return RedisStateManager()
   else:
-    msg = f'State Manager type "{config.STATE_MANAGER:s}" not implemented'
-    raise TurbiniaException(msg)
+    error = f'State Manager type "{config.STATE_MANAGER:s}" not implemented'
+    raise TurbiniaException(error)
 
 
 class BaseStateManager:
   """Class to manage Turbinia state persistence."""
 
-  def get_task_dict(self, task):
+  def get_task_dict(self, task) -> Dict[str, Any]:
     """Creates a dict of the fields we want to persist into storage.
 
     This combines attributes from both the Task and the TaskResult into one flat
@@ -105,10 +103,6 @@ def get_task_dict(self, task):
         if isinstance(task_dict[attr], six.binary_type):
           task_dict[attr] = six.u(task_dict[attr])
 
-    # We'll store the run_time as seconds instead of a timedelta()
-    if task_dict.get('run_time'):
-      task_dict['run_time'] = task_dict['run_time'].total_seconds()
-
     #Importing these here to avoid circular dependencies.
     from turbinia.workers import TurbiniaTask
     from turbinia.workers import TurbiniaTaskResult
@@ -164,19 +158,60 @@ class RedisStateManager(BaseStateManager):
 
   def __init__(self):
     config.LoadConfig()
-    self.client = redis.StrictRedis(
-        host=config.REDIS_HOST, port=config.REDIS_PORT, db=config.REDIS_DB,
-        socket_timeout=10, socket_keepalive=True, socket_connect_timeout=10)
+    self.redis_client = RedisClient()
 
   def set_client(self, redis_client):
-    self.client = redis_client
+    self.redis_client = redis_client
 
   def _validate_data(self, data):
     return data
 
+  def get_task(self, task_id: str) -> dict:
+    """Returns a dictionary representing a Task object given its ID.
+
+    Args:
+      task_id (str): The ID of the stored task.
+
+    Returns:
+      task_dict (dict): Dict containing task attributes. 
+    """
+    task_key = ':'.join(('TurbiniaTask', task_id))
+    task_dict = {}
+    try:
+      for attribute_name, attribute_value in self.redis_client.iterate_attributes(
+          task_key):
+        task_dict[attribute_name] = attribute_value
+      if task_dict.get('last_update'):
+        task_dict['last_update'] = datetime.strptime(
+            task_dict.get('last_update'), DATETIME_FORMAT)
+      if task_dict.get('run_time'):
+        task_dict['run_time'] = timedelta(seconds=task_dict['run_time'])
+    except RedisClientError as exception:
+      log.error(f'Error retrieving task data {exception}')
+    return task_dict
+
+  def validate_task(
+      self, task, instance: str, days: int, group_id: str, user: str) -> bool:
+    """Returns True if the Task matches the required filters."""
+    result: bool = False
+    try:
+      if days:
+        start_time = datetime.now() - timedelta(days=days)
+        valid_days = task.get('last_update') > start_time
+      else:
+        valid_days = True
+      valid_instance = not instance or task.get('instance') == instance
+      valid_group = not group_id or task.get('group_id') == group_id
+      valid_user = not user or task.get('requester') == user
+      result = valid_days and valid_instance and valid_group and valid_user
+    except TypeError as exception:
+      log.error(f'Error validating task {task.get("id")}: {exception}')
+    return result
+
   def get_task_data(
-      self, instance, days=0, task_id=None, request_id=None, group_id=None,
-      user=None):
+      self, instance: str, days: int = 0, task_id: str = None,
+      request_id: str = None, group_id: str = None,
+      user: str = None) -> List[Dict]:
     """Gets task data from Redis.
 
     Args:
@@ -191,239 +226,176 @@ def get_task_data(
     Returns:
       List of Task dict objects.
     """
-    tasks = [
-        json.loads(self.client.get(task))
-        for task in self.client.scan_iter(match='TurbiniaTask:*', count=1000)
-        if json.loads(self.client.get(task)).get('instance') == instance or
-        not instance
-    ]
-
-    # Convert relevant date attributes back into dates/timedeltas
-    for task in tasks:
-      if task.get('last_update'):
-        task['last_update'] = datetime.strptime(
-            task.get('last_update'), DATETIME_FORMAT)
-      if task.get('run_time'):
-        task['run_time'] = timedelta(seconds=task['run_time'])
-
-    # pylint: disable=no-else-return
-    if days:
-      start_time = datetime.now() - timedelta(days=days)
-      # Redis only supports strings; we convert to/from datetime here and below
-      tasks = [task for task in tasks if task.get('last_update') > start_time]
-    if task_id:
-      tasks = [task for task in tasks if task.get('id') == task_id]
-    if request_id:
-      tasks = [task for task in tasks if task.get('request_id') == request_id]
-    if group_id:
-      tasks = [task for task in tasks if task.get('group_id') == group_id]
-    if user:
-      tasks = [task for task in tasks if task.get('requester') == user]
-
-    return tasks
-
-  def update_task(self, task):
-    task.touch()
-    key = task.state_key
-    if not key:
-      self.write_new_task(task)
-      return
-    stored_task_data = json.loads(self.client.get(f'TurbiniaTask:{task.id}'))
-    stored_evidence_size = stored_task_data.get('evidence_size')
-    stored_evidence_id = stored_task_data.get('evidence_id')
-    if not task.evidence_size and stored_evidence_size:
-      task.evidence_size = stored_evidence_size
-    if not task.evidence_id and stored_evidence_id:
-      task.evidence_id = stored_evidence_id
-    log.info(f'Updating task {task.name:s} in Redis')
-    task_data = self.get_task_dict(task)
-    task_data['last_update'] = task_data['last_update'].strftime(
-        DATETIME_FORMAT)
-    task_data['start_time'] = task_data['start_time'].strftime(DATETIME_FORMAT)
-    # Need to use json.dumps, else redis returns single quoted string which
-    # is invalid json
-    if not self.client.set(key, json.dumps(task_data)):
-      log.error(f'Error updating task {task.name:s} in Redis')
+    if task_id and request_id:
+      raise TurbiniaException(
+          'You can provide a task_id or request_id but not both.')
 
-  def write_new_task(self, task):
-    key = ':'.join(['TurbiniaTask', task.id])
-    log.info(f'Writing new task {task.name:s} into Redis')
-    task_data = self.get_task_dict(task)
-    task_data['last_update'] = task_data['last_update'].strftime(
-        DATETIME_FORMAT)
-    task_data['start_time'] = task_data['start_time'].strftime(DATETIME_FORMAT)
-    if not task_data.get('status'):
-      task_data['status'] = 'Task scheduled at {0:s}'.format(
-          datetime.now().strftime(DATETIME_FORMAT))
-    if task_data['run_time']:
-      task_data['run_time'] = task_data['run_time'].total_seconds()
-    # nx=True prevents overwriting (i.e. no unintentional task clobbering)
-    if not self.client.set(key, json.dumps(task_data), nx=True):
-      log.error(f'Error writing new task {task.name:s} into Redis')
-    task.state_key = key
-    return key
-
-  def set_attribute(
-      self, redis_key: str, attribute_name: str, json_value: str) -> bool:
-    """Sets the attribute of a Turbinia hash object in redis.
-
-    Args:
-      redis_key (str): The key of the Turbinia hash object in redis.
-      attribute_name (str): The name of the attribute to be set.
-      json_value (str): The json-serialized value to be set
-
-    Returns:
-      (bool): Boolean specifying whether the function call was successful. 
+    task_ids = []
 
-    Raises:
-      TurbiniaException: When Redis fails in updating the attribute.
-    """
-    try:
-      if not self.client.hset(redis_key, attribute_name, json_value):
-        log.error(f'Error setting {attribute_name} on {redis_key} in Redis')
-        return False
-      return True
-    except redis.RedisError as exception:
-      error_message = (
-          f'Error setting {attribute_name} on {redis_key} in Redis')
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
+    # If task_id is passed, simply gets and validates the corresponding task
+    if task_id:
+      task = self.get_task(task_id)
+      valid_request = not request_id or task.get('request_id') == request_id
+      valid_task = self.validate_task(task, instance, days, group_id, user)
+      return [task] if valid_request and valid_task else []
 
-  def get_attribute(
-      self, redis_key: str, attribute_name: str,
-      decode_json: bool = True) -> Any:
-    """Gets the attribute of a Turbinia hash object in redis.
+    request_key = self.redis_client.build_key_name('request', request_id)
 
-    Args:
-      redis_key (str): The key of the Turbinia hash object in redis.
-      attribute_name (str): The name of the attribute to be get.
-      decode_json (bool): Boolean specifying if the value should be loaded.
+    # If request_id is passed, gets valid tasks from that request
+    if request_key and self.redis_client.key_exists(request_key):
+      task_ids = self.redis_client.get_attribute(
+          request_key, 'task_ids', decode_json=True)
 
-    Returns:
-      attribute_value (any): successful. 
-
-    Raises:
-      TurbiniaException: If Redis fails in getting the attribute or if
-        json loads fails.
-    """
-    try:
-      attribute_value = self.client.hget(redis_key, attribute_name)
-    except redis.RedisError as exception:
-      error_message = (
-          f'Error getting {attribute_name} from {redis_key} in Redis')
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
-    if decode_json:
-      try:
-        return json.loads(attribute_value)
-      except (TypeError, ValueError) as exception:
-        error_message = (
-            f'Error decoding JSON {attribute_name} on {redis_key} '
-            f'in Redis')
-        log.error(f'{error_message}: {exception}')
-        raise TurbiniaException(error_message) from exception
+    # If no task_id or request_id is passed, gets all valid saved tasks
     else:
-      return attribute_value
+      task_ids = [
+          task_key.split(':')[1]
+          for task_key in self.redis_client.iterate_keys('Task')
+      ]
 
-  def iterate_keys(self, key_type: str) -> Iterator[str]:
-    """Iterates over the Turbinia keys of a specific type.
+    tasks = []
 
-    Args:
-      key_type (str): The type of the Turbinia key (e.g. Task, Evidence)
+    for task_id in task_ids:
+      task = self.get_task(task_id)
+      if self.validate_task(task, instance, days, group_id, user):
+        tasks.append(task)
 
-    Yields:
-      key (str): Decoded key of stored Turbinia object. 
+    return tasks
 
-    Raises:
-      TurbiniaException: If Redis fails in getting the keys or if
-        decode fails.
+  def update_request_task(self, task) -> None:
+    """Adds a Turbinia task to the corresponding request list.
+    
+    Args:
+      task (TurbiniaTask): Turbinia task object.
     """
+    request_key = self.redis_client.build_key_name('request', task.request_id)
+    task_key = self.redis_client.build_key_name('task', task.id)
     try:
-      keys = self.client.scan_iter(f'Turbinia{key_type.title()}:*', count=1000)
-    except redis.RedisError as exception:
-      error_message = f'Error getting {key_type} keys in Redis'
+      self.redis_client.add_to_list(request_key, 'task_ids', task.id)
+      request_last_update = datetime.strptime(
+          self.redis_client.get_attribute(request_key, 'last_update'),
+          DATETIME_FORMAT)
+      last_update = json.dumps(
+          max(request_last_update, task.last_update).strftime(DATETIME_FORMAT))
+      self.redis_client.set_attribute(request_key, 'last_update', last_update)
+      statuses_to_remove = [
+          'successful_tasks', 'failed_tasks', 'running_tasks', 'queued_tasks'
+      ]
+      # 'successful' could be None or False, which means different things.
+      # If False, the task has failed, If None, could be queued or running.
+      if hasattr(task.result, 'successful'):
+        if task.result.successful:
+          self.redis_client.add_to_list(
+              request_key, 'successful_tasks', task.id)
+          statuses_to_remove.remove('successful_tasks')
+        elif task.result.successful is False:
+          self.redis_client.add_to_list(request_key, 'failed_tasks', task.id)
+          statuses_to_remove.remove('failed_tasks')
+      task_status = self.redis_client.get_attribute(task_key, 'status')
+      if task_status == 'running':
+        self.redis_client.add_to_list(request_key, 'running_tasks', task.id)
+        statuses_to_remove.remove('running_tasks')
+      elif task_status is None or task_status == 'queued':
+        self.redis_client.add_to_list(request_key, 'queued_tasks', task.id)
+        statuses_to_remove.remove('queued_tasks')
+      for status_name in statuses_to_remove:
+        self.redis_client.remove_from_list(request_key, status_name, task.id)
+    except RedisClientError as exception:
+      error_message = f'Error encoding key {request_key} in Redis'
       log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
-    try:
-      for key in keys:
-        yield key.decode()
-    except ValueError as exception:
-      error_message = 'Error decoding key in Redis'
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
 
-  def iterate_attributes(self, key: str) -> Iterator[tuple]:
-    """Iterates over the attribute names of the Redis hash object.
+  def write_new_task(self, task) -> Optional[str]:
+    """Writes task into redis.
 
     Args:
-      key (str): The key of the stored hash object.
+      task_dict (dict[str]): A dictionary containing the serialized
+        request attributes that will be saved.
 
-    Yields:
-      attribute_name (tuple): Decoded name of object attribute.
-
-    Raises:
-      TurbiniaException: If Redis fails in getting the attributes or if
-        decode or json loads fails. 
+    Returns:
+      task_key Optional[str]: The key corresponding for the task.
     """
+    log.info(f'Writing metadata for new task {task.name:s} with id {task.id:s}')
     try:
-      attributes = self.client.hscan_iter(key, count=100)
-    except redis.RedisError as exception:
-      error_message = f'Error getting attributes from {key} in Redis'
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
+      task_key = self.redis_client.build_key_name('task', task.id)
+    except ValueError as exception:
+      log.error(exception)
     try:
-      for attribute in attributes:
-        yield (attribute[0].decode(), json.loads(attribute[1]))
-    except (TypeError, ValueError) as exception:
-      error_message = f'Error decoding attribute in {key} in Redis'
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
-
-  def key_exists(self, redis_key) -> bool:
-    """Checks if the key is saved in Redis.
+      task_dict = self.update_task_helper(task)
+      if task_key := self.redis_client.write_hash_object(task_key, task_dict):
+        task.state_key = task_key
+        self.update_request_task(task)
+      return task_key
+    except RedisClientError as exception:
+      log.error(f'Error writing task {task.id} data: {exception}')
+    except TurbiniaException as exception:
+      log.error(f'Error decoding task {task.id} metadata: {exception}')
+
+  def update_task_helper(self, task) -> Dict[str, Any]:
+    """Retrieves TurbiniaTask metadata to update time-related and status
+    attributes.
 
     Args:
-      key (str): The key to be checked.
+      task (TurbiniaTask): A TurbiniaTask object.
 
     Returns:
-      exists (bool): Boolean indicating if evidence is saved. 
-
-    Raises:
-      TurbiniaException: If Redis fails in checking the existence of the key.
+      task_dict: A dictionary containing updated task metadata.
     """
-    try:
-      return self.client.exists(redis_key)
-    except redis.RedisError as exception:
-      error_message = f'Error checking existence of {redis_key} in Redis'
-      log.error(f'{error_message}: {exception}')
-      raise TurbiniaException(error_message) from exception
-
-  def write_hash_object(self, redis_key, object_dict):
-    """Writes new hash object into redis. To save storage, the function does not
-    write values that are null, empty lists or empty dictionaries. Thus, if the 
-    value is deserialized from Redis into the original object, the default 
-    values will be used for those attributes.
+    task_dict = self.get_task_dict(task)
+    task_dict['last_update'] = task_dict['last_update'].strftime(
+        DATETIME_FORMAT)
+    task_dict['start_time'] = task_dict['start_time'].strftime(DATETIME_FORMAT)
+    if task_dict['run_time']:
+      task_dict['run_time'] = task_dict['run_time'].total_seconds()
+    for key, value in task_dict.items():
+      try:
+        task_dict[key] = json.dumps(value)
+      except (TypeError, ValueError) as exception:
+        error_message = (
+            f'Error serializing attribute {key}:{value} for task {task.id}.')
+        log.error(f'{error_message}: {exception}')
+        raise TurbiniaException(error_message) from exception
+    return task_dict
 
+  def update_task(self, task) -> Optional[str]:
+    """Updates a Turbinia task key.
+    
     Args:
-      object_dict (dict[str]): A dictionary containing the serialized
-      attributes that will be saved.
+      task: A TurbiniaTask object.
 
     Returns:
-      redis_key (str): The key corresponding to the object in Redis
+      task_key: The task key associated with this TurbiniaTask
+        or None.
     """
-    log.info(f'Writing hash object {redis_key} into Redis')
-    for attribute_name, attribute_value in object_dict.items():
-      if attribute_value not in EMPTY_JSON_VALUES:
-        self.set_attribute(redis_key, attribute_name, attribute_value)
-
-  def write_evidence(self, evidence_dict: dict[str], update=False) -> str:
+    task.touch()
+    task_key = task.state_key
+    if not task_key:
+      # if the task does not have a state_key we will write a new object.
+      if task_key := self.write_new_task(task):
+        return task_key
+      return None
+    try:
+      log.debug(f'Updating metadata for task {task.name} with key {task.id}')
+      task_dict = self.update_task_helper(task)
+      self.redis_client.write_hash_object(task_key, task_dict)
+      # Add the task to the associated TurbiniaReqest task_ids list.
+      self.update_request_task(task)
+      # Set the current status for the TurbiniaRequest
+      request_status = self.get_request_status(task.request_id)
+      request_key = self.redis_client.build_key_name('request', task.request_id)
+      self.redis_client.set_attribute(
+          request_key, 'status', json.dumps(request_status))
+    except TurbiniaException as exception:
+      log.error(f'Error uupdating task {task.id}: {exception}')
+    except RedisClientError as exception:
+      log.error(f'Error writing task data for task {task.id}: {exception}')
+    return task_key
+
+  def write_evidence(self, evidence_dict: dict[str]) -> str:
     """Writes evidence into redis.
 
     Args:
       evidence_dict (dict[str]): A dictionary containing the serialized
         evidence attributes that will be saved.
-      update (bool): Allows overwriting previous key and blocks writing new 
-        ones.
 
     Returns:
       evidence_key (str): The key corresponding to the evidence in Redis
@@ -431,21 +403,37 @@ def write_evidence(self, evidence_dict: dict[str], update=False) -> str:
     Raises:
       TurbiniaException: If the attribute deserialization fails.
     """
+    evidence_key = ''
+    request_key = ''
     try:
-      evidence_key = ':'.join(
-          ('TurbiniaEvidence', json.loads(evidence_dict['id'])))
+      evidence_id = json.loads(evidence_dict['id'])
       evidence_hash = json.loads(evidence_dict.get('hash'))
+      request_id = json.loads(evidence_dict['request_id'])
+      request_key = self.redis_client.build_key_name('request', request_id)
+      evidence_key = self.redis_client.build_key_name('evidence', evidence_id)
     except (TypeError, ValueError) as exception:
       error_message = 'Error deserializing evidence attribute.'
       log.error(f'{error_message}: {exception}')
       raise TurbiniaException(error_message) from exception
-    # Either updates or write new key
-    if update == self.key_exists(evidence_key):
-      self.write_hash_object(evidence_key, evidence_dict)
-      if evidence_hash:
-        self.set_attribute(
-            'TurbiniaEvidenceHashes', evidence_hash, evidence_key)
+
+    # Don't keep the config value since we don't really use it
+    # and it can be quite verbose if a complex recipe is used.
+    try:
+      evidence_dict.pop('config')
+    except KeyError:
+      # Nothing to do if the key doesn't exist.
+      pass
+
+    try:
+      if not self.redis_client.key_exists(evidence_key):
+        self.redis_client.write_hash_object(evidence_key, evidence_dict)
+        if evidence_hash:
+          self.redis_client.set_attribute(
+              'TurbiniaEvidenceHashes', evidence_hash, evidence_key)
+      self.redis_client.add_to_list(request_key, 'evidence_ids', evidence_id)
       return evidence_key
+    except RedisClientError as exception:
+      log.error(f'There was an error writing data to Redis: {exception}')
 
   def get_evidence_data(self, evidence_id: str) -> dict:
     """Returns a dictionary representing an Evidence object given its ID.
@@ -458,7 +446,7 @@ def get_evidence_data(self, evidence_id: str) -> dict:
     """
     evidence_key = ':'.join(('TurbiniaEvidence', evidence_id))
     evidence_dict = {}
-    for attribute_name, attribute_value in self.iterate_attributes(
+    for attribute_name, attribute_value in self.redis_client.iterate_attributes(
         evidence_key):
       evidence_dict[attribute_name] = attribute_value
     return evidence_dict
@@ -476,9 +464,9 @@ def get_evidence_summary(
       summary (dict | list | int): Object containing evidences. 
     """
     if output == 'count' and not group:
-      return sum(1 for _ in self.iterate_keys('Evidence'))
+      return sum(1 for _ in self.redis_client.iterate_keys('Evidence'))
     summary = {} if group else []
-    for evidence_key in self.iterate_keys('Evidence'):
+    for evidence_key in self.redis_client.iterate_keys('Evidence'):
       evidence_dictionary = self.get_evidence_data(evidence_key.split(':')[1])
       stored_value = evidence_dictionary if output == 'content' else (
           evidence_key)
@@ -508,8 +496,9 @@ def query_evidence(
       query_result (list | int): Result of the query. 
     """
     keys = []
-    for evidence_key in self.iterate_keys('Evidence'):
-      if stored_value := self.get_attribute(evidence_key, attribute_name):
+    for evidence_key in self.redis_client.iterate_keys('Evidence'):
+      if stored_value := self.redis_client.get_attribute(evidence_key,
+                                                         attribute_name):
         if (attribute_name == 'tasks' and attribute_value
             in stored_value) or stored_value == attribute_value or str(
                 stored_value) == str(attribute_value):
@@ -531,7 +520,7 @@ def get_evidence_key_by_hash(self, file_hash: str) -> str | None:
     """
     try:
       if file_hash:
-        return self.get_attribute(
+        return self.redis_client.get_attribute(
             'TurbiniaEvidenceHashes', file_hash, decode_json=False)
     except TurbiniaException:
       return None
@@ -547,3 +536,117 @@ def get_evidence_by_hash(self, file_hash: str) -> dict:
     """
     evidence_id = self.get_evidence_key_by_hash(file_hash).split(':')[1]
     return self.get_evidence_data(evidence_id)
+
+  def write_request(self, request_dict: dict, overwrite=False):
+    """Writes request into redis.
+
+    Args:
+      request_dict (dict[str]): A dictionary containing the serialized
+        request attributes that will be saved.
+      overwrite (bool): Allows overwriting previous key and blocks writing new 
+        ones.
+
+    Returns:
+      request_key (str): The key corresponding to the evidence in Redis
+    
+    Raises:
+      TurbiniaException: If the attribute deserialization fails or tried to
+          overwrite an existing key without overwrite=True
+    """
+    try:
+      request_id = json.loads(request_dict['request_id'])
+      request_key = self.redis_client.build_key_name('request', request_id)
+    except (TypeError, ValueError) as exception:
+      error_message = 'Error deserializing request attribute.'
+      log.error(f'{error_message}: {exception}')
+      raise TurbiniaException(error_message) from exception
+    try:
+      if not request_dict.get('start_time'):
+        request_dict['start_time'] = json.dumps(
+            datetime.now().strftime(DATETIME_FORMAT))
+      if not request_dict.get('last_update'):
+        request_dict['last_update'] = json.dumps(
+            datetime.now().strftime(DATETIME_FORMAT))
+      # Either updates or write new key
+      if not overwrite and self.redis_client.key_exists(request_key):
+        raise TurbiniaException(
+            f'Error, attempted to overwrite an existing key: {request_key} '
+            f'but overwrite was not set.')
+      else:
+        self.redis_client.write_hash_object(request_key, request_dict)
+    except redis.RedisError as exception:
+      error_message = f'Error encoding key {request_key} in Redis'
+      log.error(f'{error_message}: {exception}')
+      raise TurbiniaException(error_message) from exception
+
+    return request_key
+
+  def get_request_data(self, request_id: str) -> dict:
+    """Returns a dictionary representing a Request object given its ID.
+
+    Args:
+      request_id (str): The ID of the stored request.
+
+    Returns:
+      request_dict (dict): Dict containing request attributes. 
+    """
+    request_key = self.redis_client.build_key_name('request', request_id)
+    request_dict = {}
+    try:
+      for (
+          attribute_name,
+          attribute_value) in self.redis_client.iterate_attributes(request_key):
+        request_dict[attribute_name] = attribute_value
+      request_dict['last_update'] = datetime.strptime(
+          request_dict.get('last_update'), DATETIME_FORMAT)
+      request_dict['start_time'] = datetime.strptime(
+          request_dict.get('start_time'), DATETIME_FORMAT)
+    except TurbiniaException as exception:
+      log.error(f'Error retrieving request data: {exception}')
+    return request_dict
+
+  def get_request_status(self, request_id):
+    request_data = self.get_request_data(request_id)
+    request_status = 'pending'
+    finished_tasks = len(request_data['failed_tasks']) + (
+        len(request_data['successful_tasks']))
+    all_tasks_finished = finished_tasks == len(request_data['task_ids'])
+    if len(request_data['task_ids']) == len(request_data['successful_tasks']):
+      request_status = 'successful'
+    elif len(request_data['task_ids']) == len(request_data['failed_tasks']):
+      request_status = 'failed'
+    elif len(request_data['running_tasks']) > 0:
+      request_status = 'running'
+    elif len(request_data['failed_tasks']) > 0 and all_tasks_finished:
+      request_status = 'completed_with_errors'
+    else:
+      request_status = 'pending'
+    log.info(f'Request {request_id} status: {request_status}')
+    return request_status
+
+  def query_requests(
+      self, attribute_name: str, attribute_value: Any,
+      output: str = 'keys') -> list | int:
+    """Queries for requests with the specified attribute attribute_value.
+
+    Args:
+      attribute_name (str): Name of the attribute to be queried.
+      attribute_value (Any): Value stored in the attribute.
+      output (str): Output of the function (keys | content | count).
+
+    Returns:
+      query_result (list | int): Result of the query. 
+    """
+    keys = []
+    for request_key in self.redis_client.iterate_keys('request'):
+      if stored_value := self.redis_client.get_attribute(request_key,
+                                                         attribute_name):
+        if (attribute_name in ('evidence_ids', 'task_ids') and attribute_value
+            in stored_value) or stored_value == attribute_value or str(
+                stored_value) == str(attribute_value):
+          keys.append(request_key)
+    if output == 'content':
+      return [self.get_request_data(key.split(':')[1]) for key in keys]
+    elif output == 'count':
+      return len(keys)
+    return keys
diff --git a/turbinia/task_manager.py b/turbinia/task_manager.py
index 5370fd01c..81774031e 100644
--- a/turbinia/task_manager.py
+++ b/turbinia/task_manager.py
@@ -14,9 +14,8 @@
 # limitations under the License.
 """Task manager for Turbinia."""
 
-from __future__ import unicode_literals, absolute_import
-
 import logging
+from copy import deepcopy
 from datetime import datetime
 import time
 
@@ -46,8 +45,22 @@
 # Job/Task timeout value before it times out a given Task. This is to make sure
 # that the Server doesn't time out the Task before the Worker has a chance to
 # and should account for the Task scheduling and setup time that happens before
-# the Task starts.
-SERVER_TASK_TIMEOUT_BUFFER = 86400
+# the Task starts.  This time will be measured from the time the task is
+# enqueue'd, not from when it actually starts on the worker so if there is a
+# long wait for tasks to be executed they could potentially be timed out before
+# even getting a chance to start so this limit is set conservatively high.
+SERVER_TASK_TIMEOUT_BUFFER = 14400  # 4hr
+# Amount of buffer time to give between task timeout and the celery soft timeout
+# as we'd prefer for the task to timeout itself if possible so it has the most
+# control over setting the correct results.  This should be caught in the
+# `TurbiniaTask.run_wrapper()`.
+CELERY_SOFT_TIMEOUT_BUFFER = 120
+# Buffer time between task timeout and the hard celery timeout.  The hard
+# timeout cannot be caught by the worker so we want to give the task timeout and
+# soft timeout a chance for a graceful exit before falling back to this.
+# Because the worker is killed it will not send any results back to the server
+# and the server will have to time out the task there.
+CELERY_HARD_TIMEOUT_BUFFER = 240
 
 # Define metrics
 turbinia_server_tasks_total = Counter(
@@ -177,7 +190,7 @@ def abort_request(self, request_id, requester, evidence_name, message):
     continue, an AbortTask will be created with the error message and is written
     directly to the state database. This way the client will get a reasonable
     error in response to the failure.
-    
+
     Args:
       request_id(str): The request ID.
       requester(str): The username of the requester.
@@ -407,18 +420,19 @@ def add_task(self, task, job, evidence_):
 
     evidence_.config = job.evidence.config
     task.evidence_name = evidence_.name
+    task.evidence_id = evidence_.id
     task.base_output_dir = config.OUTPUT_DIR
     task.requester = evidence_.config.get('globals', {}).get('requester')
     task.group_name = evidence_.config.get('globals', {}).get('group_name')
     task.reason = evidence_.config.get('globals', {}).get('reason')
-    task.all_args = evidence_.config.get('globals', {}).get('all_args')
     task.group_id = evidence_.config.get('globals', {}).get('group_id')
     if job:
       task.job_id = job.id
       task.job_name = job.name
       job.tasks.append(task)
     self.state_manager.write_new_task(task)
-    self.enqueue_task(task, evidence_)
+    timeout_limit = jobs_manager.JobsManager.GetTimeoutValue(task.job_name)
+    self.enqueue_task(task, evidence_, timeout_limit)
     turbinia_server_tasks_total.inc()
     if task.id not in evidence_.tasks:
       evidence_.tasks.append(task.id)
@@ -456,12 +470,13 @@ def remove_job(self, job_id):
       turbinia_jobs_completed_total.inc()
     return bool(remove_job)
 
-  def enqueue_task(self, task, evidence_):
+  def enqueue_task(self, task, evidence_, timeout_limit):
     """Enqueues a task and evidence in the implementation specific task queue.
 
     Args:
       task: An instantiated Turbinia Task
       evidence_: An Evidence object to be processed.
+      timeout_limit(int): The timeout for the Task in seconds.
     """
     raise NotImplementedError
 
@@ -477,16 +492,13 @@ def process_result(self, task_result):
 
     Args:
       task_result: The TurbiniaTaskResult object
-
-    Returns:
-      TurbiniaJob|None: The Job for the processed task, else None
     """
     if task_result.successful is None:
       log.error(
-          f'''Task {task_result.task_name} from {task_result.worker_name}  
-          returned invalid success status "None". Setting this to False 
-          so the client knows the Task is complete. Usually this means 
-          that the Task returning the TurbiniaTaskResult did not call 
+          f'''Task {task_result.task_name} from {task_result.worker_name}
+          returned invalid success status "None". Setting this to False
+          so the client knows the Task is complete. Usually this means
+          that the Task returning the TurbiniaTaskResult did not call
           the close() method on it.
         ''')
       turbinia_result_success_invalid.inc()
@@ -497,17 +509,18 @@ def process_result(self, task_result):
 
     if not task_result.successful:
       log.error(
-          f'Task {task_result.task_name} from {task_result.worker_name} '
-          f'was not successful')
+          f'Task {task_result.task_id} {task_result.task_name} '
+          f'from {task_result.worker_name} was not successful')
     else:
       log.info(
-          f'Task {task_result.task_name} from {task_result.worker_name} '
-          f'executed with status [{task_result.status}]')
+          f'Task {task_result.task_id} {task_result.task_name} '
+          f'from {task_result.worker_name} executed with status [{task_result.status}]'
+      )
 
     if not isinstance(task_result.evidence, list):
       log.warning(
-          f'Task {task_result.task_name} from {task_result.worker_name} '
-          f'did not return evidence list')
+          f'Task {task_result.task_id} {task_result.task_name} '
+          f'from {task_result.worker_name}did not return evidence list')
       task_result.evidence = []
 
     job = self.get_job(task_result.job_id)
@@ -531,8 +544,6 @@ def process_result(self, task_result):
             f'Task {task_result.task_name} from {task_result.worker_name} '
             f'returned non-Evidence output type {type(task_result.evidence)}')
 
-    return job
-
   def process_job(self, job, task):
     """Processes the Job after Task completes.
 
@@ -545,7 +556,6 @@ def process_job(self, job, task):
       task (TurbiniaTask): The Task that just completed.
     """
     log.debug(f'Processing Job {job.name:s} for completed Task {task.id:s}')
-    self.state_manager.update_task(task)
     job.remove_task(task.id)
     turbinia_server_tasks_completed_total.inc()
     if job.check_done() and not (job.is_finalize_job or task.is_finalize_task):
@@ -587,15 +597,16 @@ def run(self, under_test=False):
 
       for task in self.process_tasks():
         if task.result:
-          job = self.process_result(task.result)
-          if job:
-            self.process_job(job, task)
+          self.process_result(task.result)
+        job = self.get_job(task.job_id)
+        if job:
+          self.process_job(job, task)
+        else:
+          log.warning(
+              f'Received task results for unknown Job {task.job_id} from Task '
+              f'ID {task.id:s}')
         self.state_manager.update_task(task)
 
-      if config.SINGLE_RUN and self.check_done():
-        log.info('No more tasks to process.  Exiting now.')
-        return
-
       if under_test:
         break
 
@@ -702,7 +713,8 @@ def get_evidence(self):
     requests = self.kombu.check_messages()
     evidence_list = []
     for request in requests:
-      #todo(igormr): Create request object in redis
+      self.state_manager.write_request(
+          deepcopy(request.to_json(json_values=True)))
       for evidence_ in request.evidence:
         if not evidence_.request_id:
           evidence_.request_id = request.request_id
@@ -720,7 +732,6 @@ def get_evidence(self):
           evidence_.config['globals']['requester'] = request.requester
           evidence_.config['globals']['group_name'] = request.group_name
           evidence_.config['globals']['reason'] = request.reason
-          evidence_.config['globals']['all_args'] = request.all_args
 
           # A recipe could contain a group_id key so that tasks can be grouped
           # together, but this is optional. If the recipe doesn't specify a
@@ -737,9 +748,20 @@ def get_evidence(self):
 
     return evidence_list
 
-  def enqueue_task(self, task, evidence_):
+  def enqueue_task(self, task, evidence_, timeout):
     log.info(
-        f'Adding Celery task {task.name:s} with evidence  {evidence_.name:s}'
-        f' to queue')
-    task.stub = self.celery_runner.delay(
-        task.serialize(), evidence_.serialize())
+        f'Adding Celery task {task.name:s} with evidence {evidence_.name:s}'
+        f' to queue with base task timeout {timeout}')
+    # https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-time-limit
+    # Hard limit in seconds, the worker processing the task will be killed and
+    # replaced with a new one when this is exceeded.
+    celery_soft_timeout = timeout + CELERY_SOFT_TIMEOUT_BUFFER
+    celery_hard_timeout = timeout + CELERY_HARD_TIMEOUT_BUFFER
+    self.celery_runner.max_retries = 0
+    self.celery_runner.task_time_limit = celery_hard_timeout
+    # Time limits described here:
+    #     https://docs.celeryq.dev/en/stable/userguide/workers.html#time-limits
+    task.stub = self.celery_runner.apply_async(
+        (task.serialize(), evidence_.serialize()), retry=False,
+        soft_time_limit=celery_soft_timeout, time_limit=celery_hard_timeout,
+        expires=celery_hard_timeout)
diff --git a/turbinia/tcelery.py b/turbinia/tcelery.py
index 0fdf390a6..5eb741e9f 100644
--- a/turbinia/tcelery.py
+++ b/turbinia/tcelery.py
@@ -19,14 +19,11 @@
 
 # fix celery naming collision (this file vs. module)
 from __future__ import absolute_import
-from __future__ import unicode_literals
 
 import logging
-
-from six.moves import queue
-
 import celery
 import kombu
+
 from kombu.exceptions import OperationalError
 from amqp.exceptions import ChannelError
 
@@ -102,7 +99,7 @@ def check_messages(self):
           requests.append(request)
           if self.queue.queue.durable:
             message.ack()
-      except queue.Empty:
+      except self.queue.Empty:
         break
       except ChannelError:
         break
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index 6ae55e130..1824f9e19 100644
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -204,8 +204,7 @@ def main():
   try:
     process_args(sys.argv[1:])
   except TurbiniaException as exception:
-    log.error(f'There was a problem processing arguments: {str(exception):s}')
-    sys.exit(1)
+    log.error(f'{str(exception):s}')
   log.info('Done.')
   sys.exit(0)
 
diff --git a/turbinia/worker.py b/turbinia/worker.py
index f63e2835b..9728a54c3 100644
--- a/turbinia/worker.py
+++ b/turbinia/worker.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Client objects for Turbinia."""
 
-from __future__ import unicode_literals
-
 import logging
 import os
 import stat
@@ -251,7 +249,7 @@ def start(self):
     # no apparent benefit from having this enabled at the moment.
     self.worker.task(task_utils.task_runner, name='task_runner')
     argv = [
-        'worker', '--loglevel=info', '--pool=solo', '--without-gossip',
+        'worker', '--loglevel=info', '--concurrency=1', '--without-gossip',
         '--without-mingle'
     ]
     self.worker.start(argv)
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index 3f8749970..a2e223414 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Turbinia task."""
 
-from __future__ import unicode_literals
-
 from copy import deepcopy
 from datetime import datetime
 from datetime import timedelta
@@ -27,6 +25,7 @@
 import pickle
 import platform
 import pprint
+import signal
 import subprocess
 import sys
 import tempfile
@@ -46,6 +45,8 @@
 from turbinia import TurbiniaException
 from turbinia import log_and_report
 
+from celery.exceptions import SoftTimeLimitExceeded
+
 METRICS = {}
 # Set the maximum size that the report can be before truncating it.  This is a
 # best effort estimate and not a guarantee and comes from the limit for
@@ -57,6 +58,9 @@
 log = logging.getLogger(__name__)
 
 registry = CollectorRegistry()
+turbinia_worker_exception_failure = Counter(
+    'turbinia_worker_exception_failure',
+    'Total number Tasks failed due to uncaught exception', registry=registry)
 turbinia_worker_tasks_started_total = Counter(
     'turbinia_worker_tasks_started_total',
     'Total number of started worker tasks', registry=registry)
@@ -71,7 +75,12 @@
     registry=registry)
 turbinia_worker_tasks_timeout_total = Counter(
     'turbinia_worker_tasks_timeout_total',
-    'Total number of worker tasks timed out.', registry=registry)
+    'Total number of worker tasks timed out during dependency execution.',
+    registry=registry)
+turbinia_worker_tasks_timeout_celery_soft = Counter(
+    'turbinia_worker_tasks_timeout_celery_soft',
+    'Total number of Tasks timed out due to Celery soft timeout',
+    registry=registry)
 
 
 class Priority(IntEnum):
@@ -95,6 +104,7 @@ class TurbiniaTaskResult:
       output_dir (str): Full path for local output
       error (dict): Error data ('error' and 'traceback' are some valid keys)
       evidence (list[Evidence]): Newly created Evidence objects.
+      evidence_size (int): Size of evidence in bytes.
       id (str): Unique Id of result (string of hex)
       input_evidence (Evidence): The evidence this task processed.
       job_id (str): The ID of the Job that generated this Task/TaskResult
@@ -118,7 +128,7 @@ class TurbiniaTaskResult:
   # The list of attributes that we will persist into storage
   STORED_ATTRIBUTES = [
       'worker_name', 'report_data', 'report_priority', 'run_time', 'status',
-      'saved_paths', 'successful'
+      'saved_paths', 'successful', 'evidence_size'
   ]
 
   def __init__(
@@ -135,12 +145,10 @@ def __init__(
     self.job_id = job_id
     self.base_output_dir = base_output_dir
     self.request_id = request_id
-
     self.task_id = None
     self.task_name = None
     self.requester = None
     self.output_dir = None
-
     self.report_data = None
     self.report_priority = Priority.MEDIUM
     self.run_time = None
@@ -234,7 +242,7 @@ def close(self, task, success, status=None):
             'The text_data attribute has a size {0:d} larger than the max '
             'size {1:d} so truncating the response.'.format(
                 len(evidence.text_data), REPORT_MAXSIZE))
-        self.log(message, logging.WARN)
+        self.log(message, logging.WARNING)
         evidence.text_data = evidence.text_data[:REPORT_MAXSIZE] + '\n'
         evidence.text_data += message
 
@@ -273,7 +281,7 @@ def close(self, task, success, status=None):
             log.error(f'Error updating evidence in redis: {exception}')
           else:
             self.state_manager.write_evidence(
-                evidence.serialize(json_values=True), update=True)
+                evidence.serialize(json_values=True))
 
     # Now that we've post-processed the input_evidence, we can unset it
     # because we don't need to return it.
@@ -310,8 +318,8 @@ def log(self, message, level=logging.INFO, traceback_=None):
       log.debug(message)
     elif level == logging.INFO:
       log.info(message)
-    elif level == logging.WARN:
-      log.warn(message)
+    elif level == logging.WARNING:
+      log.warning(message)
     elif level == logging.ERROR:
       log.error(message)
     elif level == logging.CRITICAL:
@@ -320,23 +328,6 @@ def log(self, message, level=logging.INFO, traceback_=None):
     if traceback_:
       self.set_error(message, traceback_)
 
-  def update_task_status(self, task, status=None):
-    """Updates the task status and pushes it directly to datastore.
-
-    Args:
-      task (TurbiniaTask): The calling Task object
-      status (str): Brief word or phrase for Task state. If not supplied, the
-          existing Task status will be used.
-    """
-    if status:
-      task.result.status = 'Task {0!s} is {1!s} on {2!s}'.format(
-          self.task_name, status, self.worker_name)
-    if self.state_manager:
-      self.state_manager.update_task(task)
-    else:
-      self.log(
-          'No state_manager initialized, not updating Task info', logging.DEBUG)
-
   def add_evidence(self, evidence, evidence_config):
     """Populate the results list.
 
@@ -406,7 +397,7 @@ def deserialize(cls, input_dict):
     result = TurbiniaTaskResult()
     result.__dict__.update(input_dict)
     if result.state_manager:
-      result.state_manager = None
+      result.state_manager = state_manager.get_state_manager()
     if result.run_time:
       result.run_time = timedelta(seconds=result.run_time)
     if result.input_evidence:
@@ -420,8 +411,14 @@ class TurbiniaTask:
   """Base class for Turbinia tasks.
 
   Attributes:
+      _evidence_config (dict): The config that we want to pass to all new
+          evidence created from this task.
       base_output_dir (str): The base directory that output will go into.
           Per-task directories will be created under this.
+      evidence (list): List of Evidence objects.
+      evidence_size (int): The size of the evidence.
+      group_id (str): group id for the evidence
+      group_name (str): group name for the evidence
       id (str): Unique Id of task (string of hex)
       is_finalize_task (bool): Whether this is a finalize Task or not.
       job_id (str): Job ID the Task was created by.
@@ -432,31 +429,30 @@ class TurbiniaTask:
       output_dir (str): The directory output will go into (including per-task
           folder).
       output_manager (OutputManager): The object that manages saving output.
-      result (TurbiniaTaskResult): A TurbiniaTaskResult object.
+      reason (str): reason of the evidence
+      recipe (dict): Validated recipe to be used as the task configuration.
       request_id (str): The id of the initial request to process this evidence.
+      requester (str): The user who requested the task.
+      result (TurbiniaTaskResult): A TurbiniaTaskResult object.
       start_time (datetime): When the task was started
       state_key (str): A key used to manage task state
-      stub (psq.task.TaskResult|celery.app.Task): The task manager
-          implementation specific task stub that exists server side to keep a
-          reference to the remote task objects.  For PSQ this is a task result
-          object, but other implementations have their own stub objects.
+      state_manager (state_manager): Turbinia state manager object.
+      stub (celery.result.AsyncResult): The task manager implementation
+          specific task stub that exists server side to keep a reference to the
+          remote task objects.
+      task_config (dict): Default task configuration, in effect if no recipe is
+          explicitly provided for the task.
       tmp_dir (str): Temporary directory for Task to write to.
-      requester (str): The user who requested the task.
-      _evidence_config (dict): The config that we want to pass to all new
-            evidence created from this task.
-      recipe (dict): Validated recipe to be used as the task configuration.
-      task_config (dict): Default task configuration, in effect if
-            no recipe is explicitly provided for the task.
-      group_name (str): group name for the evidence
-      reason (str): reason of the evidence
-      all_args (str): Terminal arguments input by user for evidence
-  """
+      turbinia_version (str): The version of Turbinia that was used to run the
+          task.
+      worker_start_time (datetime): The time the worker started the task.
+      """
 
   # The list of attributes that we will persist into storage
   STORED_ATTRIBUTES = [
       'id', 'job_id', 'start_time', 'last_update', 'name', 'evidence_name',
-      'evidence_id', 'evidence_size', 'request_id', 'requester', 'group_name',
-      'reason', 'all_args', 'group_id'
+      'evidence_id', 'request_id', 'requester', 'group_name', 'reason',
+      'group_id'
   ]
 
   # The list of evidence states that are required by a Task in order to run.
@@ -469,7 +465,7 @@ class TurbiniaTask:
 
   def __init__(
       self, name=None, base_output_dir=None, request_id=None, requester=None,
-      group_name=None, reason=None, all_args=None, group_id=None):
+      group_name=None, reason=None, group_id=None):
     """Initialization for TurbiniaTask.
 
     Args:
@@ -491,9 +487,9 @@ def __init__(
     self.name = name if name else self.__class__.__name__
     self.evidence_name = None
     self.evidence_id = None
-    self.evidence_size = None
     self.output_dir = None
     self.output_manager = output_manager.OutputManager()
+    self.state_manager = state_manager.get_state_manager()
     self.result = None
     self.request_id = request_id
     self.state_key = None
@@ -508,8 +504,8 @@ def __init__(
     self.task_config = {}
     self.group_name = group_name
     self.reason = reason
-    self.all_args = all_args
     self.group_id = group_id
+    self.worker_name = platform.node()
 
   def serialize(self):
     """Converts the TurbiniaTask object into a serializable dict.
@@ -517,6 +513,7 @@ def serialize(self):
     Returns:
       Dict: Dictionary representing this object, ready to be serialized.
     """
+    self.state_manager = None
     task_copy = deepcopy(self.__dict__)
     task_copy['output_manager'] = self.output_manager.__dict__
     task_copy['last_update'] = self.last_update.strftime(DATETIME_FORMAT)
@@ -542,6 +539,7 @@ def check_worker_role(cls):
     Returns:
       bool: If the current execution is in a worker or nosetests.
     """
+    config.LoadConfig()
     if config.TURBINIA_COMMAND in ('celeryworker', 'psqworker'):
       return True
 
@@ -563,8 +561,8 @@ def evidence_setup(self, evidence):
     evidence.validate()
     evidence.preprocess(
         self.id, tmp_dir=self.tmp_dir, required_states=self.REQUIRED_STATES)
+    self.evidence_name = evidence.name
     self.evidence_id = evidence.id
-    self.evidence_size = evidence.size
 
     # Final check to make sure that the required evidence state has been met
     # for Evidence types that have those capabilities.
@@ -644,6 +642,7 @@ def execute(
       Tuple of the return code, and the TurbiniaTaskResult object
     """
     # Avoid circular dependency.
+    import psutil
     from turbinia.jobs import manager as job_manager
 
     save_files = save_files if save_files else []
@@ -691,13 +690,31 @@ def execute(
               cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, cwd=cwd,
               env=env, text=True, encoding="utf-8")
           stdout, stderr = proc.communicate(timeout=timeout_limit)
-      except subprocess.TimeoutExpired as exception:
+      except (subprocess.TimeoutExpired, SoftTimeLimitExceeded) as exception:
+        # Catching the celery soft time limit here in addition to in the
+        # `run_wrapper()` so we can allow this except block to clean up the
+        # child processes appropriately when we are in this method.
+        if isinstance(exception, SoftTimeLimitExceeded):
+          timeout_type = 'celery soft'
+          turbinia_worker_tasks_timeout_celery_soft.inc()
+        else:
+          timeout_type = 'subprocess'
+        result.log(
+            'Job {0:s} with Task {1:s} has reached {2:s} timeout limit of '
+            '{3:d} so killing child processes.'.format(
+                self.job_id, self.id, timeout_type, timeout_limit))
+        # Kill child processes and parent process so we can return, otherwise
+        # communicate() will hang waiting for the grand-children to be reaped.
+        psutil_proc = psutil.Process(proc.pid)
+        for child in psutil_proc.children(recursive=True):
+          child.send_signal(signal.SIGKILL)
         proc.kill()
         # Get any potential partial output so we can save it later.
         stdout, stderr = proc.communicate()
         fail_message = (
-            'Execution of [{0!s}] failed due to job timeout of '
-            '{1:d} seconds has been reached.'.format(cmd, timeout_limit))
+            'Execution of [{0!s}] failed due to {1:s} timeout of '
+            '{2:d} seconds has been reached.'.format(
+                cmd, timeout_type, timeout_limit))
         result.log(fail_message)
         # Increase timeout metric. Not re-raising an exception so we can save
         # any potential output.
@@ -877,7 +894,7 @@ def create_result(
 
   def check_serialization_errors(self, result):
     """Checks the TurbiniaTaskResult is valid for serialization.
-    
+
     This method checks the 'result'' object is the correct type and whether
     it is pickle/JSON serializable or not.
 
@@ -1019,10 +1036,11 @@ def run_wrapper(self, evidence):
     from turbinia.jobs import manager as job_manager
 
     log.debug(f'Task {self.name:s} {self.id:s} awaiting execution')
+    failure_message = None
     try:
       evidence = evidence_decode(evidence)
       self.result = self.setup(evidence)
-      self.result.update_task_status(self, 'queued')
+      self.update_task_status(self, 'queued')
       turbinia_worker_tasks_queued_total.inc()
       task_runtime_metrics = self.get_metrics()
     except TurbiniaException as exception:
@@ -1062,6 +1080,7 @@ def run_wrapper(self, evidence):
           return self.result.serialize()
 
         self.evidence_setup(evidence)
+        self.result.evidence_size = evidence.size
 
         if config.VERSION_CHECK:
           if self.turbinia_version != __version__:
@@ -1073,33 +1092,42 @@ def run_wrapper(self, evidence):
             self.result.successful = False
             return self.result.serialize()
 
-        self.result.update_task_status(self, 'running')
         self._evidence_config = evidence.config
         self.task_config = self.get_task_recipe(evidence.config)
         self.worker_start_time = datetime.now()
+        self.update_task_status(self, 'running')
         self.result = self.run(evidence, self.result)
 
       # pylint: disable=broad-except
+      except SoftTimeLimitExceeded as exception:
+        failure_message = (
+            f'{self.name:s} Task timed out via Celery soft limit: {exception}')
+        if self.result:
+          self.result.log(failure_message, level=logging.ERROR)
+        else:
+          log.error(failure_message)
+        turbinia_worker_tasks_timeout_celery_soft.inc()
+
       except Exception as exception:
-        message = (f'{self.name:s} Task failed with exception: [{exception!s}]')
+        failure_message = (
+            f'{self.name:s} Task failed with exception: [{exception!s}]')
         # Logging explicitly here because the result is in an unknown state
         trace = traceback.format_exc()
-        log_and_report(message, trace)
+        log_and_report(failure_message, trace)
 
         if self.result:
-          self.result.log(message, level=logging.ERROR)
+          self.result.log(failure_message, level=logging.ERROR)
           self.result.log(trace)
           if hasattr(exception, 'message'):
             self.result.set_error(exception.message, traceback.format_exc())
           else:
             self.result.set_error(exception.__class__, traceback.format_exc())
-          self.result.status = message
+          self.result.status = failure_message
         else:
           log.error('No TurbiniaTaskResult object found after task execution.')
+        turbinia_worker_exception_failure.inc()
 
     self.result = self.validate_result(self.result)
-    if self.result:
-      self.result.update_task_status(self)
 
     # Trying to close the result if possible so that we clean up what we can.
     # This has a higher likelihood of failing because something must have gone
@@ -1113,20 +1141,22 @@ def run_wrapper(self, evidence):
         status = self.result.status
       else:
         status = 'No previous status'
-      message = (
-          'Task Result was auto-closed from task executor on {0:s} likely '
-          'due to previous failures.  Previous status: [{1:s}]'.format(
-              self.result.worker_name, status))
-      self.result.log(message)
+      # Failure message can be set during previous exception handling.
+      if not failure_message:
+        failure_message = (
+            'Task Result was auto-closed from task executor on {0:s} likely '
+            'due to previous failures.  Previous status: [{1:s}]'.format(
+                self.result.worker_name, status))
+      self.result.log(failure_message)
       try:
-        self.result.close(self, False, message)
+        self.result.close(self, False, failure_message)
       # Using broad except here because lots can go wrong due to the reasons
       # listed above.
       # pylint: disable=broad-except
       except Exception as exception:
         log.error(f'TurbiniaTaskResult close failed: {exception!s}')
         if not self.result.status:
-          self.result.status = message
+          self.result.status = failure_message
       # Check the result again after closing to make sure it's still good.
       self.result = self.validate_result(self.result)
 
@@ -1152,3 +1182,24 @@ def run(self, evidence, result):
         TurbiniaTaskResult object.
     """
     raise NotImplementedError
+
+  def update_task_status(self, task, status=None):
+    """Updates the task status and pushes it directly to datastore.
+
+    Args:
+      task (TurbiniaTask): The calling Task object
+      status (str): Brief word or phrase for Task state. If not supplied, the
+          existing Task status will be used.
+    """
+    if status:
+      task.status = 'Task {0!s} is {1!s} on {2!s}'.format(
+          self.name, status, self.worker_name)
+    if not self.state_manager:
+      self.state_manager = state_manager.get_state_manager()
+    if self.state_manager:
+      task_key = self.state_manager.redis_client.build_key_name('task', task.id)
+      self.state_manager.redis_client.set_attribute(
+          task_key, 'status', json.dumps(status))
+      self.state_manager.update_request_task(task)
+    else:
+      log.info('No state_manager initialized, not updating Task info')
diff --git a/turbinia/workers/analysis/jenkins.py b/turbinia/workers/analysis/jenkins.py
index fbe9acf8a..3bd90689d 100644
--- a/turbinia/workers/analysis/jenkins.py
+++ b/turbinia/workers/analysis/jenkins.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for analysing Jenkins."""
 
-from __future__ import unicode_literals
-
 import os
 import re
 
diff --git a/turbinia/workers/analysis/jupyter.py b/turbinia/workers/analysis/jupyter.py
index ad7f3c820..690e9bbb5 100644
--- a/turbinia/workers/analysis/jupyter.py
+++ b/turbinia/workers/analysis/jupyter.py
@@ -14,13 +14,8 @@
 # limitations under the License.
 """Task for analysing Jupyter."""
 
-from __future__ import unicode_literals
-
 import os
-import re
 
-from turbinia import TurbiniaException
-from turbinia.evidence import EvidenceState as state
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import TurbiniaTask
diff --git a/turbinia/workers/analysis/redis.py b/turbinia/workers/analysis/redis.py
index 98b16dd2f..564bf9570 100644
--- a/turbinia/workers/analysis/redis.py
+++ b/turbinia/workers/analysis/redis.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for analysing redis_config files."""
 
-from __future__ import unicode_literals
-
 import os
 import re
 
diff --git a/turbinia/workers/analysis/sshd.py b/turbinia/workers/analysis/sshd.py
index 130b88283..ab5a3cbe4 100644
--- a/turbinia/workers/analysis/sshd.py
+++ b/turbinia/workers/analysis/sshd.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for analysing sshd_config files."""
 
-from __future__ import unicode_literals
-
 import os
 import re
 
diff --git a/turbinia/workers/analysis/tomcat.py b/turbinia/workers/analysis/tomcat.py
index 5abb1e06d..591cc5353 100644
--- a/turbinia/workers/analysis/tomcat.py
+++ b/turbinia/workers/analysis/tomcat.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for analysing Tomcat files."""
 
-from __future__ import unicode_literals
-
 import os
 import re
 
diff --git a/turbinia/workers/analysis/wordpress_access.py b/turbinia/workers/analysis/wordpress_access.py
index c6f4e25cf..c305906a7 100644
--- a/turbinia/workers/analysis/wordpress_access.py
+++ b/turbinia/workers/analysis/wordpress_access.py
@@ -14,13 +14,10 @@
 # limitations under the License.
 """Task for analysing Wordpress access logs."""
 
-from __future__ import unicode_literals
-
 import gzip
 import os
 import re
 
-from turbinia.evidence import EvidenceState as state
 from turbinia.evidence import ReportText
 from turbinia.lib import text_formatter as fmt
 from turbinia.workers import Priority
diff --git a/turbinia/workers/artifact.py b/turbinia/workers/artifact.py
index 2323f90cf..849e80bbd 100644
--- a/turbinia/workers/artifact.py
+++ b/turbinia/workers/artifact.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for running Plaso."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import config
diff --git a/turbinia/workers/binary_extractor.py b/turbinia/workers/binary_extractor.py
index 387a70659..898eb6498 100644
--- a/turbinia/workers/binary_extractor.py
+++ b/turbinia/workers/binary_extractor.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task to extract binary files from an evidence object provided."""
 
-from __future__ import unicode_literals
-
 import logging
 import json
 import os
diff --git a/turbinia/workers/containerd.py b/turbinia/workers/containerd.py
index a586cd98b..6cc32ff02 100644
--- a/turbinia/workers/containerd.py
+++ b/turbinia/workers/containerd.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for analyzing containerd containers."""
 
-from __future__ import unicode_literals
-
 import json
 import logging
 import subprocess
diff --git a/turbinia/workers/dfdewey.py b/turbinia/workers/dfdewey.py
index 9a2ed0d4b..f25575222 100644
--- a/turbinia/workers/dfdewey.py
+++ b/turbinia/workers/dfdewey.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for running dfDewey."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import config
diff --git a/turbinia/workers/docker.py b/turbinia/workers/docker.py
index f99dff35e..7f9f920b1 100644
--- a/turbinia/workers/docker.py
+++ b/turbinia/workers/docker.py
@@ -13,8 +13,6 @@
 # limitations under the License.
 """Task for running docker-explorer."""
 
-from __future__ import unicode_literals
-
 import json
 import logging
 from os import path
diff --git a/turbinia/workers/file_system_timeline.py b/turbinia/workers/file_system_timeline.py
index 260d0775d..78be1ed9f 100644
--- a/turbinia/workers/file_system_timeline.py
+++ b/turbinia/workers/file_system_timeline.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task to run dfimagetools FileEntryLister on disk partitions."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import TurbiniaException
diff --git a/turbinia/workers/finalize_request.py b/turbinia/workers/finalize_request.py
index 4168a0967..60c5c8c81 100644
--- a/turbinia/workers/finalize_request.py
+++ b/turbinia/workers/finalize_request.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task to finalize the request."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import config
diff --git a/turbinia/workers/fsstat.py b/turbinia/workers/fsstat.py
index b26f796fb..85b8abb8f 100644
--- a/turbinia/workers/fsstat.py
+++ b/turbinia/workers/fsstat.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Task to run fsstat on disk partitions."""
-from __future__ import unicode_literals
 
 import os
 
diff --git a/turbinia/workers/grep.py b/turbinia/workers/grep.py
index afce8e8bf..11a45f169 100644
--- a/turbinia/workers/grep.py
+++ b/turbinia/workers/grep.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task to filter a text file using extended regular expression patterns."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia.evidence import FilteredTextFile
diff --git a/turbinia/workers/hindsight.py b/turbinia/workers/hindsight.py
index 2850d34fe..030af52eb 100644
--- a/turbinia/workers/hindsight.py
+++ b/turbinia/workers/hindsight.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for executing hindsight."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia.evidence import TextFile
diff --git a/turbinia/workers/photorec.py b/turbinia/workers/photorec.py
index 768fbe05f..4b9aa8b6a 100644
--- a/turbinia/workers/photorec.py
+++ b/turbinia/workers/photorec.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Task to run Photorec on disk images and retrieve deleted files ."""
-from __future__ import unicode_literals
 
 import os
 
diff --git a/turbinia/workers/plaso.py b/turbinia/workers/plaso.py
index d882fb368..254f03eea 100644
--- a/turbinia/workers/plaso.py
+++ b/turbinia/workers/plaso.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for running Plaso."""
 
-from __future__ import unicode_literals
-
 import os
 import logging
 
diff --git a/turbinia/workers/psort.py b/turbinia/workers/psort.py
index dfa3d5847..8fa62d407 100644
--- a/turbinia/workers/psort.py
+++ b/turbinia/workers/psort.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for running Psort."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import config
diff --git a/turbinia/workers/strings.py b/turbinia/workers/strings.py
index 92588fef3..70ea092b9 100644
--- a/turbinia/workers/strings.py
+++ b/turbinia/workers/strings.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for gathering ascii strings."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia.evidence import EvidenceState as state
diff --git a/turbinia/workers/volatility.py b/turbinia/workers/volatility.py
index b74a54ded..44cd13408 100644
--- a/turbinia/workers/volatility.py
+++ b/turbinia/workers/volatility.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Task for executing volatility."""
 
-from __future__ import unicode_literals
-
 import os
 
 from turbinia import config
diff --git a/turbinia/workers/worker_stat.py b/turbinia/workers/worker_stat.py
index fe92c6c1e..b0b2deabf 100644
--- a/turbinia/workers/worker_stat.py
+++ b/turbinia/workers/worker_stat.py
@@ -12,14 +12,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Task for running a test evidence stat call on the supplied evidence.
-
-TODO(aarontp): In the future we can use this for doing a count and healthcheck
-               of all PSQ workers since there is currently no mechanism for that
-               in PSQ.
-"""
-
-from __future__ import unicode_literals
+"""Task for running a test evidence stat call on the supplied evidence."""
 
 import os
 
diff --git a/web/package-lock.json b/web/package-lock.json
index 5f3281055..815e96f27 100644
--- a/web/package-lock.json
+++ b/web/package-lock.json
@@ -10,6 +10,7 @@
       "dependencies": {
         "@mdi/font": "7.0.96",
         "axios": "^1.6.5",
+        "streamsaver": "^2.0.6",
         "vue": "^3.2.0",
         "vuetify": "^3.0.0"
       },
@@ -1189,6 +1190,17 @@
         "node": ">=0.10.0"
       }
     },
+    "node_modules/streamsaver": {
+      "version": "2.0.6",
+      "resolved": "https://registry.npmjs.org/streamsaver/-/streamsaver-2.0.6.tgz",
+      "integrity": "sha512-LK4e7TfCV8HzuM0PKXuVUfKyCB1FtT9L0EGxsFk5Up8njj0bXK8pJM9+Wq2Nya7/jslmCQwRK39LFm55h7NBTw==",
+      "funding": [
+        {
+          "type": "github",
+          "url": "https://github.com/sponsors/jimmywarting"
+        }
+      ]
+    },
     "node_modules/to-regex-range": {
       "version": "5.0.1",
       "resolved": "https://registry.npmjs.org/to-regex-range/-/to-regex-range-5.0.1.tgz",
diff --git a/web/package.json b/web/package.json
index 0e94e102c..69901d572 100644
--- a/web/package.json
+++ b/web/package.json
@@ -9,6 +9,7 @@
   "dependencies": {
     "@mdi/font": "7.0.96",
     "axios": "^1.6.5",
+    "streamsaver": "^2.0.6",
     "vue": "^3.2.0",
     "vuetify": "^3.0.0"
   },
diff --git a/web/src/components/RequestList.vue b/web/src/components/RequestList.vue
index cf0cd1b45..f86faae95 100644
--- a/web/src/components/RequestList.vue
+++ b/web/src/components/RequestList.vue
@@ -31,9 +31,9 @@ limitations under the License.
         </v-text-field>
         <v-spacer></v-spacer>
       </v-card-title>
-      <v-data-table :headers="headers" :items="requestSummary" :search="search"
-        :footer-props="{ itemsPerPageOptions: [10, 20, 40, -1] }" multi-sort item-value="request_id_reason" show-expand
-        hover>
+      <v-data-table :headers="headers" :items="requestSummary" :search="search" density="compact"
+        item-value="request_id" :footer-props="{ itemsPerPageOptions: [10, 20, 40, -1] }" :loading="isLoading"
+        :sort-by="sortBy" show-expand hover>
         <template v-slot:[`item.status`]="{ item }">
           <div v-if="item.status === 'successful'">
             <v-tooltip text="Completed successfully">
@@ -104,11 +104,13 @@ export default {
   data() {
     return {
       search: '',
+      isLoading: false,
       headers: [
-        { title: 'Request', key: 'request_id_reason' },
-        { title: 'Last Task Update Time', key: 'last_task_update_time' },
+        { title: 'Request', key: 'request_id' },
+        { title: 'Last Task Update Time', key: 'last_task_update_time', sortable: true },
         { title: 'Evidence Name', key: 'evidence_name' },
         { title: 'Requester', key: 'requester' },
+        { title: 'Reason', key: 'request_id_reason' },
         { title: 'Total Tasks', key: 'total_tasks' },
         { title: 'Running Tasks', key: 'running_tasks' },
         { title: 'Successful Tasks', key: 'successful_tasks' },
@@ -117,6 +119,7 @@ export default {
         { title: 'Results', key: 'request_results' },
       ],
       requestSummary: [],
+      sortBy: [{ key: 'last_task_update_time', order: 'desc' }],
     }
   },
   methods: {
@@ -124,14 +127,21 @@ export default {
     getRequestList: function () {
       ApiClient.getRequestList()
         .then((response) => {
+          this.isLoading = true
           let requestSummary = []
           let data = response.data['requests_status']
           for (const req in data) {
             let outstanding_perc = Math.round(
               ((data[req].failed_tasks + data[req].successful_tasks) / data[req].task_count) * 100
             )
+            let reason = null
+            if (data[req].reason) {
+              reason = data[req].reason
+            } else {
+              reason = 'N/A'
+            }
             requestSummary.push({
-              request_id_reason: data[req].request_id + ' - ' + data[req].reason,
+              request_id_reason: reason,
               request_id: data[req].request_id,
               last_task_update_time: data[req].last_task_update_time,
               requester: data[req].requester,
@@ -141,10 +151,11 @@ export default {
               failed_tasks: data[req].failed_tasks,
               outstanding_perc: outstanding_perc,
               status: data[req].status,
-              evidence_name: data[req].evidence_name,
+              evidence_name: data[req].evidence_name.substring(0, 64) + '...',
               evidence_id: data[req].evidence_id,
             })
           }
+          this.isLoading = false
           this.requestSummary = requestSummary
         })
         .catch((e) => {
diff --git a/web/src/components/TaskDetails.vue b/web/src/components/TaskDetails.vue
index c74f2ade9..2a306361c 100644
--- a/web/src/components/TaskDetails.vue
+++ b/web/src/components/TaskDetails.vue
@@ -39,25 +39,40 @@ limitations under the License.
       {{ taskDetails.status }}
     </v-alert>
     <v-card>
-      <v-list v-model:opened="openGroups">
+      <v-list density="compact" v-model:opened="openGroups">
         <v-list-group value="ids">
           <template v-slot:activator="{ props }">
             <v-list-item v-bind="props" title="Associated IDs"></v-list-item>
           </template>
           <v-list-item title="Task ID:">
-            {{ taskDetails.id }}
+            <div v-if="taskDetails.id">
+              {{ taskDetails.id }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Request ID:">
-            {{ taskDetails.request_id }}
+            <div v-if="taskDetails.request_id">
+              {{ taskDetails.request_id }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Evidence ID:">
-            {{ taskDetails.evidence_id }}
+            <div v-if="taskDetails.evidence_id">
+              {{ taskDetails.evidence_id }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Group ID:">
-            {{ taskDetails.group_id }}
+            <div v-if="taskDetails.group_id">
+              {{ taskDetails.group_id }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Job ID:">
-            {{ taskDetails.job_id }}
+            <div v-if="taskDetails.job_id">
+              {{ taskDetails.job_id }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
         </v-list-group>
         <v-list-group value="details">
@@ -65,28 +80,46 @@ limitations under the License.
             <v-list-item v-bind="props" title="Processing Details"></v-list-item>
           </template>
           <v-list-item title="Evidence Name:">
-            {{ taskDetails.evidence_name }}
+            <div v-if="taskDetails.evidence_name">
+              {{ taskDetails.evidence_name }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Evidence Size:">
-            {{ taskDetails.evidence_size }} bytes
-          </v-list-item>
-          <v-list-item title="Arguments:">
-            {{ taskDetails.all_args }}
+            <div v-if="taskDetails.evidence_size">
+              {{ taskDetails.evidence_size }} bytes
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Last Update:">
-            {{ taskDetails.last_update }}
+            <div v-if="taskDetails.last_update">
+              {{ taskDetails.last_update }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Requester:">
-            {{ taskDetails.requester }}
+            <div v-if="taskDetails.requester">
+              {{ taskDetails.requester }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Worker:">
-            {{ taskDetails.worker_name }}
+            <div v-if="taskDetails.worker_name">
+              {{ taskDetails.worker_name }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Successful:">
-            {{ taskDetails.successful }}
+            <div v-if="taskDetails.successful">
+              {{ taskDetails.successful }}
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
           <v-list-item title="Run Time:">
-            {{ taskDetails.run_time }} seconds
+            <div v-if="taskDetails.run_time">
+              {{ taskDetails.run_time }} seconds
+            </div>
+            <div v-else>N/A</div>
           </v-list-item>
         </v-list-group>
         <div v-if="taskDetails.saved_paths">
diff --git a/web/src/components/TaskList.vue b/web/src/components/TaskList.vue
index f95840ed0..66fb99971 100644
--- a/web/src/components/TaskList.vue
+++ b/web/src/components/TaskList.vue
@@ -14,7 +14,7 @@ limitations under the License.
 <template>
   <div>
     <v-list density="compact">
-      <v-virtual-scroll :items="taskList" :item-height="40" :height="400">
+      <v-virtual-scroll :items="taskList" :item-height="40" :height="400" :width="auto">
         <template v-slot:default="{ item }">
           <v-list-item :key="item.task_id" v-slot:prepend>
             <div v-if="item.task_success">
@@ -30,21 +30,15 @@ limitations under the License.
             </div>
             <div>
               <v-list-item-action>
-                <v-btn
-                  variant="text"
-                  :ripple="true"
-                  :key="item.task_id"
-                  selected-class="activated"
+                <v-btn variant="text" :ripple="true" :key="item.task_id" selected-class="activated"
                   :class="{ activated: isActive == item.task_id }"
-                  @click="getTaskDetails(item.task_id) + selectActiveStatus(item.task_id)"
-                >
+                  @click="getTaskDetails(item.task_id) + selectActiveStatus(item.task_id)">
                   {{ item.task_id }}
                 </v-btn>
               </v-list-item-action>
             </div>
             <v-list-item>
-              {{ item.task_name }}
-              >: {{ item.task_status }}
+              {{ item.task_name }} {{ $filters.truncate(item.task_status, 128, '...') }}
             </v-list-item>
           </v-list-item>
           <v-divider> </v-divider>
diff --git a/web/src/main.js b/web/src/main.js
index d31eee220..a24269eea 100644
--- a/web/src/main.js
+++ b/web/src/main.js
@@ -17,4 +17,14 @@ const app = createApp(App)
 
 registerPlugins(app)
 
+app.config.globalProperties.$filters = {
+    truncate(text, length, suffix) {
+        if (text.length > length) {
+            return text.substring(0, length) + suffix;
+        } else {
+            return text;
+        }
+    }
+}
+
 app.mount('#app')


Test patch:
diff --git a/test_data/state_manager_test_data.json b/test_data/state_manager_test_data.json
index d20c0b9bf..37fa00f35 100644
--- a/test_data/state_manager_test_data.json
+++ b/test_data/state_manager_test_data.json
@@ -24,7 +24,6 @@
         "sketch_id": null,
         "group_name": "",
         "reason": "",
-        "all_args": "turbinia/turbiniactl.py rawdisk -l /workspaces/turbinia/test_data/artifact_disk.dd",
         "group_id": "55ce6e98dc154e73990b24f0c79ab07e",
         "requester": "root"
       }
@@ -70,7 +69,6 @@
         "sketch_id": null,
         "group_name": "",
         "reason": "",
-        "all_args": "turbinia/turbiniactl.py rawdisk -l /workspaces/turbinia/test_data/artifact_disk.dd",
         "group_id": "55ce6e98dc154e73990b24f0c79ab07e",
         "requester": "root"
       }
@@ -115,7 +113,6 @@
         "sketch_id": null,
         "group_name": "",
         "reason": "",
-        "all_args": "turbinia/turbiniactl.py rawdisk -l /workspaces/turbinia/test_data/artifact_disk.dd",
         "group_id": "55ce6e98dc154e73990b24f0c79ab07e",
         "requester": "root"
       }
diff --git a/turbinia/api/api_server_test.py b/turbinia/api/api_server_test.py
index db0ffa759..dde0b6828 100644
--- a/turbinia/api/api_server_test.py
+++ b/turbinia/api/api_server_test.py
@@ -30,24 +30,20 @@
 from turbinia.api.api_server import app
 from turbinia.api.routes.router import api_router
 from turbinia.api.routes.ui import ui_router
-from turbinia.api.cli.turbinia_client.helpers import formatter
 
 from turbinia import config as turbinia_config
 from turbinia import state_manager
 from turbinia.jobs import manager as jobs_manager
 from turbinia.workers import TurbiniaTask
 
-from textwrap import dedent
-
 
 class testTurbiniaAPIServer(unittest.TestCase):
   """ Test Turbinia API server."""
 
   _TASK_TEST_DATA = {
       'id': 'c8f73a5bc5084086896023c12c7cc026',
-      'evidence_name': '/evidence/test.tgz',
+      'evidence_name': 'test_data/artifact_disk.dd',
       'evidence_id': '084d5904f3d2412b99dc29ed34853a16',
-      'all_args': 'compresseddirectory -l /evidence/test.tgz',
       'job_id': '1db0dc47d8f244f5b4fa7e15b8a87861',
       'start_time': '2022-04-01T19:15:14.791074Z',
       'last_update': '2022-04-01T19:17:14.791074Z',
@@ -67,12 +63,62 @@ class testTurbiniaAPIServer(unittest.TestCase):
   }
 
   _REQUEST_TEST_DATA = {
-      'evidence_name': '/evidence/test.tgz',
+      'group_name': '',
+      'evidence_ids': ["084d5904f3d2412b99dc29ed34853a16"],
+      'status': 'successful',
+      'type': 'TurbiniaRequest',
+      'running_tasks': [],
+      'context': {},
+      'reason': '',
+      'task_ids': ["c8f73a5bc5084086896023c12c7cc026"],
+      'requester': 'root',
+      'last_update': '2022-04-01T19:17:14.791074Z',
+      'group_id': '1234',
+      'original_evidence': {
+          "id": "084d5904f3d2412b99dc29ed34853a16",
+          "name": "test_data/artifact_disk.dd"
+      },
+      'successful_tasks': ["c8f73a5bc5084086896023c12c7cc026"],
+      'recipe': {
+          "globals": {
+              "debug_tasks": False,
+              "jobs_allowlist": [],
+              "jobs_denylist": [],
+              "yara_rules": "",
+              "filter_patterns": [],
+              "sketch_id": '',
+              "group_name": "",
+              "reason": "",
+              "group_id": ""
+          }
+      },
+      'request_id': '41483253079448e59685d88f37ab91f7',
+      'start_time': '2022-04-01T19:15:14.791074Z',
+  }
+
+  _REQUEST_STATUS_TEST_DATA = {
+      'evidence_id': '084d5904f3d2412b99dc29ed34853a16',
+      'evidence_name': 'test_data/artifact_disk.dd',
+      'failed_tasks': 0,
+      'last_task_update_time': '2022-04-01T19:17:14.791074Z',
+      'queued_tasks': 0,
+      'reason': '',
+      'request_id': '41483253079448e59685d88f37ab91f7',
+      'requester': 'root',
+      'running_tasks': 0,
+      'status': 'successful',
+      'successful_tasks': 1,
+      'task_count': 1,
+      'tasks': ['c8f73a5bc5084086896023c12c7cc026']
+  }
+
+  _REQUEST_STATUS_SUMMARY_DATA = {
       'evidence_id': '084d5904f3d2412b99dc29ed34853a16',
+      'evidence_name': 'test_data/artifact_disk.dd',
       'failed_tasks': 0,
       'last_task_update_time': '2022-04-01T19:17:14.791074Z',
       'queued_tasks': 0,
-      'reason': None,
+      'reason': '',
       'request_id': '41483253079448e59685d88f37ab91f7',
       'requester': 'root',
       'running_tasks': 0,
@@ -83,15 +129,12 @@ class testTurbiniaAPIServer(unittest.TestCase):
   }
 
   _EVIDENCE_TEST_DATA = {
-      'request_id': '5581344e306b42ccb965a19028d4fc58',
-      'tasks': [
-          'b73d484634164e0eb1870d101ca9ce2f',
-          'dd810119ac2443e18b69ea56c10c0a9b', 'ac4dc14080b144478437818a694e2f4d'
-      ],
+      'request_id': '41483253079448e59685d88f37ab91f7',
+      'tasks': ['c8f73a5bc5084086896023c12c7cc026'],
       'copyable': False,
       'cloud_only': False,
-      'local_path': '/workspaces/turbinia/test_data/artifact_disk.dd',
-      'source_path': '/workspaces/turbinia/test_data/artifact_disk.dd',
+      'local_path': 'test_data/artifact_disk.dd',
+      'source_path': 'test_data/artifact_disk.dd',
       'resource_tracked': False,
       'processed_by': [],
       'resource_id': None,
@@ -106,8 +149,7 @@ class testTurbiniaAPIServer(unittest.TestCase):
               'sketch_id': None,
               'group_name': '',
               'reason': '',
-              'all_args': 'turbinia/turbiniactl.py rawdisk -l disk.dd',
-              'group_id': '55ce6e98dc154e73990b24f0c79ab07e',
+              'group_id': '1234',
               'requester': 'root'
           }
       },
@@ -121,10 +163,10 @@ class testTurbiniaAPIServer(unittest.TestCase):
       'has_child_evidence': False,
       'save_metadata': False,
       'type': 'RawDisk',
-      '_name': '/workspaces/turbinia/test_data/artifact_disk.dd',
+      '_name': 'test_data/artifact_disk.dd',
       'context_dependent': False,
       'state': {},
-      'id': 'b510ab6bf11a410da1fd9d9b128e7d74',
+      'id': '084d5904f3d2412b99dc29ed34853a16',
       'hash': '4cf679344af02c2b89e4a902f939f4608bcac0fbf81511da13d7d9b9',
       'description': None
   }
@@ -141,17 +183,13 @@ class testTurbiniaAPIServer(unittest.TestCase):
 
   _COUNT_SUMMARY = 3
 
-  def _get_state_manager(self):
-    """Gets a Redis State Manager object for test."""
-    turbinia_config.STATE_MANAGER = 'Redis'
-    # force state_manager module to reload using Redis state manager.
-    importlib.reload(state_manager)
-    return state_manager.get_state_manager()
-
-  def setUp(self):
-    """This method will write a temporary key to redis for testing purposes."""
+  @mock.patch('redis.StrictRedis')
+  @mock.patch('turbinia.state_manager.get_state_manager')
+  def setUp(self, _, mock_redis):
     self.client = TestClient(app)
-    self.state_manager = self._get_state_manager()
+    mock_redis = fakeredis.FakeStrictRedis()
+    self.state_manager = state_manager.get_state_manager()
+    self.state_manager.redis_client.client = mock_redis
 
   def testWebRoutes(self):
     """Test Web UI routes."""
@@ -173,7 +211,7 @@ def testGetConfig(self):
 
   def testRequestResultsNotFound(self):
     """Test getting empty request result files."""
-    request_id = self._REQUEST_TEST_DATA.get('request_id')
+    request_id = self._REQUEST_STATUS_TEST_DATA.get('request_id')
     response = self.client.get(f'/api/result/request/{request_id}')
     log_path = turbinia_config.toDict().get('OUTPUT_DIR')
     output_path = os.path.join(log_path, request_id)
@@ -215,83 +253,130 @@ def testGetTaskStatus(self, testTaskData):
     result = json.loads(result.content)
     self.assertEqual(expected_result_dict, result)
 
+  @mock.patch('turbinia.state_manager.RedisStateManager.get_request_data')
   @mock.patch('turbinia.state_manager.RedisStateManager.get_task_data')
-  def testRequestStatus(self, testTaskData):
+  @mock.patch('turbinia.redis_client.RedisClient.key_exists')
+  def testRequestStatus(self, testKeyExists, testTaskData, testRequestData):
     """Test getting Turbinia Request status."""
+    self.maxDiff = None
     redis_client = fakeredis.FakeStrictRedis()
     input_task = TurbiniaTask().deserialize(self._TASK_TEST_DATA)
     input_task_serialized = input_task.serialize()
-    expected_result = self._REQUEST_TEST_DATA.copy()
+    expected_result = self._REQUEST_STATUS_TEST_DATA.copy()
     expected_result['tasks'] = [input_task_serialized]
-    expected_result_dict = OrderedDict(sorted(expected_result.items()))
+    input_request = self._REQUEST_TEST_DATA
+    testKeyExists.return_value = True
+
+    redis_client.set(
+        'TurbiniaRequest:41483253079448e59685d88f37ab91f7',
+        json.dumps(input_request))
 
     redis_client.set(
-        'TurbiniaTask:41483253079448e59685d88f37ab91f7',
+        'TurbiniaTask:c8f73a5bc5084086896023c12c7cc026',
         json.dumps(input_task_serialized))
+
+    testRequestData.return_value = json.loads(
+        redis_client.get('TurbiniaRequest:41483253079448e59685d88f37ab91f7'))
+
     testTaskData.return_value = [
         json.loads(
-            redis_client.get('TurbiniaTask:41483253079448e59685d88f37ab91f7'))
+            redis_client.get('TurbiniaTask:c8f73a5bc5084086896023c12c7cc026'))
     ]
 
     result = self.client.get(
-        f"/api/request/{self._REQUEST_TEST_DATA.get('request_id')}")
+        f"/api/request/{self._REQUEST_STATUS_TEST_DATA.get('request_id')}")
     result = json.loads(result.content)
-    self.assertEqual(expected_result_dict, result)
+    self.assertEqual(expected_result, result)
 
+  @mock.patch('turbinia.state_manager.RedisStateManager.get_request_data')
   @mock.patch('turbinia.state_manager.RedisStateManager.get_task_data')
-  def testRequestSummary(self, testTaskData):
+  @mock.patch('turbinia.redis_client.RedisClient.key_exists')
+  @mock.patch('turbinia.redis_client.RedisClient.iterate_keys')
+  def testRequestSummary(
+      self, testIterateKeys, testKeyExists, testTaskData, testRequestData):
     """Test getting Turbinia Request status summary."""
+    self.maxDiff = None
     redis_client = fakeredis.FakeStrictRedis()
+    expected_result = self._REQUEST_STATUS_SUMMARY_DATA.copy()
+    expected_result = {'requests_status': [self._REQUEST_STATUS_SUMMARY_DATA]}
     input_task = TurbiniaTask().deserialize(self._TASK_TEST_DATA)
     input_task_serialized = input_task.serialize()
-    expected_result = {'requests_status': [self._REQUEST_TEST_DATA]}
+    input_request = self._REQUEST_TEST_DATA
+    testKeyExists.return_value = True
+    testIterateKeys.return_value = [
+        f'TurbiniaRequest:{self._REQUEST_TEST_DATA.get("request_id")}'
+    ]
+
+    redis_client.set(
+        'TurbiniaRequest:41483253079448e59685d88f37ab91f7',
+        json.dumps(input_request))
 
     redis_client.set(
-        'TurbiniaTask:41483253079448e59685d88f37ab91f7',
+        'TurbiniaTask:c8f73a5bc5084086896023c12c7cc026',
         json.dumps(input_task_serialized))
 
+    testRequestData.return_value = json.loads(
+        redis_client.get('TurbiniaRequest:41483253079448e59685d88f37ab91f7'))
+
     testTaskData.return_value = [
         json.loads(
-            redis_client.get('TurbiniaTask:41483253079448e59685d88f37ab91f7'))
+            redis_client.get('TurbiniaTask:c8f73a5bc5084086896023c12c7cc026'))
     ]
 
     result = self.client.get('/api/request/summary')
     result = json.loads(result.content)
     self.assertEqual(expected_result, result)
 
+  @mock.patch('turbinia.state_manager.RedisStateManager.get_request_data')
   @mock.patch('turbinia.state_manager.RedisStateManager.get_task_data')
-  def testRequestEvidenceNoArgs(self, testTaskData):
-    """Test getting Turbinia Request evidence name without all_args."""
-    redis_client = fakeredis.FakeStrictRedis()
+  @mock.patch('turbinia.redis_client.RedisClient.iterate_keys')
+  @mock.patch('turbinia.redis_client.RedisClient.key_exists')
+  def testRequestEvidenceNoArgs(
+      self, testKeyExists, testIterateKeys, testTaskData, testRequestData):
+    """Test getting Turbinia Request evidence name."""
+    redis_client = self.state_manager.redis_client.client
+    input_request = self._REQUEST_TEST_DATA
     input_task = TurbiniaTask().deserialize(self._TASK_TEST_DATA)
     input_task_serialized = input_task.serialize()
-    input_task_serialized.pop('all_args')
-    expected_result = self._REQUEST_TEST_DATA['evidence_name']
+    expected_result = self._REQUEST_STATUS_TEST_DATA['evidence_name']
+    testIterateKeys.return_value = [
+        f'TurbiniaRequest:{self._REQUEST_TEST_DATA.get("request_id")}'
+    ]
+    testKeyExists.return_value = True
+    redis_client.set(
+        'TurbiniaRequest:41483253079448e59685d88f37ab91f7',
+        json.dumps(input_request))
 
     redis_client.set(
-        'TurbiniaTask:41483253079448e59685d88f37ab91f7',
+        'TurbiniaTask:c8f73a5bc5084086896023c12c7cc026',
         json.dumps(input_task_serialized))
 
+    testRequestData.return_value = json.loads(
+        redis_client.get('TurbiniaRequest:41483253079448e59685d88f37ab91f7'))
+
     testTaskData.return_value = [
         json.loads(
-            redis_client.get('TurbiniaTask:41483253079448e59685d88f37ab91f7'))
+            redis_client.get('TurbiniaTask:c8f73a5bc5084086896023c12c7cc026'))
     ]
 
     result = self.client.get('/api/request/summary')
     result = json.loads(result.content)
     evidence_name = result['requests_status'][0]['evidence_name']
-
     self.assertEqual(expected_result, evidence_name)
 
+  @mock.patch('turbinia.state_manager.RedisStateManager.get_request_data')
   @mock.patch('turbinia.state_manager.RedisStateManager.get_task_data')
-  def testRequestNotFound(self, testTaskData):
+  @mock.patch('turbinia.redis_client.RedisClient.key_exists')
+  def testRequestNotFound(self, testKeyExists, testTaskData, testRequestData):
     """Test getting invalid Turbinia Request status."""
     expected_result = {
         'detail': 'Request ID not found or the request had no associated tasks.'
     }
     testTaskData.return_value = []
+    testRequestData.return_value = self._REQUEST_TEST_DATA
+    testKeyExists.return_value = False
     result = self.client.get(
-        f"/api/request/{self._REQUEST_TEST_DATA.get('request_id')}")
+        f"/api/request/{self._REQUEST_STATUS_TEST_DATA.get('request_id')}")
     result = json.loads(result.content)
     self.assertEqual(expected_result, result)
 
@@ -308,7 +393,7 @@ def testGetJobs(self):
     """Test getting Turbinia job names."""
     _jobs_manager = jobs_manager.JobsManager()
     registered_jobs = set(_jobs_manager.GetJobNames())
-    disabled_jobs = set(turbinia_config.CONFIG.DISABLED_JOBS)
+    disabled_jobs = set(turbinia_config.DISABLED_JOBS)
     expected_result = list(registered_jobs.difference(disabled_jobs))
 
     result = self.client.get('/api/jobs')
@@ -335,7 +420,7 @@ def testGetEvidenceNotFound(self, testGetEvidence):
         response.json(), {
             'detail':
                 f'UUID {evidence_id} not found or it had no associated '
-                f'evidences.'
+                f'evidence.'
         })
 
   @mock.patch('turbinia.api.routes.evidence.redis_manager.get_evidence_summary')
@@ -404,16 +489,14 @@ def testEvidenceQueryNotFound(self, testQueryEvidence):
                 f'request_id.')
         })
 
-  @mock.patch('turbinia.state_manager.RedisStateManager.get_attribute')
+  @mock.patch('turbinia.redis_client.RedisClient.get_attribute')
   @mock.patch('turbinia.api.routes.evidence.datetime')
-  @mock.patch('turbinia.api.routes.evidence.os.makedirs')
-  def testEvidenceUpload(
-      self, mock_makedirs, mock_datetime, mock_get_attribute):
+  def testEvidenceUpload(self, mock_datetime, mock_get_attribute):
     """Tests uploading evidence."""
     mocked_now = datetime.datetime.now()
     mock_datetime.now.return_value = mocked_now
     mocked_now_str = mocked_now.strftime(turbinia_config.DATETIME_FORMAT)
-    self.maxDiff = None
+    mock_get_attribute.return_value = None
     filedir = os.path.dirname(os.path.realpath(__file__))
     evidence_1_name = 'wordpress_access_logs.txt'
     evidence_2_name = 'mbr.raw'
diff --git a/turbinia/client_stats_test.py b/turbinia/client_stats_test.py
index 0dca007f8..f66ee8b55 100644
--- a/turbinia/client_stats_test.py
+++ b/turbinia/client_stats_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia client module."""
 
-from __future__ import unicode_literals
-
 from datetime import datetime
 from datetime import timedelta
 import unittest
@@ -85,4 +83,4 @@ def testTurbiniaStatsFormatStatsCsv(self):
     stats.add_task(test_task1)
     stats.calculate_stats()
     report = stats.format_stats_csv()
-    self.assertEqual(report, test_output)
\ No newline at end of file
+    self.assertEqual(report, test_output)
diff --git a/turbinia/client_test.py b/turbinia/client_test.py
index 18ffd37b9..4b4be9b63 100644
--- a/turbinia/client_test.py
+++ b/turbinia/client_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia client module."""
 
-from __future__ import unicode_literals
-
 from datetime import datetime
 from datetime import timedelta
 import importlib
diff --git a/turbinia/config/turbinia_config_test.py b/turbinia/config/turbinia_config_test.py
index 833cf546d..7a5fe3a66 100644
--- a/turbinia/config/turbinia_config_test.py
+++ b/turbinia/config/turbinia_config_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia config."""
 
-from __future__ import unicode_literals
-
 import os
 import tempfile
 import unittest
diff --git a/turbinia/evidence_test.py b/turbinia/evidence_test.py
index 24cc3ad17..73d92e374 100644
--- a/turbinia/evidence_test.py
+++ b/turbinia/evidence_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia evidence."""
 
-from __future__ import unicode_literals
-
 import json
 import mock
 import unittest
diff --git a/turbinia/jobs/imports_test.py b/turbinia/jobs/imports_test.py
index 76a39d606..034ea8392 100644
--- a/turbinia/jobs/imports_test.py
+++ b/turbinia/jobs/imports_test.py
@@ -15,8 +15,6 @@
 # -*- coding: utf-8 -*-
 """Tests that all jobs are imported correctly."""
 
-from __future__ import unicode_literals
-
 import io
 import re
 import os
diff --git a/turbinia/jobs/interface_test.py b/turbinia/jobs/interface_test.py
index e442833b5..f1ddff702 100644
--- a/turbinia/jobs/interface_test.py
+++ b/turbinia/jobs/interface_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia Jobs interface."""
 
-from __future__ import unicode_literals
-
 from turbinia import evidence
 from turbinia.jobs import interface
 from turbinia.workers.workers_test import TestTurbiniaTaskBase
diff --git a/turbinia/jobs/manager_test.py b/turbinia/jobs/manager_test.py
index e377d251c..f50b19326 100644
--- a/turbinia/jobs/manager_test.py
+++ b/turbinia/jobs/manager_test.py
@@ -2,8 +2,6 @@
 # -*- coding: utf-8 -*-
 """Tests for the job manager."""
 
-from __future__ import unicode_literals
-
 import unittest
 
 from turbinia import TurbiniaException
diff --git a/turbinia/lib/dockermanager_test.py b/turbinia/lib/dockermanager_test.py
index 391fc068f..796e9a4df 100644
--- a/turbinia/lib/dockermanager_test.py
+++ b/turbinia/lib/dockermanager_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for docker_manager module."""
 
-from __future__ import unicode_literals
-
 import unittest
 import codecs
 import mock
diff --git a/turbinia/lib/text_formatter_test.py b/turbinia/lib/text_formatter_test.py
index 7bfcb83e8..9d8454b2c 100644
--- a/turbinia/lib/text_formatter_test.py
+++ b/turbinia/lib/text_formatter_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for text_formatter."""
 
-from __future__ import unicode_literals
-
 import unittest
 
 from turbinia.lib import text_formatter as fmt
diff --git a/turbinia/output_manager_test.py b/turbinia/output_manager_test.py
index fc2018b2e..529805da6 100644
--- a/turbinia/output_manager_test.py
+++ b/turbinia/output_manager_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia OutputWriters."""
 
-from __future__ import unicode_literals
-
 import json
 import unittest
 import os
diff --git a/turbinia/processors/archive_test.py b/turbinia/processors/archive_test.py
index 30d73ce2e..3a37611c2 100644
--- a/turbinia/processors/archive_test.py
+++ b/turbinia/processors/archive_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Archive processor to compress and decompress folders."""
 
-from __future__ import unicode_literals
-
 import os
 import tarfile
 import unittest
diff --git a/turbinia/processors/mount_local_test.py b/turbinia/processors/mount_local_test.py
index f6fb91970..25a206248 100644
--- a/turbinia/processors/mount_local_test.py
+++ b/turbinia/processors/mount_local_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for mount_local processor."""
 
-from __future__ import unicode_literals
-
 import os
 from subprocess import CalledProcessError
 from subprocess import STDOUT
diff --git a/turbinia/processors/resource_manager_test.py b/turbinia/processors/resource_manager_test.py
index 37d4f7a70..afc46ff7c 100644
--- a/turbinia/processors/resource_manager_test.py
+++ b/turbinia/processors/resource_manager_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for resource_manager handler."""
 
-from __future__ import unicode_literals
-
 import unittest
 import os
 import shutil
diff --git a/turbinia/server_test.py b/turbinia/server_test.py
index 7086946f8..ba271763f 100644
--- a/turbinia/server_test.py
+++ b/turbinia/server_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia server module."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
diff --git a/turbinia/state_manager_test.py b/turbinia/state_manager_test.py
index 9b7ec5555..02f20c9fd 100644
--- a/turbinia/state_manager_test.py
+++ b/turbinia/state_manager_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests the state manager module."""
 
-from __future__ import unicode_literals
-
 import fakeredis
 import importlib
 import json
@@ -27,7 +25,7 @@
 from turbinia import state_manager
 
 
-class TestRedisEvidenceStateManager(unittest.TestCase):
+class TestRedisStateManager(unittest.TestCase):
   """Test RedisStateManager class."""
 
   def get_evidence_data(self):
@@ -61,13 +59,13 @@ def get_evidence_data(self):
   def write_evidence_in_fake_redis(self):
     for evidence_key, evidence_value in self.test_data.items():
       for attribute_name, attribute_value in evidence_value.items():
-        self.state_manager.client.hset(
+        self.state_manager.redis_client.client.hset(
             evidence_key, attribute_name, json.dumps(attribute_value))
 
   def get_data_from_fake_redis(self):
     for evidence_key, evidence_value in self.test_data.items():
       for attribute_name, attribute_value in evidence_value.items():
-        self.state_manager.client.hset(
+        self.state_manager.redis_client.client.hset(
             evidence_key, attribute_name, json.dumps(attribute_value))
 
   @mock.patch('redis.StrictRedis')
@@ -75,14 +73,12 @@ def get_data_from_fake_redis(self):
   def setUp(self, _, mock_redis):
     self.state_manager = None
     config.LoadConfig()
-    self.state_manager_save = config.STATE_MANAGER
-
     mock_redis = fakeredis.FakeStrictRedis()
     config.STATE_MANAGER = 'Redis'
     # force state_manager module to reload using Redis state manager.
     importlib.reload(state_manager)
     self.state_manager = state_manager.get_state_manager()
-    self.state_manager.set_client(mock_redis)
+    self.state_manager.redis_client.client = mock_redis
 
     self.get_evidence_data()
 
@@ -110,17 +106,10 @@ def testStateManagerWriteEvidence(self):
     self.state_manager.write_evidence(json_dumped_evidence)
 
     result = {}
-    for attribute_name, attribute_value in self.state_manager.client.hscan_iter(
-        evidence_key):
+    for attribute_name, attribute_value in (
+        self.state_manager.redis_client.client.hscan_iter(evidence_key)):
       result[attribute_name.decode()] = attribute_value.decode()
 
-    null_keys = []
-    for key, value in json_dumped_evidence.items():
-      if value in ('null', '[]', '{}'):
-        null_keys.append(key)
-    for key in null_keys:
-      json_dumped_evidence.pop(key)
-
     # Check if the stored evidence contains all of our test data
     self.assertEqual(result, json_dumped_evidence)
 
@@ -224,3 +213,17 @@ def testStateManagerEvidenceQueryCount(self):
 
     # Check if the returned count is equal to our test data
     self.assertEqual(result, self.grouped_count_summary[True])
+
+  def testBuildKeyName(self):
+    """Test State Manager build_key_name() method."""
+    key_types = ('evidence', 'task', 'request')
+    for key_type in key_types:
+      key_name = self.state_manager.redis_client.build_key_name(
+          key_type, '1234')
+      self.assertIsInstance(key_name, str)
+
+  def testBuildInvalidKeyName(self):
+    """Test State Manager build_key_name() method, invalid key type"""
+    self.assertRaises(
+        ValueError, self.state_manager.redis_client.build_key_name, 'wrong',
+        '1234')
diff --git a/turbinia/task_manager_test.py b/turbinia/task_manager_test.py
index a0f8a24de..2c7eb477b 100644
--- a/turbinia/task_manager_test.py
+++ b/turbinia/task_manager_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia task_manager module."""
 
-from __future__ import unicode_literals
-
 import mock
 
 from turbinia import TurbiniaException
@@ -212,7 +210,8 @@ def testFinalizeResult(self, _):
     self.result.evidence.append(self.evidence)
     self.manager.add_evidence = mock.MagicMock()
     self.manager.running_jobs.append(self.job1)
-    test_job = self.manager.process_result(self.result)
+    self.manager.process_result(self.result)
+    test_job = self.manager.get_job(self.job1.id)
     self.assertEqual(test_job.id, job_id)
     self.assertEqual(test_job, self.manager.running_jobs[0])
     self.assertEqual(test_job.evidence.collection[0], self.evidence)
@@ -248,7 +247,7 @@ def testFinalizeJobGenerateJobFinalizeTasks(self, _):
 
     self.job1.create_final_task.assert_called()
     self.manager.enqueue_task.assert_called()
-    test_task, test_evidence = self.manager.enqueue_task.call_args[0]
+    test_task, test_evidence, _ = self.manager.enqueue_task.call_args[0]
     # Make sure the evidence originally associated with the job is the same as
     # the new evidence collection.
     self.assertListEqual(test_evidence.collection, [self.evidence])
@@ -305,7 +304,8 @@ def testRun(self):
     self.manager.add_evidence = mock.MagicMock()
     self.task.result = self.result
     self.manager.process_tasks = mock.MagicMock(return_value=[self.task])
-    self.manager.process_result = mock.MagicMock(return_value=self.job1)
+    self.manager.process_result = mock.MagicMock()
+    self.manager.get_job = mock.MagicMock(return_value=self.job1)
     self.manager.process_job = mock.MagicMock()
     self.manager.run(under_test=True)
 
diff --git a/turbinia/task_utils_test.py b/turbinia/task_utils_test.py
index d3134e841..dd84e5af4 100644
--- a/turbinia/task_utils_test.py
+++ b/turbinia/task_utils_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia task_utils module."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
@@ -72,4 +70,4 @@ def testTaskRunner(self, mock_task_deserialize):
     mock_task_deserialize.return_value = task
     task_dict = task.serialize()
     ret = task_utils.task_runner(task_dict, 'testValue')
-    self.assertEqual(ret, 'testValue')
\ No newline at end of file
+    self.assertEqual(ret, 'testValue')
diff --git a/turbinia/worker_test.py b/turbinia/worker_test.py
index f7cd3374e..d474d0610 100644
--- a/turbinia/worker_test.py
+++ b/turbinia/worker_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for Turbinia worker module."""
 
-from __future__ import unicode_literals
-
 import unittest
 import os
 import shutil
diff --git a/turbinia/workers/analysis/jenkins_test.py b/turbinia/workers/analysis/jenkins_test.py
index cba724533..079d00871 100644
--- a/turbinia/workers/analysis/jenkins_test.py
+++ b/turbinia/workers/analysis/jenkins_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Jenkins analysis task."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
diff --git a/turbinia/workers/analysis/jupyter_test.py b/turbinia/workers/analysis/jupyter_test.py
index bbf407175..9a19fe949 100644
--- a/turbinia/workers/analysis/jupyter_test.py
+++ b/turbinia/workers/analysis/jupyter_test.py
@@ -1,5 +1,4 @@
 # -*- coding: utf-8 -*-
-from __future__ import unicode_literals
 
 import unittest
 
diff --git a/turbinia/workers/analysis/llm_analyzer_test.py b/turbinia/workers/analysis/llm_analyzer_test.py
index dab1ecaeb..9a6e70617 100644
--- a/turbinia/workers/analysis/llm_analyzer_test.py
+++ b/turbinia/workers/analysis/llm_analyzer_test.py
@@ -28,6 +28,7 @@ class LLMAnalyzerTaskTest(unittest.TestCase):
   @mock.patch("google.generativeai.GenerativeModel")
   def test_llm_analyze_artifact(self, mock_gen_model, mock_gen_config):
     config.LoadConfig()
+    config.LLM_PROVIDER = "vertexai"
     config.GCP_GENERATIVE_LANGUAGE_API_KEY = "fakeGcpApiKey"
     task = llm_analyzer.LLMAnalyzerTask()
     mock_gen_config.side_effect = None
diff --git a/turbinia/workers/analysis/redis_test.py b/turbinia/workers/analysis/redis_test.py
index 8852eeafc..a9b431f03 100644
--- a/turbinia/workers/analysis/redis_test.py
+++ b/turbinia/workers/analysis/redis_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Redis analysis task."""
 
-from __future__ import unicode_literals
-
 import unittest
 
 from turbinia import config
diff --git a/turbinia/workers/analysis/sshd_test.py b/turbinia/workers/analysis/sshd_test.py
index 51a23330b..51d99fac9 100644
--- a/turbinia/workers/analysis/sshd_test.py
+++ b/turbinia/workers/analysis/sshd_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the SSHD analysis task."""
 
-from __future__ import unicode_literals
-
 import unittest
 
 from turbinia import config
diff --git a/turbinia/workers/analysis/tomcat_test.py b/turbinia/workers/analysis/tomcat_test.py
index 0ed9d14ad..47ab680c1 100644
--- a/turbinia/workers/analysis/tomcat_test.py
+++ b/turbinia/workers/analysis/tomcat_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Tomcat analysis task."""
 
-from __future__ import unicode_literals
-
 import unittest
 
 from turbinia import config
diff --git a/turbinia/workers/analysis/wordpress_access_test.py b/turbinia/workers/analysis/wordpress_access_test.py
index f2ed5235b..f4ac4cb36 100644
--- a/turbinia/workers/analysis/wordpress_access_test.py
+++ b/turbinia/workers/analysis/wordpress_access_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Wordpress access log analysis task."""
 
-from __future__ import unicode_literals
-
 import os
 import unittest
 
diff --git a/turbinia/workers/binary_extractor_test.py b/turbinia/workers/binary_extractor_test.py
index 411c2f266..5cd584787 100644
--- a/turbinia/workers/binary_extractor_test.py
+++ b/turbinia/workers/binary_extractor_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Binary Extractor task."""
 
-from __future__ import unicode_literals
-
 import json
 import mock
 import os
diff --git a/turbinia/workers/bulk_extractor_test.py b/turbinia/workers/bulk_extractor_test.py
index 978e00214..2b3772ca4 100644
--- a/turbinia/workers/bulk_extractor_test.py
+++ b/turbinia/workers/bulk_extractor_test.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 """Tests for the Bulk Extractor job."""
 
-from __future__ import unicode_literals
 from shutil import rmtree
 from io import StringIO
 
diff --git a/turbinia/workers/containerd_test.py b/turbinia/workers/containerd_test.py
index 4c5072a83..135ca3065 100644
--- a/turbinia/workers/containerd_test.py
+++ b/turbinia/workers/containerd_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for ContainerdEnumerationTask."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
diff --git a/turbinia/workers/dfdewey_test.py b/turbinia/workers/dfdewey_test.py
index 01c191ebd..977d161e1 100644
--- a/turbinia/workers/dfdewey_test.py
+++ b/turbinia/workers/dfdewey_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the dfDewey job."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
diff --git a/turbinia/workers/docker_test.py b/turbinia/workers/docker_test.py
index 13eb70b1e..571663694 100644
--- a/turbinia/workers/docker_test.py
+++ b/turbinia/workers/docker_test.py
@@ -14,7 +14,6 @@
 # limitations under the License.
 """Tests for the Docker Containers Enumeration job."""
 
-from __future__ import unicode_literals
 from io import StringIO
 
 import unittest
diff --git a/turbinia/workers/file_system_timeline_test.py b/turbinia/workers/file_system_timeline_test.py
index b8a3c9593..92a891bbf 100644
--- a/turbinia/workers/file_system_timeline_test.py
+++ b/turbinia/workers/file_system_timeline_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the FileSystemTimelineJob job."""
 
-from __future__ import unicode_literals
-
 import unittest
 import os
 import mock
diff --git a/turbinia/workers/fsstat_test.py b/turbinia/workers/fsstat_test.py
index f8805decf..2232bdb89 100644
--- a/turbinia/workers/fsstat_test.py
+++ b/turbinia/workers/fsstat_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Fsstat job."""
 
-from __future__ import unicode_literals
-
 import unittest
 import mock
 
diff --git a/turbinia/workers/photorec_test.py b/turbinia/workers/photorec_test.py
index 31a5d7473..e07bf9b55 100644
--- a/turbinia/workers/photorec_test.py
+++ b/turbinia/workers/photorec_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for the Photorec job."""
 
-from __future__ import unicode_literals
-
 import os
 import unittest
 import textwrap
diff --git a/turbinia/workers/volatility_test.py b/turbinia/workers/volatility_test.py
index 1c1723f15..5eb9ecf20 100644
--- a/turbinia/workers/volatility_test.py
+++ b/turbinia/workers/volatility_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for VolatilityTask."""
 
-from __future__ import unicode_literals
-
 import os
 import mock
 
diff --git a/turbinia/workers/workers_test.py b/turbinia/workers/workers_test.py
index 39cefd97d..84ef013c3 100644
--- a/turbinia/workers/workers_test.py
+++ b/turbinia/workers/workers_test.py
@@ -14,8 +14,6 @@
 # limitations under the License.
 """Tests for workers __init__."""
 
-from __future__ import unicode_literals
-
 import json
 import os
 import tempfile
@@ -58,6 +56,7 @@ def setUp(self, task_class=TurbiniaTask, evidence_class=evidence.RawDisk):
         None, None)
     self.task = self.task_class(base_output_dir=self.base_output_dir)
     self.task.job_name = 'PlasoJob'
+    self.task.job_id = 'FakeJobID'
     self.task.output_manager = mock.MagicMock()
     self.task.output_manager.get_local_output_dirs.return_value = (None, None)
     self.task.get_metrics = mock.MagicMock()
@@ -106,7 +105,7 @@ def setResults(
 
     self.result.input_evidence = evidence.RawDisk()
     self.result.status = 'TestStatus'
-    self.result.update_task_status = mock.MagicMock()
+    self.update_task_status = mock.MagicMock()
     self.result.close = mock.MagicMock()
     self.task.setup = mock.MagicMock(return_value=setup)
     self.result.worker_name = 'worker1'
@@ -146,7 +145,8 @@ def testTurbiniaTaskSerialize(self):
     out_obj.output_manager = None
     self.assertEqual(out_obj.__dict__, self.plaso_task.__dict__)
 
-  def testTurbiniaTaskRunWrapper(self):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskRunWrapper(self, _):
     """Test that the run wrapper executes task run."""
     self.unregisterMetrics()
     self.setResults()
@@ -157,7 +157,8 @@ def testTurbiniaTaskRunWrapper(self):
     self.assertEqual(new_result.status, 'TestStatus')
     self.result.close.assert_not_called()
 
-  def testTurbiniaTaskRunWrapperAutoClose(self):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskRunWrapperAutoClose(self, _):
     """Test that the run wrapper closes the task."""
     self.unregisterMetrics()
     self.setResults()
@@ -167,7 +168,8 @@ def testTurbiniaTaskRunWrapperAutoClose(self):
     self.result.close.assert_called()
 
   @mock.patch('turbinia.state_manager.get_state_manager')
-  def testTurbiniaTaskRunWrapperBadResult(self, _):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskRunWrapperBadResult(self, _, __):
     """Test that the run wrapper recovers from run returning bad result."""
     self.unregisterMetrics()
     bad_result = 'Not a TurbiniaTaskResult'
@@ -181,7 +183,8 @@ def testTurbiniaTaskRunWrapperBadResult(self, _):
     self.assertEqual(type(new_result), TurbiniaTaskResult)
     self.assertIn('CheckedResult', new_result.status)
 
-  def testTurbiniaTaskJobUnavailable(self):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskJobUnavailable(self, _):
     """Test that the run wrapper can fail if the job doesn't exist."""
     self.unregisterMetrics()
     self.setResults()
@@ -193,7 +196,8 @@ def testTurbiniaTaskJobUnavailable(self):
     new_result = TurbiniaTaskResult.deserialize(new_result)
     self.assertEqual(new_result.status, canary_status)
 
-  def testTurbiniaTaskRunWrapperExceptionThrown(self):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskRunWrapperExceptionThrown(self, _):
     """Test that the run wrapper recovers from run throwing an exception."""
     self.unregisterMetrics()
     self.setResults()
@@ -206,7 +210,8 @@ def testTurbiniaTaskRunWrapperExceptionThrown(self):
 
   @mock.patch('turbinia.workers.TurbiniaTask.create_result')
   @mock.patch('turbinia.state_manager.get_state_manager')
-  def testTurbiniaTaskRunWrapperSetupFail(self, _, mock_create_result):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskRunWrapperSetupFail(self, _, __, mock_create_result):
     """Test that the run wrapper recovers from setup failing."""
     self.task.result = None
     canary_status = 'exception_message'
@@ -246,7 +251,8 @@ def testTurbiniaTaskValidateResultBadResult(self, _, __):
     self.assertNotEqual(new_result.error, {})
 
   @mock.patch('turbinia.workers.evidence_decode')
-  def testTurbiniaTaskEvidenceValidationFailure(self, evidence_decode_mock):
+  @mock.patch('turbinia.redis_client.RedisClient.set_attribute')
+  def testTurbiniaTaskEvidenceValidationFailure(self, _, evidence_decode_mock):
     """Tests Task fails when evidence validation fails."""
     self.setResults()
     test_evidence = evidence.RawDisk()