majuna/app/portal/list.py

import json
from datetime import datetime
from typing import Optional

from flask import render_template, url_for, flash, redirect, Blueprint, Response
from flask.typing import ResponseReturnValue
from flask_wtf import FlaskForm
from sqlalchemy import exc
from wtforms import SelectField, StringField, SubmitField
from wtforms.validators import DataRequired

from app.extensions import db
from app.lists.bc2 import mirror_sites
from app.lists.bridgelines import bridgelines
from app.lists.mirror_mapping import mirror_mapping
from app.models.base import MirrorList
from app.portal.util import response_404, view_lifecycle

bp = Blueprint("list", __name__)


@bp.app_template_filter("provider_name")
def list_provider_name(key: str) -> str:
    return MirrorList.providers_supported.get(key, "Unknown")


@bp.app_template_filter("format_name")
def list_format_name(key: str) -> str:
    return MirrorList.formats_supported.get(key, "Unknown")


@bp.app_template_filter("list_encoding_name")
def list_encoding_name(key: str) -> str:
    return MirrorList.encodings_supported.get(key, "Unknown")


@bp.route('/list')
def list_list() -> ResponseReturnValue:
    lists = MirrorList.query.filter(MirrorList.destroyed.is_(None)).all()
    return render_template("list.html.j2",
                           section="list",
                           title="Mirror Lists",
                           item="mirror list",
                           new_link=url_for("portal.list.list_new"),
                           items=lists,
                           extra_buttons=[
                               {
                                   "link": url_for("portal.list.list_preview", format_=k),
                                   "text": f"Preview {v}",
                                   "style": "secondary"
                               }
                               for k, v in MirrorList.formats_supported.items()
                           ]
                           )


@bp.route('/preview/<format_>')
def list_preview(format_: str) -> ResponseReturnValue:
    if format_ == "bca":
        return Response(json.dumps(mirror_mapping()), content_type="application/json")
    if format_ == "bc2":
        return Response(json.dumps(mirror_sites()), content_type="application/json")
    if format_ == "bridgelines":
        return Response(json.dumps(bridgelines()), content_type="application/json")
    return response_404(message="Format not found")


@bp.route("/destroy/<list_id>", methods=['GET', 'POST'])
def list_destroy(list_id: int) -> ResponseReturnValue:
    list_ = MirrorList.query.filter(MirrorList.id == list_id, MirrorList.destroyed.is_(None)).first()
    if list_ is None:
        return response_404("The requested bridge configuration could not be found.")
    return view_lifecycle(
        header="Destroy mirror list?",
        message=list_.description,
        success_view="portal.list.list_list",
        success_message="This list will no longer be updated and may be deleted depending on the provider.",
        section="list",
        resource=list_,
        action="destroy"
    )


@bp.route("/new", methods=['GET', 'POST'])
@bp.route("/new/<group_id>", methods=['GET', 'POST'])
def list_new(group_id: Optional[int] = None) -> ResponseReturnValue:
    form = NewMirrorListForm()
    form.provider.choices = list(MirrorList.providers_supported.items())
    form.format.choices = list(MirrorList.formats_supported.items())
    form.encoding.choices = list(MirrorList.encodings_supported.items())
    if form.validate_on_submit():
        list_ = MirrorList()
        list_.provider = form.provider.data
        list_.format = form.format.data
        list_.encoding = form.encoding.data
        list_.description = form.description.data
        list_.container = form.container.data
        list_.branch = form.branch.data
        list_.role = form.role.data
        list_.filename = form.filename.data
        list_.created = datetime.utcnow()
        list_.updated = datetime.utcnow()
        try:
            db.session.add(list_)
            db.session.commit()
            flash("Created new mirror list.", "success")
            return redirect(url_for("portal.list.list_list"))
        except exc.SQLAlchemyError:
            flash("Failed to create new mirror list.", "danger")
            return redirect(url_for("portal.list.list_list"))
    if group_id:
        form.group.data = group_id
    return render_template("new.html.j2", section="list", form=form)


class NewMirrorListForm(FlaskForm):  # type: ignore
    provider = SelectField('Provider', validators=[DataRequired()])
    format = SelectField('Distribution Method', validators=[DataRequired()])
    encoding = SelectField('Encoding', validators=[DataRequired()])
    description = StringField('Description', validators=[DataRequired()])
    container = StringField('Container', validators=[DataRequired()],
                            description="GitHub Project, GitLab Project or AWS S3 bucket name.")
    branch = StringField('Git Branch/AWS Region', validators=[DataRequired()],
                         description="For GitHub/GitLab, set this to the desired branch name, e.g. main. For AWS S3, "
                                     "set this field to the desired region, e.g. us-east-1.")
    role = StringField('Role ARN',
                       description="(Optional) ARN for IAM role to assume for interaction with the S3 bucket.")
    filename = StringField('Filename', validators=[DataRequired()])
    submit = SubmitField('Save Changes')