From 98be55ab575ab514e6db657a1269c64445ebb840 Mon Sep 17 00:00:00 2001 From: CC Worker Date: Sat, 6 Jun 2026 18:46:50 +0000 Subject: [PATCH] fix(storage): de-double braces in storage.py (set-of-dict crash) Pre-existing bug (E7): the whole file had doubled braces, so every dict literal was a set containing a dict -> 'unhashable type: dict' at runtime, and log f-strings printed literal {braces}. This broke StorageManager.upload_file / list_bucket_contents / create_bucket / bucket-init for ALL callers (incl. files.py uploads), not just exam scans. Mechanical de-double; no logic change. Co-Authored-By: Claude Opus 4.8 --- modules/database/supabase/utils/storage.py | 98 +++++++++++----------- 1 file changed, 49 insertions(+), 49 deletions(-) diff --git a/modules/database/supabase/utils/storage.py b/modules/database/supabase/utils/storage.py index 9d0d1d7..a0c1c35 100644 --- a/modules/database/supabase/utils/storage.py +++ b/modules/database/supabase/utils/storage.py @@ -23,76 +23,76 @@ class StorageManager: def check_bucket_exists(self, bucket_id: str) -> bool: """Check if a storage bucket exists""" try: - self.logger.info(f"Checking if bucket {{bucket_id}} exists") + self.logger.info(f"Checking if bucket {bucket_id} exists") buckets = self.client.supabase.storage.list_buckets() return any(bucket.name == bucket_id for bucket in buckets) except Exception as e: - self.logger.error(f"Error checking bucket {{bucket_id}}: {{str(e)}}") + self.logger.error(f"Error checking bucket {bucket_id}: {str(e)}") return False def list_bucket_contents(self, bucket_id: str, path: str = "") -> Dict: """List contents of a bucket at specified path""" try: - self.logger.info(f"Listing contents of bucket {{bucket_id}} at path {{path}}") + self.logger.info(f"Listing contents of bucket {bucket_id} at path {path}") contents = self.client.supabase.storage.from_(bucket_id).list(path) - return {{ + return { "folders": [item for item in contents if item.get("id", "").endswith("/")], "files": [item for item in contents if not item.get("id", "").endswith("/")] - }} + } except Exception as e: - self.logger.error(f"Error listing bucket contents: {{str(e)}}") + self.logger.error(f"Error listing bucket contents: {str(e)}") raise StorageError(str(e)) def upload_file(self, bucket_id: str, file_path: str, file_data: bytes, content_type: str, upsert: bool = True) -> Any: """Upload a file to a storage bucket""" try: - self.logger.info(f"Uploading file to {{bucket_id}} at path {{file_path}}") + self.logger.info(f"Uploading file to {bucket_id} at path {file_path}") return self.client.supabase.storage.from_(bucket_id).upload( path=file_path, file=file_data, - file_options={{ + file_options={ "content-type": content_type, "x-upsert": "true" if upsert else "false" - }} + } ) except Exception as e: - self.logger.error(f"Error uploading file: {{str(e)}}") + self.logger.error(f"Error uploading file: {str(e)}") raise StorageError(str(e)) def download_file(self, bucket_id: str, file_path: str) -> bytes: """Download a file from a storage bucket""" try: - self.logger.info(f"Downloading file from {{bucket_id}} at path {{file_path}}") + self.logger.info(f"Downloading file from {bucket_id} at path {file_path}") return self.client.supabase.storage.from_(bucket_id).download(file_path) except Exception as e: - self.logger.error(f"Error downloading file: {{str(e)}}") + self.logger.error(f"Error downloading file: {str(e)}") raise StorageError(str(e)) def delete_file(self, bucket_id: str, file_path: str) -> None: """Delete a file from a storage bucket""" try: - self.logger.info(f"Deleting file from {{bucket_id}} at path {{file_path}}") + self.logger.info(f"Deleting file from {bucket_id} at path {file_path}") self.client.supabase.storage.from_(bucket_id).remove([file_path]) except Exception as e: - self.logger.error(f"Error deleting file: {{str(e)}}") + self.logger.error(f"Error deleting file: {str(e)}") raise StorageError(str(e)) def get_public_url(self, bucket_id: str, file_path: str) -> str: """Get public URL for a file""" try: - self.logger.info(f"Getting public URL for file in {{bucket_id}} at path {{file_path}}") + self.logger.info(f"Getting public URL for file in {bucket_id} at path {file_path}") return self.client.supabase.storage.from_(bucket_id).get_public_url(file_path) except Exception as e: - self.logger.error(f"Error getting public URL: {{str(e)}}") + self.logger.error(f"Error getting public URL: {str(e)}") raise StorageError(str(e)) def create_signed_url(self, bucket_id: str, file_path: str, expires_in: int = 3600) -> Any: """Create a signed URL for temporary file access""" try: - self.logger.info(f"Creating signed URL for file in {{bucket_id}} at path {{file_path}}") + self.logger.info(f"Creating signed URL for file in {bucket_id} at path {file_path}") return self.client.supabase.storage.from_(bucket_id).create_signed_url(file_path, expires_in) except Exception as e: - self.logger.error(f"Error creating signed URL: {{str(e)}}") + self.logger.error(f"Error creating signed URL: {str(e)}") raise StorageError(str(e)) class StorageAdmin(StorageManager): @@ -115,9 +115,9 @@ class StorageAdmin(StorageManager): ) -> Dict[str, Any]: """Create a new storage bucket with supported parameters.""" try: - self.logger.info(f"Creating bucket {{id}} with name {{name}}") + self.logger.info(f"Creating bucket {id} with name {name}") - options: Optional[CreateBucketOptions] = {{}} + options: Optional[CreateBucketOptions] = {} if public: options["public"] = public if file_size_limit is not None: @@ -133,7 +133,7 @@ class StorageAdmin(StorageManager): return bucket except Exception as e: - self.logger.error(f"Error creating bucket {{id}}: {{str(e)}}") + self.logger.error(f"Error creating bucket {id}: {str(e)}") raise StorageError(str(e)) def initialize_core_buckets(self, admin_user_id: Optional[str] = None) -> List[Dict[str, Any]]: @@ -144,7 +144,7 @@ class StorageAdmin(StorageManager): raise ValueError("Admin user ID is required for bucket initialization") core_buckets = [ - {{ + { "id": "cc.users", "name": "CC Users", "public": False, @@ -156,8 +156,8 @@ class StorageAdmin(StorageManager): 'application/msword', 'application/vnd.openxmlformats-officedocument.*', 'text/plain', 'text/csv', 'application/json' ] - }}, - {{ + }, + { "id": "cc.institutes", "name": "CC Institutes", "public": False, @@ -169,7 +169,7 @@ class StorageAdmin(StorageManager): 'application/msword', 'application/vnd.openxmlformats-officedocument.*', 'text/plain', 'text/csv', 'application/json' ] - }} + } ] results = [] @@ -177,30 +177,30 @@ class StorageAdmin(StorageManager): try: bucket_name = bucket.pop("name") result = self.create_bucket(name=bucket_name, **bucket) - results.append({{ + results.append({ "bucket": bucket["id"], "status": "success", "result": result - }}) + }) except Exception as e: - self.logger.error(f"Error creating bucket {{bucket['id']}}: {{str(e)}}") - results.append({{ + self.logger.error(f"Error creating bucket {bucket['id']}: {str(e)}") + results.append({ "bucket": bucket["id"], "status": "error", "error": str(e) - }}) + }) return results except Exception as e: - self.logger.error(f"Error initializing core buckets: {{str(e)}}") + self.logger.error(f"Error initializing core buckets: {str(e)}") raise StorageError(str(e)) def create_user_bucket(self, user_id: str, username: str) -> Dict[str, Any]: """Create a storage bucket for a specific user.""" try: - bucket_id = f"cc.users.admin.{{username}}" - bucket_name = f"User Files - {{username}}" + bucket_id = f"cc.users.admin.{username}" + bucket_name = f"User Files - {username}" return self.create_bucket( id=bucket_id, @@ -217,7 +217,7 @@ class StorageAdmin(StorageManager): ) except Exception as e: - self.logger.error(f"Error creating user bucket for {{username}}: {{str(e)}}") + self.logger.error(f"Error creating user bucket for {username}: {str(e)}") raise StorageError(str(e)) def create_school_buckets(self, school_id: str, school_name: str, admin_user_id: Optional[str] = None) -> Dict[str, Any]: @@ -228,9 +228,9 @@ class StorageAdmin(StorageManager): raise ValueError("Admin user ID is required for school bucket creation") school_buckets = [ - {{ - "id": f"cc.institutes.{{school_id}}.public", - "name": f"{{school_name}} - Public Files", + { + "id": f"cc.institutes.{school_id}.public", + "name": f"{school_name} - Public Files", "public": True, "owner": owner_id, "owner_id": school_id, @@ -240,10 +240,10 @@ class StorageAdmin(StorageManager): 'application/msword', 'application/vnd.openxmlformats-officedocument.*', 'text/plain', 'text/csv', 'application/json' ] - }}, - {{ - "id": f"cc.institutes.{{school_id}}.private", - "name": f"{{school_name}} - Private Files", + }, + { + "id": f"cc.institutes.{school_id}.private", + "name": f"{school_name} - Private Files", "public": False, "owner": owner_id, "owner_id": school_id, @@ -253,29 +253,29 @@ class StorageAdmin(StorageManager): 'application/msword', 'application/vnd.openxmlformats-officedocument.*', 'text/plain', 'text/csv', 'application/json' ] - }} + } ] - results = {{}} + results = {} for bucket in school_buckets: try: bucket_name = bucket.pop("name") result = self.create_bucket(name=bucket_name, **bucket) - results[bucket["id"]] = {{ + results[bucket["id"]] = { "status": "success", "result": result - }} + } except Exception as e: - self.logger.error(f"Error creating school bucket {{bucket['id']}}: {{str(e)}}") - results[bucket["id"]] = {{ + self.logger.error(f"Error creating school bucket {bucket['id']}: {str(e)}") + results[bucket["id"]] = { "status": "error", "error": str(e) - }} + } return results except Exception as e: - self.logger.error(f"Error creating school buckets: {{str(e)}}") + self.logger.error(f"Error creating school buckets: {str(e)}") raise StorageError(str(e)) class StorageUser(StorageManager):