Skip to content

Instantly share code, notes, and snippets.

@pkubik
Last active July 1, 2022 11:04
Show Gist options
  • Save pkubik/fb95277639e46e01fb2ca10b49f6ad1b to your computer and use it in GitHub Desktop.
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?
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