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",
)