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 galaxy.webapps.galaxy.api

This module *does not* contain API routes. It exclusively contains dependencies to be used in FastAPI routes
import inspect
from typing import (
from urllib.parse import urlencode

from fastapi import (
from fastapi.exceptions import RequestValidationError
from fastapi.params import Depends
from fastapi_utils.cbv import cbv
from fastapi_utils.inferring_router import InferringRouter
from pydantic import ValidationError
from pydantic.main import BaseModel
from starlette.routing import NoMatchFound
    from starlette_context import context as request_context
except ImportError:
    request_context = None  # type: ignore[assignment]

from galaxy import (
    app as galaxy_app,
from galaxy.exceptions import (
from galaxy.managers.session import GalaxySessionManager
from galaxy.managers.users import UserManager
from galaxy.model import User
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.structured_app import StructuredApp
from galaxy.web.framework.decorators import require_admin_message
from galaxy.webapps.base.controller import BaseAPIController
from galaxy.work.context import (

[docs]def get_app() -> StructuredApp: return cast(StructuredApp, galaxy_app.app)
[docs]async def get_app_with_request_session() -> AsyncGenerator[StructuredApp, None]: app = get_app() request_id = request_context.data['X-Request-ID'] app.model.set_request_id(request_id) try: yield app finally: app.model.unset_request_id(request_id)
DependsOnApp = Depends(get_app_with_request_session) T = TypeVar("T")
[docs]class GalaxyTypeDepends(Depends): """Variant of fastapi Depends that can also work on WSGI Galaxy controllers."""
[docs] def __init__(self, callable, dep_type): super().__init__(callable) self.galaxy_type_depends = dep_type
[docs]def depends(dep_type: Type[T]) -> T: def _do_resolve(request: Request): return get_app().resolve(dep_type) return cast(T, GalaxyTypeDepends(_do_resolve, dep_type))
[docs]def get_session_manager(app: StructuredApp = DependsOnApp) -> GalaxySessionManager: # TODO: find out how to adapt dependency for Galaxy/Report/TS return GalaxySessionManager(app.model)
[docs]def get_session(session_manager: GalaxySessionManager = Depends(get_session_manager), security: IdEncodingHelper = depends(IdEncodingHelper), galaxysession: Optional[str] = Cookie(None)) -> Optional[model.GalaxySession]: if galaxysession: session_key = security.decode_guid(galaxysession) if session_key: return session_manager.get_session_from_session_key(session_key) # TODO: What should we do if there is no session? Since this is the API, maybe nothing is the right choice? return None
[docs]def get_api_user( security: IdEncodingHelper = depends(IdEncodingHelper), user_manager: UserManager = depends(UserManager), key: Optional[str] = Query(None), x_api_key: Optional[str] = Header(None), run_as: Optional[EncodedDatabaseIdField] = Header( default=None, title='Run as User', description=( 'The user ID that will be used to effectively make this API call. ' 'Only admins and designated users can make API calls on behalf of other users.' ) ) ) -> Optional[User]: api_key = key or x_api_key if not api_key: return None user = user_manager.by_api_key(api_key=api_key) if run_as: if user_manager.user_can_do_run_as(user): try: decoded_run_as_id = security.decode_id(run_as) except Exception: raise UserInvalidRunAsException return user_manager.by_id(decoded_run_as_id) else: raise UserCannotRunAsException return user
[docs]def get_user(galaxy_session: Optional[model.GalaxySession] = Depends(get_session), api_user: Optional[User] = Depends(get_api_user)) -> Optional[User]: if galaxy_session: return galaxy_session.user return api_user
[docs]class UrlBuilder:
[docs] def __init__(self, request: Request): self.request = request
def __call__(self, name: str, **path_params): qualified = path_params.pop("qualified", False) # starlette does not support query parameters in url_path_for: https://github.com/encode/starlette/issues/560 query_params = path_params.pop('query_params', None) try: if qualified: url = self.request.url_for(name, **path_params) else: url = self.request.app.url_path_for(name, **path_params) if query_params: url = f"{url}?{urlencode(query_params)}" return url except NoMatchFound: # Fallback to legacy url_for if query_params: path_params.update(query_params) return web.url_for(name, **path_params)
[docs]class GalaxyASGIRequest(GalaxyAbstractRequest): """Wrapper around Starlette/FastAPI Request object. Implements the GalaxyAbstractRequest interface to provide access to some properties of the request commonly used."""
[docs] def __init__(self, request: Request): self.__request = request
@property def base(self) -> str: return str(self.__request.base_url) @property def host(self) -> str: client = self.__request.client assert client is not None return str(client.host)
[docs]class GalaxyASGIResponse(GalaxyAbstractResponse): """Wrapper around Starlette/FastAPI Response object. Implements the GalaxyAbstractResponse interface to provide access to some properties of the response object commonly used."""
[docs] def __init__(self, response: Response): self.__response = response
@property def headers(self): return self.__response.headers
DependsOnUser = Depends(get_user)
[docs]def get_current_history_from_session(galaxy_session: Optional[model.GalaxySession]) -> Optional[model.History]: if galaxy_session: return galaxy_session.current_history return None
[docs]def get_trans(request: Request, response: Response, app: StructuredApp = DependsOnApp, user: Optional[User] = Depends(get_user), galaxy_session: Optional[model.GalaxySession] = Depends(get_session), ) -> SessionRequestContext: url_builder = UrlBuilder(request) galaxy_request = GalaxyASGIRequest(request) galaxy_response = GalaxyASGIResponse(response) return SessionRequestContext( app=app, user=user, galaxy_session=galaxy_session, url_builder=url_builder, request=galaxy_request, response=galaxy_response, history=get_current_history_from_session(galaxy_session), )
DependsOnTrans = Depends(get_trans)
[docs]def get_admin_user(trans: SessionRequestContext = DependsOnTrans): if not trans.user_is_admin: raise AdminRequiredException(require_admin_message(trans.app.config, trans.user)) return trans.user
AdminUserRequired = Depends(get_admin_user)
[docs]class BaseGalaxyAPIController(BaseAPIController):
[docs] def __init__(self, app: StructuredApp): super().__init__(app)
[docs]class Router(InferringRouter): """A FastAPI Inferring Router tailored to Galaxy. """
[docs] def get(self, *args, **kwd): """Extend FastAPI.get to accept a require_admin Galaxy flag.""" return super().get(*args, **self._handle_galaxy_kwd(kwd))
[docs] def put(self, *args, **kwd): """Extend FastAPI.put to accept a require_admin Galaxy flag.""" return super().put(*args, **self._handle_galaxy_kwd(kwd))
[docs] def post(self, *args, **kwd): """Extend FastAPI.post to accept a require_admin Galaxy flag.""" return super().post(*args, **self._handle_galaxy_kwd(kwd))
[docs] def delete(self, *args, **kwd): """Extend FastAPI.delete to accept a require_admin Galaxy flag.""" return super().delete(*args, **self._handle_galaxy_kwd(kwd))
def _handle_galaxy_kwd(self, kwd): require_admin = kwd.pop("require_admin", False) if require_admin: if "dependencies" in kwd: kwd["dependencies"].append(AdminUserRequired) else: kwd["dependencies"] = [AdminUserRequired] return kwd @property def cbv(self): """Short-hand for frequently used Galaxy-pattern of FastAPI class based views. Creates a class-based view for for this router, for more information see: https://fastapi-utils.davidmontague.xyz/user-guide/class-based-views/ """ return cbv(self)
[docs]def as_form(cls: Type[BaseModel]): """ Adds an as_form class method to decorated models. The as_form class method can be used with FastAPI endpoints. See https://github.com/tiangolo/fastapi/issues/2387#issuecomment-731662551 """ new_params = [ inspect.Parameter( field.alias, inspect.Parameter.POSITIONAL_ONLY, default=(Form(field.default) if not field.required else Form(...)), ) for field in cls.__fields__.values() ] async def _as_form(**data): try: return cls(**data) except ValidationError as e: raise RequestValidationError(e.raw_errors) sig = inspect.signature(_as_form) sig = sig.replace(parameters=new_params) _as_form.__signature__ = sig # type: ignore[attr-defined] cls.as_form = _as_form # type: ignore[attr-defined] return cls
[docs]async def try_get_request_body_as_json(request: Request) -> Optional[Any]: """Returns the request body as a JSON object if the content type is JSON.""" if "application/json" in request.headers.get("content-type", ""): body = await request.json() return body return None