mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 12:33:13 +00:00
132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
"""Production Flask app for afterhours-shift-manager.
|
|
|
|
Local: PYTHONPATH=src:src/shared python3 -m server.app
|
|
Prod: gunicorn server.wsgi:app
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, Response, jsonify, request
|
|
|
|
import shared.sentry_init # noqa: F401
|
|
from shared.portal_http import encode_body, handle as portal_handle
|
|
from shared.roster_http import (
|
|
AuthError,
|
|
SecretUnavailable,
|
|
authorize_bearer,
|
|
remove,
|
|
upsert,
|
|
validate_extension,
|
|
)
|
|
from shared.secrets import get_secret
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_SLACK_BOT_DIR = Path(__file__).resolve().parents[1] / "slack-bot"
|
|
if str(_SLACK_BOT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(_SLACK_BOT_DIR))
|
|
|
|
|
|
def _slack_handler():
|
|
from slack_bolt.adapter.flask import SlackRequestHandler
|
|
|
|
from app import create_app as create_bolt_app
|
|
|
|
bot_token = get_secret(os.environ["SLACK_BOT_TOKEN_SECRET"])
|
|
signing_secret = get_secret(os.environ["SLACK_SIGNING_SECRET"])
|
|
schedule_channel = os.environ["SHIFT_CHANNEL"]
|
|
bolt_app = create_bolt_app(
|
|
bot_token, signing_secret, schedule_channel=schedule_channel
|
|
)
|
|
return SlackRequestHandler(bolt_app)
|
|
|
|
|
|
def create_app() -> Flask:
|
|
app = Flask(__name__)
|
|
slack_handler = None
|
|
|
|
def _get_slack_handler():
|
|
nonlocal slack_handler
|
|
if slack_handler is None:
|
|
slack_handler = _slack_handler()
|
|
return slack_handler
|
|
|
|
@app.route("/api/health")
|
|
def health():
|
|
return jsonify(
|
|
{
|
|
"stage": os.environ.get("STAGE", "local"),
|
|
"sha": os.environ.get("GIT_SHA", "dev"),
|
|
}
|
|
)
|
|
|
|
@app.route("/slack/events", methods=["POST"])
|
|
def slack_events():
|
|
return _get_slack_handler().handle(request)
|
|
|
|
@app.route("/api/shifts", methods=["GET", "POST", "DELETE", "OPTIONS"])
|
|
@app.route("/api/shifts/<path:rest>", methods=["GET", "POST", "DELETE", "OPTIONS"])
|
|
def portal(rest: str | None = None):
|
|
status, headers, payload = portal_handle(
|
|
method=request.method,
|
|
path=request.path,
|
|
origin=request.headers.get("Origin", ""),
|
|
authorization=request.headers.get("Authorization", ""),
|
|
query=request.args.to_dict(flat=True),
|
|
body=request.get_data(),
|
|
)
|
|
if status == 204:
|
|
return Response(b"", status=204, headers=headers)
|
|
return Response(
|
|
encode_body(payload),
|
|
status=status,
|
|
headers=headers,
|
|
mimetype="application/json",
|
|
content_type="application/json",
|
|
)
|
|
|
|
@app.route("/roster", methods=["PUT"])
|
|
def roster_put():
|
|
try:
|
|
authorize_bearer(request.headers.get("Authorization", ""))
|
|
except AuthError:
|
|
return jsonify({"error": "unauthorized"}), 401
|
|
except SecretUnavailable:
|
|
return jsonify({"error": "service unavailable"}), 503
|
|
try:
|
|
upsert(request.get_data())
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "invalid request"}), 400
|
|
return jsonify({"ok": True}), 200
|
|
|
|
@app.route("/roster/<extension>", methods=["DELETE"])
|
|
def roster_delete(extension: str):
|
|
try:
|
|
authorize_bearer(request.headers.get("Authorization", ""))
|
|
except AuthError:
|
|
return jsonify({"error": "unauthorized"}), 401
|
|
except SecretUnavailable:
|
|
return jsonify({"error": "service unavailable"}), 503
|
|
try:
|
|
remove(validate_extension(extension))
|
|
except (TypeError, ValueError):
|
|
return jsonify({"error": "invalid request"}), 400
|
|
return Response(b"", status=204)
|
|
|
|
return app
|
|
|
|
|
|
app = create_app()
|
|
|
|
|
|
def main():
|
|
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|