Warning

This document is for an in-development version 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 .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", )