Last active
July 1, 2022 11:04
-
-
Save pkubik/fb95277639e46e01fb2ca10b49f6ad1b to your computer and use it in GitHub Desktop.
Setup Flask HTTP server as backend for simple counters storage and use fzf+curl for frontent. I have my reasons, ok?
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 __future__ import annotations | |
from flask import Flask | |
from flask import request | |
from flask import abort | |
from multiprocessing import Process | |
import subprocess as sp | |
import sys | |
import os | |
from dataclasses import dataclass, field | |
@dataclass | |
class Job: | |
name: str | |
node: str | |
counter: int = 0 | |
@dataclass | |
class K8s: | |
db: dict[str, Job] | |
entries: dict[str, str] = field(default_factory=dict) | |
def __post_init__(self): | |
self.update_entries() | |
def update_entries(self): | |
self.entries = { | |
f"{j.name}{j.counter:>20} {j.node}": j.name for j in self.db.values() | |
} | |
def decode_entry(self, encoded_entry): | |
return k8s.entries[encoded_entry.strip().strip("'")] | |
def by_entry(self, entry): | |
return k8s.db[self.decode_entry(entry)] | |
def print(self): | |
return "\n".join(k8s.entries) | |
def increase_counter(self, entries): | |
for entry in entries: | |
k8s.by_entry(entry).counter += 1 | |
self.update_entries() | |
return "\n".join(k8s.entries) | |
k8s = K8s({ | |
f"job{i}": Job(f"job{i}", "main", i) for i in range(5) | |
}) | |
app = Flask(__name__) | |
@app.route("/inc") | |
def inc(): | |
entry = request.args.get("entry") | |
if entry is None: | |
abort(404) | |
k8s.by_entry(entry).counter += 1 | |
k8s.update_entries() | |
return k8s.print() | |
@app.route("/dec") | |
def dec(): | |
entry = request.args.get("entry") | |
if entry is None: | |
abort(404) | |
k8s.by_entry(entry).counter -= 1 | |
k8s.update_entries() | |
return k8s.print() | |
@app.route("/info") | |
def info(): | |
entry = request.args.get("entry") | |
if entry is None: | |
abort(404) | |
job = k8s.by_entry(entry) | |
return str(job) | |
@app.route("/") | |
def base(): | |
return k8s.print() | |
def run_server(): | |
sys.stderr = open("/tmp/nihao-server-log.out", "w") | |
sys.stdout = open("/tmp/nihao-server-log.out", "a") | |
print(f"Starting server at PID {os.getpid()}") | |
app.run() | |
server = Process(target=run_server) | |
server.start() | |
def build_command(name="", add_entry=True): | |
if name: | |
name = f"/{name}" | |
entry_substr = " " | |
if add_entry: | |
entry_substr += ' --data-urlencode "entry={}" ' | |
return f"curl --get localhost:5000{name}{entry_substr}--fail 2>/dev/null" | |
GET_CMD = build_command(add_entry=False) | |
INC_CMD = build_command("inc") | |
DEC_CMD = build_command("dec") | |
INFO_CMD = build_command("info") | |
fzf = sp.run( | |
[ | |
"fzf", | |
"--bind", | |
f"ctrl-r:reload({GET_CMD})", | |
"--bind", | |
f"ctrl-i:reload({INC_CMD} || {GET_CMD})", | |
"--bind", | |
f"ctrl-d:reload({DEC_CMD} || {GET_CMD})", | |
], | |
stdout=sp.PIPE, | |
input=base(), | |
encoding="UTF-8", | |
) | |
if fzf.returncode == 0: | |
# could just read it directly, but let's pretend MVC | |
job_info = sp.check_output( | |
INFO_CMD.format(fzf.stdout.strip()), | |
shell=True, | |
stderr=sp.DEVNULL, | |
encoding="UTF-8", | |
) | |
server.terminate() | |
print(job_info) | |
else: | |
server.terminate() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment