Upload 5 files

app.py

@@ -0,0 +1,53 @@
+from fastapi import FastAPI,HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+import uvicorn
+from pydantic import BaseModel, Field
+from workflowschema import LibraryModel
+from workflow_invocation import WorkflowInvocation
+import asyncio
+import json
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allows all origins
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.post("/api/v1/trigger_email_workflow")
+def trigger_workflow(payLoad: LibraryModel):
+    try:
+        if payLoad:
+            wf_invocation = WorkflowInvocation(payLoad.model_dump())
+            asyncio.run(wf_invocation.invoke_workflow()) 
+            return JSONResponse(content={"message": "workflow executed successfully"}, status_code=200)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+    
+if __name__ == "__main__":
+    uvicorn.run(app, host = "0.0.0.0", port = 8000)
+#     schema_input = {
+#   "name": "Media",
+#   "action": "modified",
+#   "change_type": "MediaStatus",
+#   "changed_by": "focus",
+#   "timestamp": "2025-07-23T13:45:00Z",
+#   "change_id": 1001,
+#   "change_from": "ON",
+#   "change_to": "OFF",
+#   "unique_identifier": "library",
+#   "title":"OCA"
+#   }
+#     result = trigger_workflow(schema_input)
+#     print(json.loads(result.body))
+
+
+
+
+
+
+

config.yaml

@@ -0,0 +1,2 @@
+workflow:
+   webhook_url: "http://localhost:5678/webhook/5f90600a-ba81-4521-a90b-3852d3613181"

requirements.txt

@@ -0,0 +1,5 @@
+fastapi
+uvicorn
+pydantic
+httpx
+pyyaml

workflow_invocation.py

@@ -0,0 +1,30 @@
+from workflowschema import LibraryModel
+import httpx
+import asyncio
+import yaml
+
+
+class WorkflowInvocation:
+    def __init__(self, payload):
+        self.config = self.load_config()
+        self.payload = payload
+        
+    def load_config(self):
+        with open("config.yaml","r") as file:
+            config = yaml.safe_load(file)
+        return config
+
+    
+    async def invoke_workflow(self): 
+        try:   
+            webHookUrl = self.config['workflow']['webhook_url']   
+    
+            async with httpx.AsyncClient() as client:
+                response = await client.post(webHookUrl, json=self.payload)
+                response.raise_for_status()
+                json_response = response.json()
+                return json_response
+                #print("Response status code:", response.json())
+        except Exception as e:
+            print(f"An error occurred: {e}")  
+    

workflowschema.py

@@ -0,0 +1,15 @@
+from pydantic import BaseModel, Field
+from typing import List,Literal,Any
+
+class LibraryModel(BaseModel):
+    name: str = Field(..., description="Name of the module that has changed")
+    action: Literal["added", "removed", "modified"] = Field(..., description="Action performed on that module", examples=["added", "removed", "modified"])
+    change_type: str = Field(..., description="The field or the data that has changed in the module")
+    changed_by: str = Field(..., description="The user who made the change")
+    timestamp: str = Field(..., description="Timestamp of the change in ISO format")
+    change_id: int = Field(..., description="Unique identifier for the change in the module")
+    change_from: Any = Field(..., description="The previous value of the field that has changed")
+    change_to: Any = Field(..., description="The new value of the field that has changed") 
+    title: str = Field(..., description="Title of the module that has changed")
+    unique_identifier: Literal["library","scheduling"] = Field(..., description="Info about the module that has changed", examples=["library", "scheduling"])
+