codemicro/circuitbodge
Archived
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Configuration tweaks

Browse source 
Signed-off-by: AKP <tom@tdpain.net>
This commit is contained in:
akp 2022-11-08 11:10:45 +00:00
parent b44f25fd74
commit 0f83fec132
No known key found for this signature in database
GPG key ID: AA5726202C8879B7
2 changed files with 32 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

10
circuit-laundry-notifier/__main__.py
View file

@ -1,5 +1,11 @@
import os
from web import new
DEBUG = bool(os.environ.get("CIRCUIT__DEBUG", False))
PORT = int(os.environ.get("CIRCUIT__HTTP_PORT", 8080))
HOST = os.environ.get("CIRCUIT__HTTP_HOST", "127.0.0.1")
if __name__ == "__main__":
app = new()
app.run(port=8080, host="127.0.0.1")
app = new(debug=DEBUG)
app.run(port=PORT, host=HOST, debug=DEBUG)

27
circuit-laundry-notifier/web.py
View file

@ -1,3 +1,4 @@
import os
import uuid
from dataclasses import dataclass
from threading import Lock
@ -8,7 +9,7 @@ import flask
from circuit_scraper import CircuitScraper, Machine, ScraperError, MachineState
def new() -> flask.Flask:
def new(debug=False) -> flask.Flask:
app = flask.Flask(__name__)
app.add_url_rule(
@ -19,11 +20,15 @@ def new() -> flask.Flask:
view_func=_api_register_watcher,
methods=["POST"],
)
app.add_url_rule("/api/v1/watcher", view_func=_api_list_watched, methods=["GET"])
app.add_url_rule(
"/api/v1/watcher/run", view_func=_api_run_watcher, methods=["POST"]
)
if debug:
app.add_url_rule(
"/api/v1/watcher", view_func=_api_list_watched, methods=["GET"]
)
return app
@ -77,7 +82,7 @@ def _api_register_watcher(site_id: str) -> Union[any, Tuple[any, int]]:
jobs.append(ws)
job_lock.release()
return flask.jsonify(ws), 204
return flask.jsonify(ws), 200
def _api_list_watched() -> Union[any, Tuple[any, int]]:
@ -87,7 +92,23 @@ def _api_list_watched() -> Union[any, Tuple[any, int]]:
return res
WATCHER_BEARER_TOKEN = os.environ.get("CIRCUIT__WATCHER_TOKEN")
def _api_run_watcher() -> Union[any, Tuple[any, int]]:
if WATCHER_BEARER_TOKEN is not None:
auth_header = flask.request.headers.get("Authorization")
if auth_header is None:
return (
flask.jsonify({"ok": False, "message": "missing Authorization header"}),
401,
)
if auth_header != f"Bearer {WATCHER_BEARER_TOKEN}":
return (
flask.jsonify({"ok": False, "message": "bad Authorization token"}),
401,
)
global jobs, job_lock
job_lock.acquire()