Created
November 5, 2025 03:49
-
-
Save arivictor/485c89e78c4b8520c95112094fd00a98 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from flask import Flask, render_template_string, request | |
| app = Flask(__name__) | |
| # In-memory storage | |
| WORKFLOWS = [] | |
| NEXT_ID = 1 | |
| HTML_BASE = """ | |
| <!doctype html> | |
| <html> | |
| <head> | |
| <title>Workflow System</title> | |
| <script src="https://unpkg.com/[email protected]"></script> | |
| </head> | |
| <body> | |
| <h1>Workflow System</h1> | |
| <div id="content"> | |
| {{ body|safe }} | |
| </div> | |
| </body> | |
| </html> | |
| """ | |
| def render(body): | |
| return render_template_string(HTML_BASE, body=body) | |
| def controls(state, wid): | |
| if state == "new": | |
| return f"<a hx-post='/workflows/{wid}/start' hx-target='#content'><button>Start</button></a>" | |
| if state == "running": | |
| return f""" | |
| <a hx-post='/workflows/{wid}/pause' hx-target='#content'><button>Pause</button></a> | | |
| <a hx-post='/workflows/{wid}/stop' hx-target='#content'><button>Stop</button></a> | |
| """ | |
| if state == "paused": | |
| return f""" | |
| <a hx-post='/workflows/{wid}/resume' hx-target='#content'><button>Resume</button></a> | | |
| <a hx-post='/workflows/{wid}/stop' hx-target='#content'><button>Stop</button></a> | |
| """ | |
| if state == "stopped": | |
| return f"<a hx-post='/workflows/{wid}/restart' hx-target='#content'><button>Restart</button></a>" | |
| return "" | |
| def get_home_content(): | |
| """Generate the home page content.""" | |
| links = "".join( | |
| f"<li><a hx-get='/workflows/{w['id']}' hx-target='#content'><button>{w['name']} ({w['state']})</button></a></li>" | |
| for w in WORKFLOWS | |
| ) or "<p>No workflows yet.</p>" | |
| return f""" | |
| <h2>All Workflows</h2> | |
| <ul>{links}</ul> | |
| <p><a hx-get='/new' hx-target='#content'><button>Create new workflow</button></a></p> | |
| """ | |
| @app.route("/") | |
| def home(): | |
| body = get_home_content() | |
| return body if "HX-Request" in request.headers else render(body) | |
| @app.route("/new") | |
| def new_workflow_form(): | |
| body = """ | |
| <h2>Create Workflow</h2> | |
| <form hx-post="/create" hx-target="#content"> | |
| <label>Name: <input type="text" name="name" required></label> | |
| <button type="submit">Create</button> | |
| </form> | |
| <p><a hx-get='/' hx-target='#content'><button>Cancel</button></a></p> | |
| """ | |
| return body if "HX-Request" in request.headers else render(body) | |
| @app.route("/create", methods=["POST"]) | |
| def create_workflow(): | |
| global NEXT_ID | |
| name = request.form.get("name", "Untitled") | |
| WORKFLOWS.append({"id": NEXT_ID, "name": name, "state": "new"}) | |
| NEXT_ID += 1 | |
| return get_home_content() | |
| @app.route("/workflows/<int:wid>") | |
| def show_workflow(wid): | |
| w = next((x for x in WORKFLOWS if x["id"] == wid), None) | |
| if not w: | |
| return "Not found", 404 | |
| body = f""" | |
| <h2>{w['name']}</h2> | |
| <p>State: <strong>{w['state']}</strong></p> | |
| <p>{controls(w['state'], w['id'])}</p> | |
| <p><a hx-get='/' hx-target='#content'><button>Back to list</button></a></p> | |
| """ | |
| return body if "HX-Request" in request.headers else render(body) | |
| # State transitions | |
| def update_state(wid, new_state): | |
| w = next((x for x in WORKFLOWS if x["id"] == wid), None) | |
| if w: | |
| w["state"] = new_state | |
| return show_workflow(wid) | |
| @app.route("/workflows/<int:wid>/start", methods=["POST"]) | |
| def start(wid): return update_state(wid, "running") | |
| @app.route("/workflows/<int:wid>/pause", methods=["POST"]) | |
| def pause(wid): return update_state(wid, "paused") | |
| @app.route("/workflows/<int:wid>/resume", methods=["POST"]) | |
| def resume(wid): return update_state(wid, "running") | |
| @app.route("/workflows/<int:wid>/stop", methods=["POST"]) | |
| def stop(wid): return update_state(wid, "stopped") | |
| @app.route("/workflows/<int:wid>/restart", methods=["POST"]) | |
| def restart(wid): return update_state(wid, "running") | |
| if __name__ == "__main__": | |
| app.run(debug=True, port=8000) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment