Galaxy Brain

frameworks

Galaxy Web Frameworks

Learning Questions

Learning Objectives

Client-Server Communications

Bits and pieces of older client technologies appear throughout - ranging from Python mako templates to generate HTML, lower-level jQuery, Axios interactions, and Backbone legacy MVC.

Processing requests on the server

Expanding the right side of that diagram. We will move through the components left to right.

ASGI - Application

Spiritual successor to WSGI. An ASGI application is an async callable that takes in a scope (dict describing the connection), send (an async callable to respond to events via), and receive (an async callable to receive messages).

async def application(scope, receive, send):
    event = await receive()
    ...
    await send({"type": "http.response.body", ...})

Checkout ASGI documentation for more details.

ASGI & Starlette Low-level Example

We will talk a lot about Galaxy and FastAPI - but much of its plumbing is just aliases for starlette ASGI handling.

from starlette.responses import PlainTextResponse

async def app(scope, receive, send):
    assert scope['type'] == "http"
    response = PlainTextResponse('Hello, world!')
    await response(scope, receive, send)

If this is placed in a file called example.py with Starlette on the Python path, the application server uvicorn can then host this application with the following shell command:

$ uvicorn example:app
INFO: Started server process [11509]
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Example from starlette.io.

ASGI - Starlette High-level Example

Building a higher-level example.py with Starlette.

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route


async def homepage(request):
    return JSONResponse({'hello': 'world'})


app = Starlette(debug=True, routes=[
    Route('/', homepage),
])

A small framework for building web applications.

ASGI - FastAPI

From https://github.com/tiangolo/fastapi/blob/master/fastapi/applications.py

...
from starlette.applications import Starlette
...

class FastAPI(Starlette):
    ...

FastAPI (the library and the application base) extends starlette framework with features for building APIs. These include data validation, serialization, documentation generation.

Processing requests on the server

FastAPI __call__

https://github.com/tiangolo/fastapi/blob/master/fastapi/applications.py

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    if self.root_path:
        scope["root_path"] = self.root_path
    if AsyncExitStack:
        async with AsyncExitStack() as stack:
             scope["fastapi_astack"] = stack
             await super().__call__(scope, receive, send)
    else:
        await super().__call__(scope, receive, send)  # pragma: no cover

A light wrapper around Starlette’s call entry point.

Starlette __call__

https://github.com/encode/starlette/blob/master/starlette/applications.py

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
   scope["app"] = self
   await self.middleware_stack(scope, receive, send)

Walks through Starlette middleware.

Starlette build_middleware_stack

https://github.com/encode/starlette/blob/master/starlette/applications.py

def build_middleware_stack(self) -> ASGIApp:
    ...

    app = self.router
    for cls, options in reversed(middleware):
        app = cls(app=app, **options)
    return app

Start with the router and surround it with each layer of configured middleware.

ASGI Middleware

It is possible to have ASGI “middleware” - code that plays the role of both server and application, taking in a scope and the send/receive awaitable callables, potentially modifying them, and then calling an inner application.

https://asgi.readthedocs.io/en/latest/specs/main.html#middleware

Starlette Middleware

Starlette includes several middleware classes for adding behavior that is applied across your entire application. These are all implemented as standard ASGI middleware classes, and can be applied either to Starlette or to any other ASGI application.

https://www.starlette.io/middleware/

Starlette build_middleware_stack

https://github.com/encode/starlette/blob/master/starlette/applications.py

def build_middleware_stack(self) -> ASGIApp:
    ...

    app = self.router
    for cls, options in reversed(middleware):
        app = cls(app=app, **options)
    return app

Notice the inner most layer is the router.

FastAPI Router Initialization

https://github.com/tiangolo/fastapi/blob/master/fastapi/applications.py

class FastAPI(Starlette):
    def __init__(self, ...):
        self.router: routing.APIRouter = routing.APIRouter(
            routes=routes,
        )

FastAPI Router

from starlette import routing

class APIRouter(routing.Router):
   ...

Starlette Router

https://github.com/encode/starlette/blob/master/starlette/routing.py

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
   ...
   for route in self.routes:
       # Determine if any route matches the incoming scope,
       # and hand over to the matching route if found.
       match, child_scope = route.matches(scope)
       if match == Match.FULL:
           scope.update(child_scope)
           await route.handle(scope, receive, send)
           return
       ...

https://www.starlette.io/routing/

Router Class Diagram

In order to understand how the routing classes are setup within Galaxy, lets step back and look really quickly at how Galaxy’s FastAPI application (ASGI endpoint) is constructed.

FastAPI Factory

lib/galaxy/webapps/galaxy/fast_factory.py

def factory():
    props = WebappSetupProps(
        app_name='galaxy',
        default_section_name=DEFAULT_CONFIG_SECTION,
        env_config_file='GALAXY_CONFIG_FILE',
        env_config_section='GALAXY_CONFIG_SECTION',
        check_galaxy_root=True
    )
    config_provider = WebappConfigResolver(props)
    config = config_provider.resolve_config()
    gx_webapp, gx_app = app_pair(
        global_conf=config.global_conf,
        load_app_kwds=config.load_app_kwds,
        wsgi_preflight=config.wsgi_preflight
    )
    return initialize_fast_app(gx_webapp, gx_app)

FastAPI Application

lib/galaxy/webapps/galaxy/fast_app.py

def initialize_fast_app(gx_webapp, gx_app):
    app = FastAPI(
        title="Galaxy API",
        docs_url="/api/docs",
        openapi_tags=api_tags_metadata,
    )
    add_exception_handler(app)
    add_galaxy_middleware(app, gx_app)
    add_request_id_middleware(app)
    include_all_package_routers(app, 'galaxy.webapps.galaxy.api')
    wsgi_handler = WSGIMiddleware(gx_webapp)
    app.mount('/', wsgi_handler)
    return app

Finding API Routers

Following this line:

include_all_package_routers(app, 'galaxy.webapps.galaxy.api')

to the file

lib/galaxy/webapps/base/api.py

def include_all_package_routers(app: FastAPI, package_name: str):
    for _, module in walk_controller_modules(package_name):
        router = getattr(module, "router", None)
        if router:
            app.include_router(router)

Routing inside the Application

router = Router(tags=['tags'])


@router.cbv
class FastAPITags:
    manager: TagsManager = depends(TagsManager)

    @router.put(
        '/api/tags',
        summary="Apply a new set of tags to an item.",
        status_code=status.HTTP_204_NO_CONTENT,
    )
    def update(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        payload: ItemTagsPayload = Body(
            ...,  # Required
            title="Payload",
            description="Request body containing the item and the tags to be assigned.",
        ),
    ):
        """Replaces the tags associated with an item with the new ones specified in the payload.

        - The previous tags will be __deleted__.
        - If no tags are provided in the request body, the currently associated tags will also be __deleted__.
        """
        self.manager.update(trans, payload)

Router Class Diagram

WSGI Fallback

Back to initialize_fast_app, two of the final lines were as follows:

wsgi_handler = WSGIMiddleware(gx_webapp)
app.mount('/', wsgi_handler)

This effectively provides a fallback to our legacy WSGI application.

WSGI

Processing requests on the server

WSGI Middleware

A WSGI function:

def app(environ, start_response):

Galaxy’s WSGI Middleware

Middleware configured in galaxy.webapps.galaxy.buildapp#wrap_in_middleware.

Processing requests on the server (WSGI)

Instances

webapp

Classes

GalaxyWebApplication class diagram

Routes

Setup on webapp in galaxy.webapps.galaxy.buildapp.py.

webapp.add_route(
    '/datasets/:dataset_id/display/{filename:.+?}',
    controller='dataset', action='display',
    dataset_id=None, filename=None
)

URL /datasets/278043/display matches this route, so handle_request will

Uses popular Routes library (https://pypi.python.org/pypi/Routes).

Simplified handle_request from lib/galaxy/web/framework/base.py.

def handle_request(self, environ, start_response):
    path_info = environ.get( 'PATH_INFO', '' )
    map = self.mapper.match( path_info, environ )
    if path_info.startswith('/api'):
        controllers = self.api_controllers
    else:
        controllers = self.controllers

    trans = self.transaction_factory( environ )

    controller_name = map.pop( 'controller', None )
    controller = controllers.get( controller_name, None )

    # Resolve action method on controller
    action = map.pop( 'action', 'index' )
    method = getattr( controller, action, None )

    kwargs = trans.request.params.mixed()
    # Read controller arguments from mapper match
    kwargs.update( map )

    body = method( trans, **kwargs )
    # Body may be a file, string, etc... respond with it.

Controllers

Three varieties

  1. FastAPI ASGI API controllers
  2. WSGI API controllers
  3. Legacy WSGI web controllers.

Ideally each of these are thin. Focused on “web things” - adapting parameters and responses and move “business logic” to components not bound to web functionality.

FastAPI Controllers

FastAPI Controller Example

lib/galaxy/webapps/galaxy/controllers/api/roles.py

@router.cbv
class FastAPIRoles:
    role_manager: RoleManager = depends(RoleManager)

    @router.get('/api/roles')
    def index(self, trans: ProvidesUserContext = DependsOnTrans) -> RoleListModel:
        roles = self.role_manager.list_displayable_roles(trans)
        return RoleListModel(__root__=[role_to_model(trans, r) for r in roles])

    @router.get('/api/roles/{id}')
    def show(self, id: EncodedDatabaseIdField, trans: ProvidesUserContext = DependsOnTrans) -> RoleModel:
        role_id = trans.app.security.decode_id(id)
        role = self.role_manager.get(trans, role_id)
        return role_to_model(trans, role)

    @router.post("/api/roles", require_admin=True)
    def create(self, trans: ProvidesUserContext = DependsOnTrans, role_definition_model: RoleDefinitionModel = Body(...)) -> RoleModel:
        role = self.role_manager.create_role(trans, role_definition_model)
        return role_to_model(trans, role)

Pitfall: async def Doing Blocking I/O

A common mistake — a helper declared async that only does blocking work:

async def list_history_items(session: Session, history_id: int) -> str:
    hda_rows = session.execute(
        select(HDA.id, HDA.hid, HDA.name)
        .join(Dataset, HDA.dataset_id == Dataset.id)
        .where(HDA.history_id == history_id)
    ).all()
    ...

Declared async, but session is a synchronous SQLAlchemy Session — session.execute(...) is a blocking call.

Event Loop Blocking vs Threadpool

Why This Breaks Galaxy

Convention: Default to Sync def

-async def list_history_items(session: Session, history_id: int) -> str:
+def list_history_items(session: Session, history_id: int) -> str:
     rows = session.execute(select(...)).all()
rows = await anyio.to_thread.run_sync(partial(list_history_items, session, history_id))

The aiocop Guard

Opt-in via GALAXY_TEST_AIOCOP=1 — sys.audit hooks catch blocking syscalls inside async tasks and tag the response with X-Aiocop-Violations; the test interactor fails any high-severity hit.

lib/galaxy/web/framework/middleware/aiocop_integration.py · test/integration/test_event_loop_blocking.py

https://github.com/galaxyproject/galaxy/pull/22207

Galaxy’s aiocop integration (lib/galaxy/web/framework/middleware/aiocop_integration.py) installs sys.audit hooks that catch specific blocking syscalls — socket.connect, open, subprocess.Popen, and similar — when they run inside an async task, and records the offending call site. Galaxy wraps this in an ASGI middleware that attaches any per-request violations to an X-Aiocop-Violations response header (count=…;severity=…;first=…); the API test interactor (galaxy_test.base.api._check_aiocop_violations) fails any request whose maximum severity reaches aiocop’s high threshold. It is a test-only dependency, opt-in via GALAXY_TEST_AIOCOP, and is not imported by production servers.

The limitation that matters: aiocop is a runtime audit hook, not a static check. It only observes a blocking call on a code path that is actually executed while aiocop is active. An async def that does synchronous I/O but is never exercised by such a test slips straight through; so does one whose blocking call stays below the severity threshold. The guard is a safety net for covered paths, not a substitute for getting the declaration right — treat sync-vs-async correctness as a code-review responsibility and give new async endpoints integration coverage that runs them.

FastAPI and Pydantic

RoleIdField = Field(title="ID", description="Encoded ID of the role")
RoleNameField = Field(title="Name", description="Name of the role")
RoleDescriptionField = Field(title="Description", description="Description of the role")


class BasicRoleModel(BaseModel):
    id: EncodedDatabaseIdField = RoleIdField
    name: str = RoleNameField
    type: str = Field(title="Type", description="Type or category of the role")


class RoleModel(BasicRoleModel):
    description: str = RoleDescriptionField
    url: str = Field(title="URL", description="URL for the role")
    model_class: str = Field(title="Model class", description="Database model class (Role)")


class RoleDefinitionModel(BaseModel):
    name: str = RoleNameField
    description: str = RoleDescriptionField
    user_ids: Optional[List[EncodedDatabaseIdField]] = Field(title="User IDs", default=[])
    group_ids: Optional[List[EncodedDatabaseIdField]] = Field(title="Group IDs", default=[])

FastAPI and OpenAPI

FastAPI(title="Galaxy API", docs_url="/api/docs", ...)

OpenAPI Docs from FastAPI at api/docs

OpenAPI Docs from FastAPI at api/docs for roles

WSGI API Controllers

WSGI API Controller Example

lib/galaxy/webapps/galaxy/controllers/api/roles.py

class RoleAPIController(BaseGalaxyAPIController):
    role_manager: RoleManager = depends(RoleManager)

    @web.expose_api
    def index(self, trans: ProvidesUserContext, **kwd):
        """
        GET /api/roles
        Displays a collection (list) of roles.
        """
        roles = self.role_manager.list_displayable_roles(trans)
        return RoleListModel(__root__=[role_to_model(trans, r) for r in roles])

    @web.expose_api
    def show(self, trans: ProvidesUserContext, id: str, **kwd):
        """
        GET /api/roles/{encoded_role_id}
        Displays information about a role.
        """
        role_id = decode_id(self.app, id)
        role = self.role_manager.get(trans, role_id)
        return role_to_model(trans, role)

    @web.expose_api
    @web.require_admin
    def create(self, trans: ProvidesUserContext, payload, **kwd):
        """
        POST /api/roles
        Creates a new role.
        """
        expand_json_keys(payload, ["user_ids", "group_ids"])
        role_definition_model = RoleDefinitionModel(**payload)
        role = self.role_manager.create_role(trans, role_definition_model)
        return role_to_model(trans, role)

Legacy WSGI Controllers

Key Takeaways