Source code for tool_shed.webapp.fast_app

import logging
import os
from pathlib import Path
from typing import (
    Any,
    cast,
    Dict,
    Optional,
)

from a2wsgi import WSGIMiddleware
from fastapi import (
    Depends,
    FastAPI,
)
from fastapi.responses import (
    HTMLResponse,
    RedirectResponse,
)
from fastapi.staticfiles import StaticFiles
from starlette_graphene3 import (
    GraphQLApp,
    make_graphiql_handler,
)

from galaxy.webapps.base.api import (
    add_exception_handler,
    add_request_id_middleware,
    include_all_package_routers,
)
from galaxy.webapps.openapi.utils import get_openapi
from tool_shed.structured_app import ToolShedApp
from tool_shed.webapp.api2 import (
    ensure_valid_session,
    get_trans,
)
from tool_shed.webapp.graphql.schema import schema

log = logging.getLogger(__name__)

api_tags_metadata = [
    {
        "name": "authenticate",
        "description": "Authentication-related endpoints.",
    },
    {
        "name": "categories",
        "description": "Category-related endpoints.",
    },
    {
        "name": "repositories",
        "description": "Repository-related endpoints.",
    },
    {
        "name": "users",
        "description": "User-related endpoints.",
    },
    {"name": "undocumented", "description": "API routes that have not yet been ported to FastAPI."},
]

# Set this if asset handling should be sent to vite.
# Run vite with:
#   yarn dev
# Start tool shed with:
#   TOOL_SHED_VITE_PORT=4040 TOOL_SHED_API_VERSION=v2 ./run_tool_shed.sh
TOOL_SHED_VITE_PORT: Optional[str] = os.environ.get("TOOL_SHED_VITE_PORT", None)
TOOL_SHED_FRONTEND_TARGET: str = os.environ.get("TOOL_SHED_FRONTEND_TARGET") or "auto"  # auto, src, or node
TOOL_SHED_USE_HMR: bool = TOOL_SHED_VITE_PORT is not None
WEBAPP_DIR = Path(__file__).parent.resolve()
FRONTEND = WEBAPP_DIR / "frontend"
FRONTEND_DIST = FRONTEND / "dist"
INSTALLED_FRONTEND = WEBAPP_DIR / "node_modules" / "@galaxyproject" / "tool-shed-frontend" / "dist"
INDEX_FILENAME = "index.html"


def find_frontend_target() -> Path:
    src_target = FRONTEND_DIST
    node_target = INSTALLED_FRONTEND
    if TOOL_SHED_FRONTEND_TARGET == "src":
        return src_target
    elif TOOL_SHED_FRONTEND_TARGET == "node":
        return node_target
    elif src_target.exists():
        return src_target
    else:
        return node_target


def frontend_controller(app):
    shed_entry_point = "main.ts"
    vite_runtime = "@vite/client"

    def index(trans=Depends(get_trans)):
        if TOOL_SHED_USE_HMR:
            if TOOL_SHED_FRONTEND_TARGET != "auto":
                raise Exception("Cannot configure HMR and with this frontend target.")
            index = FRONTEND / INDEX_FILENAME
            index_html = index.read_text()
            index_html = index_html.replace(
                f"""<script type="module" src="/src/{shed_entry_point}"></script>""",
                f"""<script type="module" src="http://localhost:{TOOL_SHED_VITE_PORT}/{vite_runtime}"></script><script type="module" src="http://localhost:{TOOL_SHED_VITE_PORT}/src/{shed_entry_point}"></script>""",
            )
        else:
            index = find_frontend_target() / INDEX_FILENAME
            index_html = index.read_text()
        ensure_valid_session(trans)
        cookie = trans.session_csrf_token
        r: HTMLResponse = cast(HTMLResponse, trans.response)
        r.set_cookie("session_csrf_token", cookie)
        return index_html

    return app, index


def redirect_route(app, from_url: str, to_url: str):
    @app.get(from_url)
    def redirect():
        return RedirectResponse(to_url)


def frontend_route(controller, path):
    app, index = controller
    app.get(path, response_class=HTMLResponse)(index)


def mount_graphql(app: FastAPI, tool_shed_app: ToolShedApp):
    context = {
        "session": tool_shed_app.model.context,
        "security": tool_shed_app.security,
    }
    g_app = GraphQLApp(schema, on_get=make_graphiql_handler(), context_value=context, root_value=context)
    app.mount("/graphql", g_app)
    app.mount("/api/graphql", g_app)


FRONT_END_ROUTES = [
    "/",
    "/admin",
    "/login",
    "/register",
    "/logout_success",
    "/login_success",
    "/registration_success",
    "/help",
    "/repositories_by_search",
    "/repositories_by_category",
    "/repositories_by_category/{category_id}",
    "/repositories_by_owner",
    "/repositories_by_owner/{username}",
    "/repositories/{repository_id}",
    "/repositories_search",
    "/_component_showcase",
    "/user/api_key",
    "/user/change_password",
    "/view/{username}",
    "/view/{username}/{repository_name}",
    "/view/{username}/{repository_name}/{changeset_revision}",
]
LEGACY_ROUTES = {
    "/user/create": "/register",  # for twilltestcase
    "/user/login": "/login",  # for twilltestcase
}


[docs]def initialize_fast_app(gx_webapp, tool_shed_app):
    app = get_fastapi_instance()
    add_exception_handler(app)
    add_request_id_middleware(app)
    from .config import SHED_API_VERSION

    def mount_static(directory: Path):
        name = directory.name
        if directory.exists():
            app.mount(f"/{name}", StaticFiles(directory=directory), name=name)

    if SHED_API_VERSION == "v2":
        controller = frontend_controller(app)
        for route in FRONT_END_ROUTES:
            frontend_route(controller, route)

        for from_route, to_route in LEGACY_ROUTES.items():
            redirect_route(app, from_route, to_route)

        mount_graphql(app, tool_shed_app)

        mount_static(FRONTEND / "static")
        if TOOL_SHED_USE_HMR:
            mount_static(FRONTEND / "node_modules")
        else:
            mount_static(find_frontend_target() / "assets")

    routes_package = "tool_shed.webapp.api" if SHED_API_VERSION == "v1" else "tool_shed.webapp.api2"
    include_all_package_routers(app, routes_package)
    wsgi_handler = WSGIMiddleware(gx_webapp)
    tool_shed_app.haltables.append(("WSGI Middleware threadpool", wsgi_handler.executor.shutdown))
    # https://github.com/abersheeran/a2wsgi/issues/44
    app.mount("/", wsgi_handler)  # type: ignore[arg-type]
    return app


def get_fastapi_instance() -> FastAPI:
    return FastAPI(
        title="Galaxy Tool Shed API",
        description=("This API allows you to manage the Tool Shed repositories."),
        docs_url="/api/docs",
        redoc_url="/api/redoc",
        tags=api_tags_metadata,
        license_info={"name": "MIT", "url": "https://github.com/galaxyproject/galaxy/blob/dev/LICENSE.txt"},
    )


[docs]def get_openapi_schema() -> Dict[str, Any]:
    """
    Dumps openAPI schema without starting a full app and webserver.
    """
    app = get_fastapi_instance()
    include_all_package_routers(app, "tool_shed.webapp.api2")
    return get_openapi(
        title=app.title,
        version=app.version,
        openapi_version="3.1.0",
        description=app.description,
        routes=app.routes,
        license_info=app.license_info,
    )


__all__ = (
    "add_request_id_middleware",
    "get_openapi_schema",
    "initialize_fast_app",
)