app.py
"""API routes and request resolvers for a Bottle app."""
from typing import Dict, Any
import os
import sys
from datetime import date
from kedro.framework.session import KedroSession
from bottle import Bottle, run, request, response
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__)))
SRC_PATH = os.path.join(BASE_DIR, "src")
if SRC_PATH not in sys.path:
sys.path.append(SRC_PATH)
from augury import api
from augury import settings
IS_PRODUCTION = settings.ENV == "production"
TIPRESIAS_HOST = (
"http://www.tipresias.net" if IS_PRODUCTION else "http://host.docker.internal:8000"
)
PACKAGE_NAME = "augury"
app = Bottle()
if IS_PRODUCTION:
from rollbar.contrib.bottle import RollbarBottleReporter
rbr = RollbarBottleReporter(
access_token=settings.ROLLBAR_TOKEN,
environment=settings.ENV,
)
app.install(rbr)
def _run_kwargs():
run_kwargs: Dict[str, Any] = {
"port": int(os.getenv("PORT", "8008")),
"reloader": not IS_PRODUCTION,
"host": "0.0.0.0",
"server": "gunicorn",
"accesslog": "-",
"timeout": 1200,
"workers": 1,
}
return run_kwargs
def _unauthorized_response():
response.status = 401
return "Unauthorized"
def _request_is_authorized(http_request) -> bool:
auth_token = http_request.headers.get("Authorization")
if (
IS_PRODUCTION
and auth_token != f"Bearer {os.environ['DATA_SCIENCE_SERVICE_TOKEN']}"
):
return False
return True
@app.route("/predictions")
def predictions():
"""
Generate predictions for the given year and round number.
Params
------
Request with the following URL params:
year_range (str, optional): Year range for which you want prediction data.
Format = yyyy-yyyy.
Default = current year only.
round_number (int, optional): Round number for which you want prediction data.
Default = All rounds for given year.
ml_models (str, optional): Comma-separated list of names of ML model to use
for making predictions.
Default = All available models.
train_models (bool, optional): Whether to train each model
on earlier seasons' data before generating predictions
for a given season/round.
Default = False.
Returns
-------
Response with a body that has a JSON of prediction data.
"""
if not _request_is_authorized(request):
return _unauthorized_response()
this_year = date.today().year
year_range_param = (
f"{this_year}-{this_year + 1}"
if request.query.year_range in [None, ""]
else request.query.year_range
)
year_range = tuple([int(year) for year in year_range_param.split("-")])
round_number = request.query.round_number
round_number = None if round_number in [None, ""] else int(round_number)
ml_models_param = request.query.ml_models
ml_models_param = (
None if ml_models_param in [None, ""] else ml_models_param.split(",")
)
train_models_param = request.query.train_models
train_models = train_models_param.lower() == "true"
with KedroSession.create(
settings.PACKAGE_NAME,
env=settings.ENV,
project_path=settings.BASE_DIR,
extra_params={"round_number": round_number},
):
return api.make_predictions(
year_range,
round_number=round_number,
ml_model_names=ml_models_param,
train=train_models,
)
@app.route("/fixtures")
def fixtures():
"""
Fetch fixture data for the given date range.
Params
------
Request with the following URL params:
start_date (string of form 'yyyy-mm-dd', required): Start of date range
(inclusive) for which you want data.
end_date (string of form 'yyyy-mm-dd', required): End of date range
(inclusive) for which you want data.
Returns
-------
Response with a body that has a JSON of fixture data.
"""
if not _request_is_authorized(request):
return _unauthorized_response()
start_date = request.query.start_date
end_date = request.query.end_date
with KedroSession.create(
settings.PACKAGE_NAME, env=settings.ENV, project_path=settings.BASE_DIR
):
return api.fetch_fixture_data(start_date, end_date)
@app.route("/match_results")
def match_results():
"""
Fetch match results data for the given round.
Params
------
Request with the following URL params:
round_number (int): Fetch data for the given round. If missing, will fetch
all match results for the current year.
Returns
-------
Response with a body that has a JSON of match results data.
"""
if not _request_is_authorized(request):
return _unauthorized_response()
round_number = request.query.round_number
with KedroSession.create(
settings.PACKAGE_NAME, env=settings.ENV, project_path=settings.BASE_DIR
):
return api.fetch_match_results_data(round_number)
@app.route("/matches")
def matches():
"""
Fetch match data for the given date range.
Params
------
Request with the following URL params:
start_date (string of form 'yyyy-mm-dd', required): Start of date range
(inclusive) for which you want data.
end_date (string of form 'yyyy-mm-dd', required): End of date range
(inclusive) for which you want data.
Returns
-------
Response with a body that has a JSON of match data.
"""
if not _request_is_authorized(request):
return _unauthorized_response()
start_date = request.query.start_date
end_date = request.query.end_date
with KedroSession.create(
settings.PACKAGE_NAME, env=settings.ENV, project_path=settings.BASE_DIR
):
return api.fetch_match_data(start_date, end_date)
@app.route("/ml_models")
def ml_models():
"""
Fetch info for all available ML models.
Returns
-------
Response with a body that has a JSON of ML model data.
"""
if not _request_is_authorized(request):
return _unauthorized_response()
with KedroSession.create(
settings.PACKAGE_NAME, env=settings.ENV, project_path=settings.BASE_DIR
):
return api.fetch_ml_model_info()
run(app, **_run_kwargs())