Browse the data

1from fastapi import FastAPI, HTTPException, Header
2from pydantic import BaseModel
3import secrets
4import time
5from datetime import datetime
6 
7app = FastAPI()
8 
9users = {}
10tokens = {}
11workflows = {}
12workflow_id_counter = 1
13execution_logs = []
14execution_id_counter = 1
15 
16class SignupRequest(BaseModel):
17    username: str
18    password: str
19 
20class LoginRequest(BaseModel):
21    username: str
22    password: str
23 
24class WorkflowCreate(BaseModel):
25    trigger: dict
26    actions: list
27 
28class WorkflowUpdate(BaseModel):
29    trigger: dict = None
30    actions: list = None
31 
32def get_current_user(authorization: str = Header(None)):
33    if not authorization:
34        raise HTTPException(status_code=401, detail="Missing auth header")
35    token = authorization.replace("Bearer ", "")
36    if token not in tokens:
37        raise HTTPException(status_code=401, detail="Invalid token")
38    return tokens[token]
39 
40@app.post("/signup")
41def signup(req: SignupRequest):
42    if req.username in users:
43        raise HTTPException(status_code=400, detail="User already exists")
44    users[req.username] = {"password": req.password}
45    return {"message": "User created"}
46 
47@app.post("/login")
48def login(req: LoginRequest):
49    user = users.get(req.username)
50    if not user or user["password"] != req.password:
51        raise HTTPException(status_code=401, detail="Invalid credentials")
52    token = secrets.token_hex(16)
53    tokens[token] = req.username
54    return {"token": token}
55 
56@app.post("/workflows")
57def create_workflow(workflow: WorkflowCreate, authorization: str = Header(None)):
58    global workflow_id_counter
59    user = get_current_user(authorization)
60    wid = workflow_id_counter
61    workflow_id_counter += 1
62    workflows[wid] = {
63        "id": wid,
64        "trigger": workflow.trigger,
65        "actions": workflow.actions,
66        "created_by": user,
67        "created_at": datetime.utcnow().isoformat()
68    }
69    return workflows[wid]
70 
71@app.get("/workflows/{workflow_id}")
72def get_workflow(workflow_id: int, authorization: str = Header(None)):
73    get_current_user(authorization)
74    wf = workflows.get(workflow_id)
75    if not wf:
76        raise HTTPException(status_code=404, detail="Workflow not found")
77    return wf
78 
79@app.post("/workflows/{workflow_id}/execute")
80def execute_workflow(workflow_id: int, authorization: str = Header(None)):
81    global execution_id_counter
82    user = get_current_user(authorization)
83    wf = workflows.get(workflow_id)
84    if not wf:
85        raise HTTPException(status_code=404, detail="Workflow not found")
86    eid = execution_id_counter
87    execution_id_counter += 1
88    log_entry = {
89        "id": eid,
90        "workflow_id": workflow_id,
91        "trigger": wf["trigger"],
92        "actions": wf["actions"],
93        "status": "executed",
94        "executed_by": user,
95        "executed_at": datetime.utcnow().isoformat()
96    }
97    execution_logs.append(log_entry)
98    return log_entry
99 
100@app.get("/executions")
101def get_executions(authorization: str = Header(None)):
102    get_current_user(authorization)
103    return execution_logs
104 
105@app.get("/executions/{execution_id}")
106def get_execution(execution_id: int, authorization: str = Header(None)):
107    get_current_user(authorization)
108    for log in execution_logs:
109        if log["id"] == execution_id:
110            return log
111    raise HTTPException(status_code=404, detail="Execution not found")

1fastapi
2uvicorn