Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

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 .buildapp 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)) app.mount("/", wsgi_handler) 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", )