def get_task_by_id()

in source/client/python/api-v0.1/api/state_table_dynamodb.py [0:0]

• 21 lines of code
• 6 McCabe index (conditional complexity)

    def get_task_by_id(self, task_id, consistent_read=False):
        """
        Args:
            task_id - string
        Returns:
            Dictionary containing a single task (dynamodb row)
            An entire (raw) row from DynamoDB by task_id
        """

        try:
            response = self.state_table.query(
                KeyConditionExpression=Key('task_id').eq(task_id),
                Select='ALL_ATTRIBUTES',
                ConsistentRead=consistent_read
            )

            if ((response is not None) and (len(response['Items']) == 1)):
                return response.get('Items')[0]
            else:
                return None

        except ClientError as e:

            if e.response['Error']['Code'] in ["ThrottlingException", "ProvisionedThroughputExceededException"]:
                logging.warning(f"Could not read row for task [{task_id}] from Status Table. Exception: {e} [{traceback.format_exc()}]")
                return None

            else:
                logging.error(f"Could not read row for task [{task_id}] from Status Table. Exception: {e} [{traceback.format_exc()}]")
                raise Exception(e)

        except Exception as e:
            logging.error(f"Could not read row for task [{task_id}] from Status Table. Exception: {e} [{traceback.format_exc()}]")
            raise e