567 lines
29 KiB
Python
567 lines
29 KiB
Python
from dotenv import load_dotenv, find_dotenv
|
|
load_dotenv(find_dotenv())
|
|
import os
|
|
import modules.logger_tool as logger
|
|
log_name = 'api_routers_database_tools_get_nodes'
|
|
log_dir = os.getenv("LOG_PATH", "/logs") # Default path as fallback
|
|
logging = logger.get_logger(
|
|
name=log_name,
|
|
log_level=os.getenv("LOG_LEVEL", "DEBUG"),
|
|
log_path=log_dir,
|
|
log_file=log_name,
|
|
runtime=True,
|
|
log_format='default'
|
|
)
|
|
import modules.database.tools.neo4j_driver_tools as driver
|
|
import modules.database.tools.neo4j_session_tools as session
|
|
from modules.database.schemas.nodes.calendars import CalendarNode
|
|
from modules.database.schemas.nodes.schools.timetable import SchoolTimetableNode, AcademicYearNode, AcademicTermNode, AcademicWeekNode, AcademicDayNode, AcademicPeriodNode, RegistrationPeriodNode
|
|
from modules.database.schemas.nodes.users import UserNode
|
|
from modules.database.schemas.nodes.workers.workers import TeacherNode, StudentNode, DeveloperNode, SchoolAdminNode
|
|
from modules.database.schemas.nodes.schools.schools import SchoolNode, DepartmentNode, SubjectClassNode, RoomNode
|
|
from modules.database.schemas.nodes.workers.timetable import TeacherTimetableNode, TimetableLessonNode, PlannedLessonNode, UserTeacherTimetableNode
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/get-node")
|
|
async def get_node(uuid_string: str = Query(...), db_name: str = Query(...)):
|
|
logging.info(f"Getting node for {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n {uuid_string: $uuid_string})
|
|
RETURN n
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
|
|
if record:
|
|
node = record['n']
|
|
node_labels = list(node.labels)
|
|
node_data = dict(node)
|
|
|
|
try:
|
|
# Convert node based on its type
|
|
node_type = node_labels[0] if node_labels else "Unknown"
|
|
logging.debug(f"Attempting to convert node of type: {node_type}")
|
|
logging.debug(f"Available node classes: {[name for name in globals() if name.endswith('Node')]}")
|
|
logging.debug(f"UserNode in globals: {'UserNode' in globals()}")
|
|
logging.debug(f"UserNode class: {UserNode}")
|
|
logging.debug(f"UserNode class name: {UserNode.__name__}")
|
|
|
|
# Try to find the node class
|
|
node_class_name = f"{node_type}Node"
|
|
if node_class_name in globals():
|
|
node_class = globals()[node_class_name]
|
|
logging.debug(f"Found node class: {node_class}")
|
|
node_object = node_class(**node_data)
|
|
node_dict = node_object.to_dict()
|
|
logging.debug(f"Successfully converted node to dict: {node_dict}")
|
|
else:
|
|
logging.warning(f"No node class found for type: {node_type} (looking for {node_class_name}), using raw data")
|
|
logging.debug(f"Available classes: {[name for name in globals() if 'Node' in name]}")
|
|
node_dict = node_data
|
|
|
|
return {
|
|
"status": "success",
|
|
"node": {
|
|
"node_type": node_type,
|
|
"node_data": node_dict
|
|
}
|
|
}
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
return {
|
|
"status": "error",
|
|
"message": "Error processing node data",
|
|
"details": str(e)
|
|
}
|
|
else:
|
|
return {"status": "not_found", "message": "Node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving node: {str(e)}")
|
|
return {"status": "error", "message": "Internal server error"}
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-user-node")
|
|
async def get_user_node(user_id: str = Query(...)):
|
|
db_name = f"cc.users.{user_id}"
|
|
logging.info(f"Getting user node for user {user_id} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
nodes = session.find_nodes_by_label_and_properties(neo_session, "User", {"user_id": user_id})
|
|
if nodes:
|
|
user_node = nodes[0]
|
|
data = UserNode(**user_node)
|
|
user_node_data = data.to_dict()
|
|
return {"status": "success", "user_node": user_node_data, "user_node_raw": nodes}
|
|
else:
|
|
return {"status": "not_found", "message": "User node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving user node: {str(e)}")
|
|
return {"status": "error", "message": "Internal server error"}
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-connected-nodes")
|
|
async def get_connected_nodes(uuid_string: str = Query(...), db_name: str = Query(...)):
|
|
logging.info(f"Getting connected nodes for {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n {uuid_string: $uuid_string})
|
|
OPTIONAL MATCH (n)-[]-(connected)
|
|
RETURN n, collect(connected) as connected_nodes
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
if record:
|
|
main_node = record['n']
|
|
connected_nodes = record['connected_nodes']
|
|
|
|
main_node_labels = list(main_node.labels)
|
|
main_node_type = main_node_labels[0] if main_node_labels else "Unknown"
|
|
main_node_data = dict(main_node)
|
|
|
|
try:
|
|
main_node_class = globals()[f"{main_node_type}Node"]
|
|
main_node_object = main_node_class(**main_node_data)
|
|
main_node_dict = main_node_object.to_dict()
|
|
except Exception as e:
|
|
logging.error(f"Error converting main node to dict: {str(e)}")
|
|
main_node_dict = main_node_data
|
|
|
|
connected_nodes_list = []
|
|
|
|
for node in connected_nodes:
|
|
node_labels = list(node.labels)
|
|
node_type = node_labels[0] if node_labels else "Unknown"
|
|
node_data = dict(node)
|
|
try:
|
|
node_class = globals()[f"{node_type}Node"]
|
|
node_object = node_class(**node_data)
|
|
connected_node_dict = node_object.to_dict()
|
|
except Exception as e:
|
|
logging.error(f"Error converting connected node to dict: {str(e)}")
|
|
connected_node_dict = node_data
|
|
|
|
connected_node_info = {
|
|
"node_type": node_type,
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
|
|
logging.debug(f"connected_nodes_list: {connected_nodes_list}")
|
|
|
|
return {
|
|
"status": "success",
|
|
"main_node": {
|
|
"node_type": main_node_type,
|
|
"node_data": main_node_dict
|
|
},
|
|
"connected_nodes": connected_nodes_list
|
|
}
|
|
else:
|
|
return {"status": "not_found", "message": "Node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving connected nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-user-connected-nodes")
|
|
async def get_user_connected_nodes(uuid_string: str = Query(...)):
|
|
logging.info(f"Getting user adjacent nodes for node {uuid_string}")
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai") # TODO: This function needs to be able to take a db_name as a parameter
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
raise HTTPException(status_code=500, detail="Failed to connect to the database")
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
user_node_and_connected_nodes = session.get_node_by_uuid_string_and_adjacent_nodes(neo_session, uuid_string)
|
|
user_node = user_node_and_connected_nodes['node']
|
|
connected_nodes = user_node_and_connected_nodes['connected_nodes']
|
|
try:
|
|
data = UserNode(**user_node)
|
|
user_node_dict = data.to_dict()
|
|
except Exception as e:
|
|
logging.error(f"Error converting user node to dict: {str(e)}")
|
|
connected_nodes_list = []
|
|
for connected_node in connected_nodes:
|
|
node_data = connected_node['node']
|
|
node_labels = list(node_data.labels)
|
|
logging.debug(f"node_labels: {node_labels}")
|
|
for label in node_labels:
|
|
logging.debug(f"label: {label}")
|
|
try:
|
|
if 'Developer' == label:
|
|
logging.debug(f"Developer node found")
|
|
node_object = DeveloperNode(**node_data)
|
|
elif 'SchoolAdmin' == label:
|
|
logging.debug(f"SchoolAdmin node found")
|
|
node_object = SchoolAdminNode(**node_data)
|
|
elif 'Teacher' == label:
|
|
logging.debug(f"Teacher node found")
|
|
node_object = TeacherNode(**node_data)
|
|
elif 'Student' == label:
|
|
logging.debug(f"Student node found")
|
|
node_object = StudentNode(**node_data)
|
|
elif 'Calendar' == label:
|
|
logging.debug(f"Calendar node found")
|
|
node_object = CalendarNode(**node_data)
|
|
elif 'TeacherTimetable' == label:
|
|
logging.debug(f"TeacherTimetable node found")
|
|
node_object = TeacherTimetableNode(**node_data)
|
|
elif 'UserTeacherTimetable' == label:
|
|
logging.debug(f"UserTeacherTimetable node found")
|
|
node_object = UserTeacherTimetableNode(**node_data)
|
|
elif 'School' == label:
|
|
logging.debug(f"School node found")
|
|
node_object = SchoolNode(**node_data)
|
|
elif 'Department' == label:
|
|
logging.debug(f"Department node found")
|
|
node_object = DepartmentNode(**node_data)
|
|
elif 'Student' == label:
|
|
logging.debug(f"Student node found")
|
|
node_object = StudentNode(**node_data)
|
|
elif 'Class' == label:
|
|
logging.debug(f"Class node found")
|
|
node_object = SubjectClassNode(**node_data)
|
|
elif 'Room' == label:
|
|
logging.debug(f"Room node found")
|
|
node_object = RoomNode(**node_data)
|
|
else:
|
|
logging.error(f"Unknown node label: {node_labels}")
|
|
continue
|
|
connected_node_dict = node_object.to_dict()
|
|
logging.debug(f"connected_node_dict: {connected_node_dict}")
|
|
connected_node_info = {
|
|
"node_type": label,
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
return {"status": "success", "user_node": user_node_dict, "user_connected_nodes": connected_nodes_list}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving adjacent nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-worker-connected-nodes")
|
|
async def get_worker_connected_nodes(uuid_string: str = Query(...)):
|
|
logging.info(f"Getting worker adjacent nodes for node {uuid_string}")
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai") # TODO: This function needs to be able to take a db_name as a parameter
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
raise HTTPException(status_code=500, detail="Failed to connect to the database")
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
node_and_connected_nodes = session.get_node_by_uuid_string_and_adjacent_nodes(neo_session, uuid_string)
|
|
worker_node = node_and_connected_nodes['node']
|
|
connected_nodes = node_and_connected_nodes['connected_nodes']
|
|
try:
|
|
data = TeacherNode(**worker_node)
|
|
worker_node_dict = data.to_dict()
|
|
except Exception as e:
|
|
logging.error(f"Error converting user node to dict: {str(e)}")
|
|
connected_nodes_list = []
|
|
for connected_node in connected_nodes:
|
|
node_data = connected_node['node']
|
|
node_labels = list(node_data.labels)
|
|
logging.debug(f"node_labels: {node_labels}")
|
|
for label in node_labels:
|
|
logging.debug(f"label: {label}")
|
|
try:
|
|
if 'Calendar' == label:
|
|
logging.debug(f"Calendar node found")
|
|
node_object = CalendarNode(**node_data)
|
|
elif 'TeacherTimetable' == label:
|
|
logging.debug(f"TeacherTimetable node found")
|
|
node_object = TeacherTimetableNode(**node_data)
|
|
elif 'UserTeacherTimetable' == label:
|
|
logging.debug(f"UserTeacherTimetable node found")
|
|
node_object = UserTeacherTimetableNode(**node_data)
|
|
elif 'School' == label:
|
|
logging.debug(f"School node found")
|
|
node_object = SchoolNode(**node_data)
|
|
elif 'Department' == label:
|
|
logging.debug(f"Department node found")
|
|
node_object = DepartmentNode(**node_data)
|
|
elif 'Student' == label:
|
|
logging.debug(f"Student node found")
|
|
node_object = StudentNode(**node_data)
|
|
elif 'Class' == label:
|
|
logging.debug(f"Class node found")
|
|
node_object = SubjectClassNode(**node_data)
|
|
elif 'Room' == label:
|
|
logging.debug(f"Room node found")
|
|
node_object = RoomNode(**node_data)
|
|
else:
|
|
logging.error(f"Unknown node label: {node_labels}")
|
|
continue
|
|
connected_node_dict = node_object.to_dict()
|
|
logging.debug(f"connected_node_dict: {connected_node_dict}")
|
|
connected_node_info = {
|
|
"node_type": label,
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
return {"status": "success", "user_node": worker_node_dict, "worker_connected_nodes": connected_nodes_list}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving worker adjacent nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-calendar-connected-nodes")
|
|
async def get_calendar_connected_nodes(uuid_string: str = Query(...)):
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai")
|
|
logging.info(f"Getting connected nodes for calendar {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n)
|
|
WHERE n.uuid_string = $uuid_string AND (n:Calendar OR n:CalendarYear OR n:CalendarMonth OR n:CalendarWeek OR n:CalendarDay OR n:CalendarTimeChunk)
|
|
OPTIONAL MATCH (n)-[]-(connected)
|
|
RETURN n, collect(connected) as connected_nodes
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
if record:
|
|
calendar_node = record['n']
|
|
connected_nodes = record['connected_nodes']
|
|
|
|
node_type = list(calendar_node.labels)[0]
|
|
calendar_dict = globals()[f"{node_type}Node"](**calendar_node).to_dict()
|
|
connected_nodes_list = []
|
|
|
|
for node in connected_nodes:
|
|
node_labels = list(node.labels)
|
|
node_data = dict(node)
|
|
try:
|
|
node_class = globals()[f"{node_labels[0]}Node"]
|
|
node_object = node_class(**node_data)
|
|
connected_node_dict = node_object.to_dict()
|
|
connected_node_info = {
|
|
"node_type": node_labels[0],
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
|
|
return {"status": "success", "calendar_node": calendar_dict, "connected_nodes": connected_nodes_list}
|
|
else:
|
|
return {"status": "not_found", "message": "Calendar node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving connected nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-teacher-timetable-connected-nodes")
|
|
async def get_teacher_timetable_connected_nodes(uuid_string: str = Query(...)):
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai")
|
|
logging.info(f"Getting connected nodes for teacher timetable {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n:TeacherTimetable {uuid_string: $uuid_string})
|
|
OPTIONAL MATCH (n)-[]-(connected)
|
|
RETURN n, collect(connected) as connected_nodes
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
if record:
|
|
teacher_timetable_node = record['n']
|
|
connected_nodes = record['connected_nodes']
|
|
|
|
teacher_timetable_dict = TeacherTimetableNode(**teacher_timetable_node).to_dict()
|
|
connected_nodes_list = []
|
|
|
|
for node in connected_nodes:
|
|
node_labels = list(node.labels)
|
|
node_data = dict(node)
|
|
try:
|
|
if 'TimetableLesson' in node_labels:
|
|
node_object = TimetableLessonNode(**node_data)
|
|
elif 'PlannedLesson' in node_labels:
|
|
node_object = PlannedLessonNode(**node_data)
|
|
else:
|
|
logging.error(f"Unknown node label: {node_labels}")
|
|
continue
|
|
connected_node_dict = node_object.to_dict()
|
|
connected_node_info = {
|
|
"node_type": node_labels[0],
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
|
|
return {"status": "success", "teacher_timetable_node": teacher_timetable_dict, "connected_nodes": connected_nodes_list}
|
|
else:
|
|
return {"status": "not_found", "message": "Teacher timetable node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving connected nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-school-timetable-connected-nodes")
|
|
async def get_school_timetable_connected_nodes(uuid_string: str = Query(...)):
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai")
|
|
logging.info(f"Getting connected nodes for school timetable {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n:SchoolTimetable {uuid_string: $uuid_string})
|
|
OPTIONAL MATCH (n)-[]-(connected)
|
|
RETURN n, collect(connected) as connected_nodes
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
if record:
|
|
school_timetable_node = record['n']
|
|
connected_nodes = record['connected_nodes']
|
|
|
|
school_timetable_dict = SchoolTimetableNode(**school_timetable_node).to_dict()
|
|
connected_nodes_list = []
|
|
|
|
for node in connected_nodes:
|
|
node_labels = list(node.labels)
|
|
node_data = dict(node)
|
|
try:
|
|
if 'AcademicYear' in node_labels:
|
|
node_object = AcademicYearNode(**node_data)
|
|
elif 'AcademicTerm' in node_labels:
|
|
node_object = AcademicTermNode(**node_data)
|
|
elif 'AcademicWeek' in node_labels:
|
|
node_object = AcademicWeekNode(**node_data)
|
|
elif 'AcademicDay' in node_labels:
|
|
node_object = AcademicDayNode(**node_data)
|
|
elif 'AcademicPeriod' in node_labels:
|
|
node_object = AcademicPeriodNode(**node_data)
|
|
elif 'RegistrationPeriod' in node_labels:
|
|
node_object = RegistrationPeriodNode(**node_data)
|
|
else:
|
|
logging.error(f"Unknown node label: {node_labels}")
|
|
continue
|
|
connected_node_dict = node_object.to_dict()
|
|
connected_node_info = {
|
|
"node_type": node_labels[0],
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
|
|
return {"status": "success", "school_timetable_node": school_timetable_dict, "connected_nodes": connected_nodes_list}
|
|
else:
|
|
return {"status": "not_found", "message": "School timetable node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving connected nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-curriculum-connected-nodes")
|
|
async def get_curriculum_connected_nodes(uuid_string: str = Query(...)):
|
|
db_name = os.getenv("NEO4J_DB_NAME", "cc.institutes.kevlarai")
|
|
logging.info(f"Getting connected nodes for curriculum {uuid_string} from database {db_name}")
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
query = """
|
|
MATCH (n)
|
|
WHERE n.uuid_string = $uuid_string AND (n:PastoralStructure OR n:YearGroup OR n:CurriculumStructure OR n:KeyStage OR n:KeyStageSyllabus OR n:YearGroupSyllabus OR n:Subject OR n:Topic OR n:TopicLesson OR n:LearningStatement OR n:ScienceLab)
|
|
OPTIONAL MATCH (n)-[]-(connected)
|
|
RETURN n, collect(connected) as connected_nodes
|
|
"""
|
|
result = neo_session.run(query, uuid_string=uuid_string)
|
|
record = result.single()
|
|
if record:
|
|
curriculum_node = record['n']
|
|
connected_nodes = record['connected_nodes']
|
|
|
|
node_type = list(curriculum_node.labels)[0]
|
|
curriculum_dict = globals()[f"{node_type}Node"](**curriculum_node).to_dict()
|
|
connected_nodes_list = []
|
|
|
|
for node in connected_nodes:
|
|
node_labels = list(node.labels)
|
|
node_data = dict(node)
|
|
try:
|
|
node_class = globals()[f"{node_labels[0]}Node"]
|
|
node_object = node_class(**node_data)
|
|
connected_node_dict = node_object.to_dict()
|
|
connected_node_info = {
|
|
"node_type": node_labels[0],
|
|
"node_data": connected_node_dict
|
|
}
|
|
connected_nodes_list.append(connected_node_info)
|
|
except Exception as e:
|
|
logging.error(f"Error converting node to dict: {str(e)}")
|
|
|
|
return {"status": "success", "curriculum_node": curriculum_dict, "connected_nodes": connected_nodes_list}
|
|
else:
|
|
return {"status": "not_found", "message": "Curriculum node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving connected nodes: {str(e)}")
|
|
raise HTTPException(status_code=500, detail="Internal server error")
|
|
finally:
|
|
driver.close_driver(neo_driver)
|
|
|
|
@router.get("/get-school-node")
|
|
async def get_school_node(school_uuid_string: str = Query(...)):
|
|
logging.info(f"Getting school node for school {school_uuid_string}...")
|
|
db_name = f"cc.institutes.{school_uuid_string}"
|
|
neo_driver = driver.get_driver(db_name=db_name)
|
|
if neo_driver is None:
|
|
return {"status": "error", "message": "Failed to connect to the database"}
|
|
|
|
try:
|
|
with neo_driver.session(database=db_name) as neo_session:
|
|
nodes = session.find_nodes_by_label_and_properties(neo_session, "School", {"uuid_string": school_uuid_string})
|
|
if nodes:
|
|
school_node_data = SchoolNode(**nodes[0]).to_dict()
|
|
return {"status": "success", "school_node": school_node_data, "school_node_raw": nodes}
|
|
else:
|
|
return {"status": "not_found", "message": "School node not found"}
|
|
except Exception as e:
|
|
logging.error(f"Error retrieving school node: {str(e)}")
|
|
return {"status": "error", "message": "Internal server error"}
|
|
finally:
|
|
driver.close_driver(neo_driver) |