import json as builtin_json
import logging
import os
import sys
from datetime import datetime
from enum import Enum
from pathlib import Path
from urllib.error import URLError
from urllib.request import urlopen
from flask import Flask, jsonify, render_template, request
from flask_cors import CORS
from flask_json import FlaskJSON
from metobsapi import data_api, files_api
from metobsapi.util import data_responses, file_responses
LOG = logging.getLogger(__name__)
app = Flask(__name__)
# Load custom configuration file is specified
app.config.from_object("metobsapi.common_config")
if os.environ.get("METOBSAPI_SETTINGS") is not None:
app.config.from_pyfile(os.environ.get("METOBSAPI_SETTINGS"))
app.config.from_prefixed_env(prefix="METOBSAPI")
# Load json handler and add custom enum encoder
json = FlaskJSON(app)
@json.encoder
def enum_encoder(o):
if isinstance(o, Enum):
return o.value
return None
# Allow for cross-domain access to the API using CORS
CORS(app, resources=r"/api/*", allow_headers="Content-Type")
@app.route("/api/")
def index():
"""Get Main App Documentation."""
return render_template("index.html")
@app.route("/api/files")
def files_index():
"""Get Files API Documentation."""
return render_template(
"files_index.html",
archive_info=file_responses.ARCHIVE_STREAMS,
instrument_streams=file_responses.INSTRUMENT_STREAMS,
)
@app.route("/api/data")
def data_index():
"""Get Data API Documentation."""
return render_template("data_index.html")
@app.errorhandler(data_api.BadHandlerFormat)
def bad_request_format(err: data_api.BadHandlerFormat):
return render_template("400.html", format=err.bad_format), 400
@app.errorhandler(data_api.FormattedBadRequest)
def formatted_bad_request(err: data_api.FormattedBadRequest):
return err.handler_func(None, None, err.response_info)
@app.errorhandler(404)
def page_not_found(_):
return render_template("404.html"), 404
@app.errorhandler(500)
def internal_server(_):
return render_template("500.html"), 500
@app.after_request
def apply_header(response):
response.headers[data_responses.api_version_header] = data_responses.api_version
return response
@app.route("/api/data.<fmt>", methods=["GET"])
def get_data(fmt):
begin_time = request.args.get("begin")
end_time = request.args.get("end")
site = request.args.get("site")
inst = request.args.get("inst")
symbols = request.args.get("symbols")
interval = request.args.get("interval", "1m")
sep = request.args.get("sep", ",")
order = request.args.get("order", "row")
epoch = request.args.get("epoch")
handler_func = data_api.get_format_handler(fmt, {"sep": sep, "order": order})
if isinstance(handler_func, tuple):
return handler_func
time_parameters = data_api.parse_time_parameters(handler_func, begin_time, end_time, interval, epoch)
return data_api.modify_data(handler_func, time_parameters, site, inst, symbols)
@app.route("/api/files.<fmt>", methods=["GET"])
def get_files(fmt):
begin_time = request.args.get("begin")
end_time = request.args.get("end")
dates = request.args.get("dates")
streams = request.args.get("streams")
return files_api.find_stream_files(fmt, begin_time, end_time, dates, streams)
@app.route("/api/archive/info", methods=["GET"])
def get_archive_info():
return jsonify(
{
"code": 200,
"message": "",
"sites": file_responses.ARCHIVE_INFO,
},
)
@app.route("/api/status", methods=["GET"])
def status_index():
return render_template("status_index.html")
def _status_dict_to_html(response):
items = "<br>\n".join(f"{k}: {v}" for k, v in sorted(response.items()))
return """<html>
<body>
{}
</body>
</html>""".format(
items,
)
def _status_render(response, fmt):
if fmt == "json":
return jsonify(response)
return _status_dict_to_html(response)
@app.route("/api/status/<site>/<inst>.<fmt>", methods=["GET"])
@app.route("/api/status/<site>/<inst>", methods=["GET"])
@app.route("/api/status/<site>.<fmt>", methods=["GET"])
@app.route("/api/status/<site>", methods=["GET"])
def get_instrument_status(site, inst=None, fmt=None):
"""See `/api/status/` for more information."""
# defaults:
response = {
"name": site if inst is None else inst,
"short_name": "",
"long_name": "",
"status_code": 255,
"status_message": "",
}
mod_time = datetime.utcnow()
if fmt is None:
fmt = "html"
if fmt not in ["html", "json"]:
return render_template("400.html", format=fmt), 400
site_path = Path(site)
json_subpath = site_path / "status.json" if inst is None else site_path / inst / "status.json"
# try to load the JSON file from the archive
archive_root = app.config.get("ARCHIVE_ROOT")
if archive_root.startswith("http"):
LOG.warning("Using URL request for status JSON, not meant for operational use")
# we aren't on a system with the archive available, fall back to URL
# loads directly to the archive
base_url = app.config.get("ARCHIVE_URL")
json_url = f"{base_url}/{json_subpath}"
try:
# Security: We check to ensure this is an HTTP URL as a base URL.
# The server configuration is also the one setting what the root URL is.
with urlopen(json_url) as json_file:
json_str = json_file.read()
except URLError:
response["status_message"] = f"Could not retrieve configured status: {json_url}"
json_str = None
else:
base_path = Path(archive_root)
json_path = base_path / json_subpath
try:
with json_path.open("r") as json_file:
json_str = json_file.read()
mod_time = datetime.fromtimestamp(os.path.getmtime(json_path))
except FileNotFoundError:
response["status_message"] = "No status information found."
json_str = None
if json_str is None:
# exit out early, we don't know what this is
return _status_render(response, fmt)
json_dict = builtin_json.loads(json_str)
response["last_updated"] = mod_time.strftime("%Y-%m-%d %H:%M:%SZ")
response.update(json_dict)
return _status_render(response, fmt)
if __name__ == "__main__":
app.debug = True
bind_addr = "0.0.0.0" if len(sys.argv) <= 1 else sys.argv[0] # noqa: S104
app.run(bind_addr, threaded=True)