🔍 Code Extractor

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.

Purpose

This class connects to a Neo4j database to fetch and organize laboratory workflow tasks across different stages (Quote, Delivery, Sample Registration, Grossing, Embedding, Sectioning, Staining, Scanning, Digital QA, Assessment, Draft Report, Final Report). It queries the database to retrieve tasks that haven't been completed in previous stages, organizing them by customer, study, sample, and target information. The class is designed for laboratory information management systems to provide task overview dashboards.

Source Code

class Total_tasks():    
    """
    Get the overview of all current tasks
    
    Attributes
    ----------
    graph : Py2neo.Graph
        Database graph
    usergroup : str
        Current usergroup accessing the information
    all_tasks : dictionary
        A dictionary mapping each task to an empty list
        
    Parameters
    ----------
    usergroup : str
        Current usergroup accessing the information, defaults to Administrators
        
    """
    tasks = ["Quote", "Delivery", "Sample Registration", "Grossing", "Embedding", "Sectioning", "Staining", "Scanning", "Digital QA","Assessment","Draft Report","Final Report"]

    def __init__(self, usergroup='Admin'):
        # 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.usergroup = usergroup #Make tasks role based in future
        self.all_tasks = {'Sample Registration':[],
                          'Grossing':[],
                          'Embedding':[],
                          'Sectioning':[],
                          'Staining':[],
                          'Assessment':[],
                         }
        self.all_tasks = {i:[] for i in self.tasks}
        self.get_tasks()
        
        
    def get_tasks(self):
        try:
            prev_keys=[]
            for key in self.tasks:
                if key in ["Quote","Delivery", "Draft Report", "Final Report"]:
                    if prev_keys:
                        gtext=f"""
                        MATCH (c:Customer)-->(s:Study)-->(n:Task)
                        WHERE n.T = '{key}'
                        AND NOT EXISTS {{
                            (s)-[*]->(t:Task)
                            WHERE t.T in {prev_keys}
                        }}
                        RETURN c.N as Customer, s.external_N as Study, n.T as Task, n.UID as Task_UID
                        """
                        self.all_tasks[key] = self.graph.run(gtext).to_data_frame()
                    else:
                        self.all_tasks[key] = self.graph.run(f"match (c:Customer)-->(s:Study)-->(n:Task) WHERE n.T = '{key}' \
                        RETURN c.N as Customer, s.external_N as Study, n.T as Task, n.UID as Task_UID").to_data_frame()
                else:
                    gtext=f"""
                    MATCH (c:Customer)-->(s:Study)-[*2]->(e:ExpItem)-[*]->(o)-->(n:Task)
                    WHERE n.T = '{key}'
                    AND NOT EXISTS {{
                        (s)-[*]->(t:Task)
                        WHERE t.T in {prev_keys}
                    }}
                    RETURN c.N as Customer, s.external_N as Study, e.N as Sample, e.UID as Sample_UID, o.N as Target, o.UID as Target_UID,n.T as Task, n.UID as Task_UID
                    """
                    if key == 'Scanning':
                        gtext += ", n.N as Magnification"
                    self.all_tasks[key] = self.graph.run(gtext).to_data_frame()
                prev_keys.append(key)
        except Exception as e:
            print(e)

Parameters

Parameter Details

usergroup: A string representing the current usergroup accessing the task information. Defaults to 'Admin'. This parameter is intended for future role-based task filtering, though currently not actively used in the query logic.

Return Value

The class constructor returns an instance of Total_tasks. The instance contains the 'all_tasks' attribute, which is a dictionary where keys are task type names (strings) and values are pandas DataFrames containing task details. Each DataFrame includes columns like Customer, Study, Task, Task_UID, and for non-administrative tasks: Sample, Sample_UID, Target, Target_UID. The Scanning task additionally includes a Magnification column.

Class Interface

Methods

Attributes

Dependencies

• py2neo
• pandas
• panel
• param
• json
• datetime
• io
• 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, DB_NAME configured
# Example: config.DB_ADDR = 'bolt://localhost:7687'
# config.DB_AUTH = ('neo4j', 'password')
# config.DB_NAME = 'neo4j'

from total_tasks import Total_tasks

# Instantiate with default admin usergroup
tasks_overview = Total_tasks()

# Access all tasks dictionary
all_tasks = tasks_overview.all_tasks

# Get specific task type data (returns pandas DataFrame)
staining_tasks = tasks_overview.all_tasks['Staining']
print(staining_tasks.head())

# Get scanning tasks with magnification info
scanning_tasks = tasks_overview.all_tasks['Scanning']
print(scanning_tasks[['Customer', 'Study', 'Sample', 'Magnification']])

# Instantiate with specific usergroup (for future role-based filtering)
tasks_for_lab = Total_tasks(usergroup='Laboratory')

Best Practices

• Ensure Neo4j database connection is available before instantiation, as the constructor immediately queries the database
• The class automatically populates all_tasks dictionary upon instantiation via get_tasks() method
• Handle potential connection exceptions when instantiating, as database connectivity issues will raise exceptions
• The usergroup parameter is currently not used in filtering logic but is reserved for future role-based access control
• Each task type returns a pandas DataFrame, so pandas operations can be applied to filter or transform results
• The class queries all tasks at instantiation time; for real-time updates, create a new instance
• Tasks are filtered to exclude those where previous stage tasks exist, ensuring only actionable tasks are returned
• The get_tasks() method is called automatically in __init__ and should not typically be called manually
• Database queries use Cypher query language with pattern matching and NOT EXISTS clauses for filtering

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• class Tasklist 77.1% similar

  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.

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

• function get_training_overview 58.3% similar

  Retrieves a comprehensive training overview for the admin panel, including training plans, active assignments, and recent completions from a Neo4j graph database.

  From: /tf/active/vicechatdev/CDocs/controllers/training_controller.py

• class Study_overview 56.7% 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_user_training_dashboard 53.2% similar

  Retrieves a user's training dashboard data by querying Neo4j for required and completed training records associated with controlled documents.

  From: /tf/active/vicechatdev/CDocs/controllers/training_controller.py

• function get_users 49.3% similar

  Retrieves user information from a Neo4j graph database for display in an admin panel, with optional filtering by user ID.

  From: /tf/active/vicechatdev/CDocs/controllers/admin_controller.py