Refactor Firebase function implementation and update dependencies

Author: Zaffer

services/_templates/web_app/frontend/functions_python/.gitignore

@@ -1 +1 @@
-*.local
+*.local

services/_templates/web_app/frontend/functions_python/firestore_onwrite_processor/common.py

@@ -0,0 +1,48 @@
+from enum import Enum
+from typing import Any, Dict, Callable, TypeVar, Union, Optional
+from google.cloud.firestore import DocumentSnapshot, GeoPoint, SERVER_TIMESTAMP
+from firebase_functions.firestore_fn import Event, Change
+
+# Type definitions
+FirestoreField = Union[str, int, float, bool, dict, list, GeoPoint, None]
+
+class ChangeType(str, Enum):
+    CREATE = "CREATE"
+    UPDATE = "UPDATE"
+    DELETE = "DELETE"
+
+class State(str, Enum):
+    PROCESSING = "PROCESSING"
+    COMPLETED = "COMPLETED"
+    ERROR = "ERROR"
+
+def now():
+    return SERVER_TIMESTAMP
+
+def get_change_type(change: Change) -> ChangeType:
+    if not change.before or not change.before.exists:
+        return ChangeType.CREATE
+    if not change.after or not change.after.exists:
+        return ChangeType.DELETE
+    return ChangeType.UPDATE
+
+def is_delete(change: Change) -> bool:
+    return get_change_type(change) == ChangeType.DELETE
+
+def is_update(change: Change) -> bool:
+    return get_change_type(change) == ChangeType.UPDATE
+
+def is_create(change: Change) -> bool:
+    return get_change_type(change) == ChangeType.CREATE
+
+def safe_get(document, field_path, default=None):
+    """Safely get a field from a document without KeyError."""
+    if document is None:
+        return default
+    if not hasattr(document, 'get'):
+        return default
+        
+    try:
+        return document.get(field_path)
+    except KeyError:
+        return default

services/_templates/web_app/frontend/functions_python/firestore_onwrite_processor/processor.py

@@ -0,0 +1,126 @@
+from typing import Any, Dict, Callable, TypeVar, Generic, Optional, Awaitable
+from firebase_functions.firestore_fn import Event, Change, DocumentSnapshot
+from .common import ChangeType, State, now, get_change_type, FirestoreField, safe_get
+
+T = TypeVar('T')
+TOutput = TypeVar('TOutput', bound=Dict[str, Any])
+
+class ProcessConfig:
+    def __init__(
+        self,
+        input_field: str,
+        process_fn: Callable[[Any, Event], Awaitable[Dict[str, FirestoreField]]],
+        error_fn: Callable[[Any], str],
+        status_field: Optional[str] = None,
+        order_field: Optional[str] = None
+    ):
+        self.input_field = input_field
+        self.process_fn = process_fn
+        self.error_fn = error_fn
+        self.status_field = status_field
+        self.order_field = order_field
+
+class FirestoreOnWriteProcessor(Generic[T, TOutput]):
+    def __init__(self, options: ProcessConfig):
+        self.input_field = options.input_field or 'prompt'
+        self.order_field = options.order_field or 'createTime'
+        self.status_field = options.status_field or 'status'
+        self.process_fn = options.process_fn
+        self.process_updates = True
+        self.error_fn = options.error_fn
+
+    def should_process(self, change: Change) -> bool:
+        change_type = get_change_type(change)
+        if change_type == ChangeType.DELETE:
+            return False
+
+        # Extract status if it exists
+        status = safe_get(change.after, self.status_field)
+        state: State = status.get("state") if isinstance(status, dict) else None
+        new_value = safe_get(change.after, self.input_field)
+        old_value = safe_get(change.before, self.input_field)
+
+        has_changed = (
+            change_type == ChangeType.CREATE or
+            (self.process_updates and 
+             change_type == ChangeType.UPDATE and 
+             old_value != new_value)
+        )
+
+        if (
+            not new_value or
+            state in [State.PROCESSING.value, State.COMPLETED.value, State.ERROR.value] or
+            not has_changed or
+            not isinstance(new_value, str)
+        ):
+            return False
+        
+        return True
+
+    async def write_start_event(self, event: Event[Change[DocumentSnapshot]]) -> None:
+        create_time = event.data.after.create_time
+        update_time = now()
+
+        status = {
+            "state": State.PROCESSING.value,
+            "startTime": update_time,
+            "updateTime": update_time
+        }
+
+        start_data = safe_get(event.data.after, self.order_field)
+        # Prepare update data
+        if start_data:
+            update_data = {self.status_field: status}
+        else:
+            update_data = {
+                self.order_field: create_time, 
+                self.status_field: status
+            }
+
+        event.data.after.reference.update(update_data)
+
+    async def write_completion_event(self, change: Change, output: Dict[str, Any]) -> None:
+        update_time = now()
+        
+        # In Firebase Python, we need to use dot notation as strings
+        update_data = dict(output)  # Create a copy to avoid modifying the original
+        update_data[f"{self.status_field}.state"] = State.COMPLETED.value
+        update_data[f"{self.status_field}.updateTime"] = update_time
+        update_data[f"{self.status_field}.completeTime"] = update_time
+        
+        change.after.reference.update(update_data)
+
+    async def write_error_event(self, change: Change, e: Any) -> None:
+        event_timestamp = now()
+        error_message = self.error_fn(e)
+        
+        change.after.reference.update({
+            self.status_field: {
+                "state": State.ERROR.value,
+                "updateTime": event_timestamp,
+                "error": error_message
+            }
+        })
+
+    async def run(self, event: Event) -> None:
+        if not event:
+            print("No event data")
+            return
+        
+        if not event.data:
+            print("No document event.data")
+            return
+
+        if not self.should_process(event.data):
+            return
+
+        try:
+            await self.write_start_event(event)
+            
+            input_data = safe_get(event.data.after, self.input_field)
+            output = await self.process_fn(input_data, event)
+            
+            await self.write_completion_event(event.data, output)
+        except Exception as e:
+            print(f"Message processing error: {e}")
+            await self.write_error_event(event.data, e)

services/_templates/web_app/frontend/functions_python/main.py

@@ -1,13 +1,46 @@
-# Welcome to Cloud Functions for Firebase for Python!
-# To get started, simply uncomment the below code or create your own.
-# Deploy with `firebase deploy`
+import asyncio
 
-from firebase_functions import https_fn
+from firebase_functions import https_fn, firestore_fn
 from firebase_admin import initialize_app
+from typing import Dict, Any
+from firestore_onwrite_processor.processor import (
+    FirestoreOnWriteProcessor,
+    ProcessConfig,
+)
+from firestore_onwrite_processor.common import FirestoreField
 
 initialize_app()
 
 
+# Example implementation of a process function
+async def process_document(
+    input_value: str, event: firestore_fn.Event
+) -> Dict[str, FirestoreField]:
+    # Your processing logic here
+    print(f"Processing document with input: {input_value}")
+    return {"result": f"Processed: {input_value}"}
+
+
+# Example error handler
+def handle_error(error: Any) -> str:
+    return f"Processing error: {str(error)}"
+
+
+# Create the processor
+processor_config = ProcessConfig(
+    input_field="prompt", process_fn=process_document, error_fn=handle_error
+)
+
+document_processor = FirestoreOnWriteProcessor(processor_config)
+
+
+@firestore_fn.on_document_written(document="collection/{docId}")
+def process_firestore_document(event: firestore_fn.Event) -> None:
+    """Process a document when it's written to Firestore."""
+    asyncio.run(document_processor.run(event))
+
+
+# example function
 @https_fn.on_request()
 def on_request_example(req: https_fn.Request) -> https_fn.Response:
     return https_fn.Response("Hello world!")

services/_templates/web_app/frontend/functions_python/requirements.txt

@@ -1 +1,3 @@
-firebase_functions~=0.1.0
+firebase_functions>=0.4.2
+firebase_admin>=6.7.0
+# google-cloud-firestore>=2.11.0