452 lines
16 KiB
Python
452 lines
16 KiB
Python
"""
|
|
Supabase Storage Tools for ClassroomCopilot
|
|
Replaces local filesystem paths with Supabase Storage bucket paths
|
|
"""
|
|
import os
|
|
from modules.logger_tool import initialise_logger
|
|
from typing import Tuple, Optional
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
|
|
class SupabaseStorageTools:
|
|
"""
|
|
Generates Supabase Storage paths for TLDraw snapshots and other files.
|
|
|
|
Path format: bucket/nodetype/node_unique_id
|
|
Example: cc.snapshots/User/fda8dca7-4d18-43c9-bb74-260777043447
|
|
"""
|
|
|
|
def __init__(self, db_name: str, init_run_type: str = None):
|
|
self.db_name = db_name
|
|
self.init_run_type = init_run_type
|
|
|
|
# Define bucket mappings based on node types
|
|
self.bucket_mappings = {
|
|
'User': 'cc.public.snapshots',
|
|
'Teacher': 'cc.public.snapshots',
|
|
'Student': 'cc.public.snapshots',
|
|
'School': 'cc.public.snapshots',
|
|
'Department': 'cc.public.snapshots',
|
|
'Subject': 'cc.public.snapshots',
|
|
'CalendarYear': 'cc.public.snapshots',
|
|
'CalendarMonth': 'cc.public.snapshots',
|
|
'CalendarWeek': 'cc.public.snapshots',
|
|
'CalendarDay': 'cc.public.snapshots',
|
|
'CalendarTimeChunk': 'cc.public.snapshots',
|
|
'KeyStage': 'cc.public.snapshots',
|
|
'YearGroup': 'cc.public.snapshots',
|
|
'KeyStageSyllabus': 'cc.public.snapshots',
|
|
'YearGroupSyllabus': 'cc.public.snapshots',
|
|
'Topic': 'cc.public.snapshots',
|
|
'TopicLesson': 'cc.public.snapshots',
|
|
'LearningStatement': 'cc.public.snapshots',
|
|
'UserTeacherTimetable': 'cc.public.snapshots',
|
|
'Class': 'cc.public.snapshots',
|
|
'TimetableLesson': 'cc.public.snapshots',
|
|
'SuperAdmin': 'cc.public.snapshots',
|
|
'Developer': 'cc.public.snapshots',
|
|
'CurriculumStructure': 'cc.public.snapshots',
|
|
'PastoralStructure': 'cc.public.snapshots',
|
|
'DepartmentStructure': 'cc.public.snapshots',
|
|
}
|
|
|
|
logger.info(f"Initializing SupabaseStorageTools with db_name: {db_name} and init_run_type: {init_run_type}")
|
|
|
|
def get_storage_path(self, node_type: str, node_id: str) -> str:
|
|
"""
|
|
Generate Supabase Storage path for a node.
|
|
|
|
Args:
|
|
node_type: The type of node (e.g., 'User', 'Teacher', 'School')
|
|
node_id: The unique identifier for the node
|
|
|
|
Returns:
|
|
str: Storage path in format bucket/nodetype/node_id
|
|
"""
|
|
bucket = self.bucket_mappings.get(node_type, 'cc.public.snapshots')
|
|
path = f"{bucket}/{node_type}/{node_id}"
|
|
|
|
logger.debug(f"Generated storage path for {node_type} {node_id}: {path}")
|
|
return path
|
|
|
|
def create_user_storage_path(self, user_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a user node.
|
|
|
|
Args:
|
|
user_id: The user's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('User', user_id)
|
|
logger.info(f"Created user storage path: {path}")
|
|
return True, path
|
|
|
|
def create_teacher_storage_path(self, teacher_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a teacher node.
|
|
|
|
Args:
|
|
teacher_id: The teacher's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Teacher', teacher_id)
|
|
logger.info(f"Created teacher storage path: {path}")
|
|
return True, path
|
|
|
|
def create_student_storage_path(self, student_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a student node.
|
|
|
|
Args:
|
|
student_id: The student's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Student', student_id)
|
|
logger.info(f"Created student storage path: {path}")
|
|
return True, path
|
|
|
|
def create_school_storage_path(self, school_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a school node.
|
|
|
|
Args:
|
|
school_id: The school's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('School', school_id)
|
|
logger.info(f"Created school storage path: {path}")
|
|
return True, path
|
|
|
|
def create_calendar_year_storage_path(self, year: int) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a calendar year node.
|
|
|
|
Args:
|
|
year: The year
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('CalendarYear', str(year))
|
|
logger.info(f"Created calendar year storage path: {path}")
|
|
return True, path
|
|
|
|
def create_calendar_month_storage_path(self, year: int, month: int) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a calendar month node.
|
|
|
|
Args:
|
|
year: The year
|
|
month: The month
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
month_id = f"{year}_{month:02d}"
|
|
path = self.get_storage_path('CalendarMonth', month_id)
|
|
logger.info(f"Created calendar month storage path: {path}")
|
|
return True, path
|
|
|
|
def create_calendar_week_storage_path(self, year: int, week: int) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a calendar week node.
|
|
|
|
Args:
|
|
year: The ISO year
|
|
week: The ISO week number
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
week_id = f"{year}_{week:02d}"
|
|
path = self.get_storage_path('CalendarWeek', week_id)
|
|
logger.info(f"Created calendar week storage path: {path}")
|
|
return True, path
|
|
|
|
def create_calendar_day_storage_path(self, year: int, month: int, day: int) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a calendar day node.
|
|
|
|
Args:
|
|
year: The year
|
|
month: The month
|
|
day: The day
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
day_id = f"{year}_{month:02d}_{day:02d}"
|
|
path = self.get_storage_path('CalendarDay', day_id)
|
|
logger.info(f"Created calendar day storage path: {path}")
|
|
return True, path
|
|
|
|
def create_calendar_time_chunk_storage_path(self, day_id: str, chunk_index: int) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a calendar time chunk node.
|
|
|
|
Args:
|
|
day_id: The day identifier (e.g., "2025_01_15")
|
|
chunk_index: The time chunk index within the day
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
chunk_id = f"{day_id}_{chunk_index:02d}"
|
|
path = self.get_storage_path('CalendarTimeChunk', chunk_id)
|
|
logger.info(f"Created calendar time chunk storage path: {path}")
|
|
return True, path
|
|
|
|
def create_department_storage_path(self, department_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a department node.
|
|
|
|
Args:
|
|
department_id: The department's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Department', department_id)
|
|
logger.info(f"Created department storage path: {path}")
|
|
return True, path
|
|
|
|
def create_subject_storage_path(self, subject_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a subject node.
|
|
|
|
Args:
|
|
subject_id: The subject's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Subject', subject_id)
|
|
logger.info(f"Created subject storage path: {path}")
|
|
return True, path
|
|
|
|
def create_key_stage_storage_path(self, key_stage_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a key stage node.
|
|
|
|
Args:
|
|
key_stage_id: The key stage's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('KeyStage', key_stage_id)
|
|
logger.info(f"Created key stage storage path: {path}")
|
|
return True, path
|
|
|
|
def create_year_group_storage_path(self, year_group_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a year group node.
|
|
|
|
Args:
|
|
year_group_id: The year group's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('YearGroup', year_group_id)
|
|
logger.info(f"Created year group storage path: {path}")
|
|
return True, path
|
|
|
|
def create_key_stage_syllabus_storage_path(self, syllabus_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a key stage syllabus node.
|
|
|
|
Args:
|
|
syllabus_id: The syllabus's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('KeyStageSyllabus', syllabus_id)
|
|
logger.info(f"Created key stage syllabus storage path: {path}")
|
|
return True, path
|
|
|
|
def create_year_group_syllabus_storage_path(self, syllabus_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a year group syllabus node.
|
|
|
|
Args:
|
|
syllabus_id: The syllabus's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('YearGroupSyllabus', syllabus_id)
|
|
logger.info(f"Created year group syllabus storage path: {path}")
|
|
return True, path
|
|
|
|
def create_topic_storage_path(self, topic_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a topic node.
|
|
|
|
Args:
|
|
topic_id: The topic's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Topic', topic_id)
|
|
logger.info(f"Created topic storage path: {path}")
|
|
return True, path
|
|
|
|
def create_topic_lesson_storage_path(self, lesson_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a topic lesson node.
|
|
|
|
Args:
|
|
lesson_id: The lesson's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('TopicLesson', lesson_id)
|
|
logger.info(f"Created topic lesson storage path: {path}")
|
|
return True, path
|
|
|
|
def create_learning_statement_storage_path(self, statement_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a learning statement node.
|
|
|
|
Args:
|
|
statement_id: The statement's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('LearningStatement', statement_id)
|
|
logger.info(f"Created learning statement storage path: {path}")
|
|
return True, path
|
|
|
|
def create_timetable_storage_path(self, timetable_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a timetable node.
|
|
|
|
Args:
|
|
timetable_id: The timetable's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('UserTeacherTimetable', timetable_id)
|
|
logger.info(f"Created timetable storage path: {path}")
|
|
return True, path
|
|
|
|
def create_class_storage_path(self, class_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a class node.
|
|
|
|
Args:
|
|
class_id: The class's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('Class', class_id)
|
|
logger.info(f"Created class storage path: {path}")
|
|
return True, path
|
|
|
|
def create_timetable_lesson_storage_path(self, lesson_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a timetable lesson node.
|
|
|
|
Args:
|
|
lesson_id: The lesson's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('TimetableLesson', lesson_id)
|
|
logger.info(f"Created timetable lesson storage path: {path}")
|
|
return True, path
|
|
|
|
def create_super_admin_storage_path(self, admin_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a super admin node.
|
|
|
|
Args:
|
|
admin_id: The admin's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('SuperAdmin', admin_id)
|
|
logger.info(f"Created super admin storage path: {path}")
|
|
return True, path
|
|
|
|
def create_curriculum_storage_path(self, curriculum_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a curriculum structure node.
|
|
|
|
Args:
|
|
curriculum_id: The curriculum's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('CurriculumStructure', curriculum_id)
|
|
logger.info(f"Created curriculum structure storage path: {path}")
|
|
return True, path
|
|
|
|
def create_pastoral_storage_path(self, pastoral_id: str) -> Tuple[bool, str]:
|
|
"""
|
|
Create storage path for a pastoral structure node.
|
|
|
|
Args:
|
|
pastoral_id: The pastoral's unique identifier
|
|
|
|
Returns:
|
|
Tuple[bool, str]: (success, storage_path)
|
|
"""
|
|
path = self.get_storage_path('PastoralStructure', pastoral_id)
|
|
logger.info(f"Created pastoral structure storage path: {path}")
|
|
return True, path
|
|
|
|
# Legacy compatibility methods that return the same interface as filesystem tools
|
|
def create_private_user_directory(self, user_id: str) -> Tuple[bool, str]:
|
|
"""Legacy compatibility method."""
|
|
return self.create_user_storage_path(user_id)
|
|
|
|
def create_user_worker_directory(self, user_path: str, worker_id: str, worker_type: str) -> Tuple[bool, str]:
|
|
"""Legacy compatibility method."""
|
|
if worker_type in ['teacher', 'email_teacher', 'ms_teacher']:
|
|
return self.create_teacher_storage_path(worker_id)
|
|
elif worker_type in ['student', 'email_student', 'ms_student']:
|
|
return self.create_student_storage_path(worker_id)
|
|
elif worker_type == 'superadmin':
|
|
return self.create_super_admin_storage_path(worker_id)
|
|
elif worker_type == 'developer':
|
|
return self.create_developer_storage_path(worker_id)
|
|
else:
|
|
# Default to generic storage path
|
|
path = self.get_storage_path(worker_type.title(), worker_id)
|
|
return True, path
|
|
|
|
def create_school_directory(self, school_uuid_string: str) -> Tuple[bool, str]:
|
|
"""Legacy compatibility method."""
|
|
return self.create_school_storage_path(school_uuid_string)
|
|
|
|
def create_school_curriculum_directory(self, school_path: Optional[str] = None) -> Tuple[bool, str]:
|
|
"""Legacy compatibility method - returns empty path since curriculum is handled by individual nodes."""
|
|
return True, ""
|
|
|
|
def create_school_pastoral_directory(self, school_path: Optional[str] = None) -> Tuple[bool, str]:
|
|
"""Legacy compatibility method - returns empty path since pastoral is handled by individual nodes."""
|
|
return True, ""
|
|
|
|
def create_directory(self, path: str) -> bool:
|
|
"""Legacy compatibility method - always returns True since we don't create physical directories."""
|
|
return True
|