Spaces:
Sleeping
Sleeping
File size: 1,460 Bytes
79ff0e6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 |
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))
|