[Insights app] Service to report the employee(s) who learnt the most or least the previous month

tech_learnt_most_least._emp.py

"""
Script to fetch the employee(s) who learnt the most and least last month, from Survey data.
"""
import json
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import common_functions

def get_tech_query():
    "Returns the Cypher query string, to fetch tech learnt data, for specified date range"
    query = '''
    MATCH (e:Employees)-[r]->(t:Technology)
    UNWIND r.learnt_dates as learnt_date
    WITH learnt_date as tech_learnt_date, e.fullName as emp_name,
        t.technology_name as tech_name
    WHERE date(tech_learnt_date) >= date($start_date)
        AND date(tech_learnt_date) <= date($end_date)
        AND NOT emp_name IN $excluded_employees
    WITH emp_name, collect(DISTINCT tech_learnt_date) as tech_learnt_dates,
        collect(DISTINCT tech_name) as tech_names
    RETURN emp_name, tech_learnt_dates, tech_names
    '''
    return query

def clean_up_tech(records):
    "Returns tech with tech count post clean-up"
    for record in records:
        if '' in record["tech_names"]:
            record["tech_names"].remove('')
        record["count"] = len(record["tech_names"])
    return records

def get_most_tech_count_data(records):
    "Returns data of employee(s) who've learnt the most that duration"
    max_count = max(record["count"] for record in records)
    most_tech_count_data = [record for record in records if record["count"] == max_count]
    return most_tech_count_data

def get_least_tech_count_data(records):
    "Returns data of employee(s) who've learnt the least that duration"
    min_count = min(record["count"] for record in records)
    min_tech_count_data = [record for record in records if record["count"] == min_count]
    return min_tech_count_data

def get_tech_learnt_data(tx, start_date, end_date, excluded_employees):
    "Returns data of employee(s) who learnt most and least, the previous month, from Survey DB"
    query_string = get_tech_query()
    try:
        neo4j_response = tx.run(query_string, start_date = start_date, end_date = end_date,
                                                 excluded_employees = excluded_employees)
        query_results = [record.data() for record in neo4j_response]
        # query_results is a list of dicts
        query_results = clean_up_tech(query_results)
        max_count_emps = get_most_tech_count_data(query_results)
        min_count_emps = get_least_tech_count_data(query_results)
    except Exception as error:
        print(f"Exception : {error}" )
        raise Exception("Unable to fetch query results!") from error
    return max_count_emps, min_count_emps

def collate_results(most_learnt_emp_data, least_learnt_emp_data, duration):
    "Collates insights to report"
    most_learnt_data = ""
    least_learnt_data = ""
    for data in most_learnt_emp_data:
        most_learnt_data = most_learnt_data + f' <i>{data["emp_name"]}</i> \n' +\
               ' Tech learnt: ' +\
               f' {data["tech_names"]} \n'

    for data in least_learnt_emp_data:
        least_learnt_data = least_learnt_data + f' <i>{data["emp_name"]}</i> \n' +\
               ' Tech learnt: ' +\
               f' {data["tech_names"]} \n'

    results = '<b>Qxf2 Insights</b> \n' + \
         f'<i>Learning in {duration}</i>: \n'  +\
         '\nMost learnt employee(s): \n' +\
         f'{most_learnt_data}' +\
         '\n' +\
         'Least learnt employee(s): \n' +\
         f'{least_learnt_data}'
    common_functions.report_results(results)

if __name__ == '__main__':    
    NEO4J_DRIVER = common_functions.initialize_db_driver()
    starting_date, ending_date, month_year = common_functions.get_last_month_start_end_dates()
    try:
        with NEO4J_DRIVER.session(database = os.environ.get("DB_NAME", None)) as session:
            most_tech_data, least_tech_data = session.execute_read(get_tech_learnt_data,
                                                starting_date, ending_date,
                                                json.loads(os.environ.get("EXCLUDED_NAMES", [])))
        NEO4J_DRIVER.close()
        collate_results(most_tech_data, least_tech_data, month_year)
    except Exception as err:
        print(f"Exception: {err}")

A service that

• filters(data between a passed duration) and fetches survey data(employee wise techs learnt, from Neo4j database)
• Determines the employee(s) who have learnt the most or least
• Collates a insight