🔍 Code Extractor

A class for tracking and managing the status of study tasks through a Neo4j graph database, monitoring progress through a predefined sequence of workflow steps.

Purpose

The Tasklist class provides a comprehensive interface for managing study workflow status in a Neo4j database. It tracks a study's progress through 13 predefined tasks (from Quote to Final Report), queries the database to determine current status, and provides methods to retrieve numerical progress indicators. The class maintains both overall study status and detailed status information for each task, making it suitable for workflow management and progress tracking in laboratory or research study contexts.

Source Code

class Tasklist():
    """
    Class for gathering and tracking the list of studies
    
    Attributes
    ----------
    graph : Py2neo.Graph
        Database graph
    tasks : list
        An internal task lists, which defines all the mandatory tasks to go through, in linear order.
    study : str
        A string containing the unique ID of a study
    status : str
        The current status of the study, by default set to Finished
    numerated_status : int
        The current status in number format, i.e. step 3 out of 10
    max_tasks : int
        The total number of tasks in our current task list
    status_dict : dictionary
        A dictionary mapping each task to its current status
    details_status : dictionary
        A dictionary mapping teach task to a list of all its current statusses,
        for example: Grossing:['Pending','Pending','Finished','Finished']
        
    Parameters
    ----------
    study : str
        A string containing the unique ID of a study
    """
    tasks = ["Quote", "Study Plan", "Delivery", "Sample Registration", "Grossing", "Embedding", "Sectioning", "Staining", "Scanning", "Digital QA","Assessment","Draft Report","Final Report"]
    
    def __init__(self, study):
        # self.graph = Graph("bolt://neodev:7687",auth=('neo4j','innoser'),name='neo4j')
        self.graph = Graph(config.DB_ADDR, auth=config.DB_AUTH, name=config.DB_NAME)
        self.study = study
        self.status = 'Finished'
        self.numerated_status = int()
        self.max_tasks = len(self.tasks)+1
        self.status_dict = {i:"Pending" for i in self.tasks}
        self.detailed_status = {i:[] for i in self.tasks}
        self._get_status()
        
    def _get_status(self):
        for task in self.tasks:
            result = self.graph.run(f"MATCH (:Study {{UID:'{self.study}'}})-[*]->(t:Task {{T:'{task}'}}) RETURN COLLECT(t.T)").evaluate()
            if len(result) > 0:
                self.status = task
            else:
                self.status_dict[task] = 'Completed'
                continue
            break
            
    def numerate_status(self):
        try:
            self.numerated_status = self.tasks.index(self.status)+1
        except Exception as e:
            self.numerated_status = 10
        return self.numerated_status, self.max_tasks

Parameters

Parameter Details

study: A unique identifier string for the study being tracked. This UID is used to query the Neo4j database to retrieve task status information and must match the UID stored in the database's Study nodes.

Return Value

Instantiation returns a Tasklist object with initialized status tracking. The numerate_status() method returns a tuple of (numerated_status: int, max_tasks: int) representing the current task number and total number of tasks respectively.

Class Interface

Methods

Attributes

Dependencies

• py2neo
• datetime
• io
• panel
• param
• json
• config

Required Imports

from neo4j_driver import *
import datetime as dt
import io
import panel as pn
import param
import json
import config

Usage Example

# Ensure config.py has DB_ADDR, DB_AUTH, and DB_NAME configured
# Example: config.DB_ADDR = 'bolt://localhost:7687'
# config.DB_AUTH = ('neo4j', 'password')
# config.DB_NAME = 'neo4j'

from tasklist import Tasklist

# Instantiate with a study UID
tasklist = Tasklist(study='STUDY-12345')

# Check current status
print(f"Current status: {tasklist.status}")

# Get numerical progress
current_step, total_steps = tasklist.numerate_status()
print(f"Progress: {current_step}/{total_steps}")

# Check status dictionary
for task, status in tasklist.status_dict.items():
    print(f"{task}: {status}")

# Access detailed status information
print(f"Detailed status: {tasklist.detailed_status}")

Best Practices

• Ensure Neo4j database is properly configured and accessible before instantiating the class
• The study UID passed to __init__ must exist in the database as a Study node
• The class automatically queries the database upon instantiation via _get_status(), so instantiation may take time depending on database performance
• The tasks class variable defines the workflow order and should not be modified unless the database schema is updated accordingly
• The _get_status() method is private and called automatically during initialization; manual calls are not necessary
• The status_dict attribute is initialized with all tasks as 'Pending' and updated to 'Completed' based on database queries
• The numerate_status() method returns a default value of 10 if an exception occurs, which may not reflect actual progress
• Database connection is established per instance; consider connection pooling for multiple instances
• The detailed_status dictionary is initialized but never populated in the current implementation

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class Total_tasks 77.1% similar

  A class that retrieves and manages an overview of all current tasks from a Neo4j graph database, organized by task type and filtered by usergroup.

  From: /tf/active/vicechatdev/resources/taskmanager.py

• class Study_overview 58.8% similar

  A class that generates comprehensive study overview reports by querying a Neo4j graph database and producing Excel files with ID mappings, audit logs, and Gantt chart visualizations of study progress.

  From: /tf/active/vicechatdev/resources/documents.py

• function get_task_status 48.3% similar

  Flask API endpoint that retrieves the current status of a background task by its task ID from an in-memory active_tasks dictionary.

  From: /tf/active/vicechatdev/docchat/blueprint.py

• function api_task_status 48.0% similar

  Flask API endpoint that retrieves and returns the status of asynchronous tasks (chat or indexing operations) by task ID.

  From: /tf/active/vicechatdev/docchat/app.py

• function api_chat_status 47.8% similar

  Flask API endpoint that retrieves the detailed status of a chat task by delegating to the api_task_status function using the provided task_id.

  From: /tf/active/vicechatdev/docchat/app.py