Structure FastAPI dependencies

For now dump them into a single module, but that could become a package
of course.

lib/galaxy/managers/session.py

@@ -0,0 +1,23 @@
+from sqlalchemy import (
+    and_,
+    true,
+)
+from sqlalchemy.orm import joinedload
+
+from galaxy import model
+
+
+class GalaxySessionManager:
+    """Manages GalaxySession."""
+
+    def __init__(self, sa_session):
+        self.session = sa_session
+
+    def get_session_from_session_key(self, session_key: str):
+        """Returns GalaxySession if session_key is valid."""
+        galaxy_session = self.session.query(model.GalaxySession).filter(
+            and_(
+                model.GalaxySession.table.c.session_key == session_key,
+                model.GalaxySession.table.c.is_valid == true())
+        ).options(joinedload("user")).first()
+        return galaxy_session

lib/galaxy/managers/users.py

@@ -9,6 +9,7 @@
 
 from markupsafe import escape
 from sqlalchemy import and_, desc, exc, func, true
+from sqlalchemy.orm.exc import NoResultFound
 
 from galaxy import (
     exceptions,
@@ -230,6 +231,22 @@ def by_email_like(self, email_with_wildcards, filters=None, order_by=None, **kwa
         order_by = order_by or (model.User.email, )
         return super().list(filters=filters, order_by=order_by, **kwargs)
 
+    def by_api_key(self, api_key, sa_session=None):
+        """
+        Find a user by API key.
+        """
+        sa_session = sa_session or self.app.model.session
+        try:
+            provided_key = sa_session.query(self.app.model.APIKeys).filter(self.app.model.APIKeys.table.c.key == api_key).one()
+        except NoResultFound:
+            raise exceptions.AuthenticationFailed('Provided API key is not valid.')
+        if provided_key.user.deleted:
+            raise exceptions.AuthenticationFailed('User account is deactivated, please contact an administrator.')
+        newest_key = provided_key.user.api_keys[0]
+        if newest_key.key != provided_key.key:
+            raise exceptions.AuthenticationFailed('Provided API key has expired.')
+        return provided_key.user
+
     # ---- admin
     def is_admin(self, user, trans=None):
         """Return True if this user is an admin (or session is authenticated as admin).

lib/galaxy/webapps/galaxy/api/dependencies.py

@@ -0,0 +1,83 @@
+"""
+This module *does not* contain API routes. It exclusively contains dependencies to be used in FastAPI routes
+"""
+from functools import lru_cache
+from typing import (
+    Optional,
+)
+
+from fastapi import (
+    Cookie,
+    Depends,
+    Header,
+    HTTPException,
+    Query,
+)
+from sqlalchemy.orm import Session
+
+from galaxy import (
+    app as galaxy_app,
+    exceptions,
+    model,
+)
+from galaxy.app import UniverseApplication
+from galaxy.managers.session import GalaxySessionManager
+from galaxy.managers.users import UserManager
+from galaxy.model import User
+from galaxy.work.context import SessionRequestContext
+
+
+@lru_cache()
+def get_app() -> UniverseApplication:
+    return galaxy_app.app
+
+
+@lru_cache()
+def get_db(app: UniverseApplication = Depends(get_app)) -> Session:
+    # TODO: return sqlachemy 2.0 style session without autocommit and expire_on_commit!
+    return app.model.session
+
+
+@lru_cache()
+def get_user_manager(app: UniverseApplication = Depends(get_app)):
+    return UserManager(app)
+
+
+@lru_cache()
+def get_session_manager(db=Depends(get_db)) -> GalaxySessionManager:
+    return GalaxySessionManager(db)
+
+
+@lru_cache()
+def get_session(session_manager: GalaxySessionManager = Depends(get_session_manager),
+                app: UniverseApplication = Depends(get_app),
+                galaxysession: Optional[str] = Cookie(None)) -> Optional[model.GalaxySession]:
+    session_key = app.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?
+
+
+@lru_cache()
+def get_api_user(user_manager: UserManager = Depends(get_user_manager), key: Optional[str] = Query(None), x_api_key: Optional[str] = Header(None)) -> Optional[User]:
+    api_key = key or x_api_key
+    if not api_key:
+        return None
+    try:
+        return user_manager.by_api_key(api_key=api_key)
+    except exceptions.AuthenticationFailed as e:
+        raise HTTPException(status_code=e.status_code, detail=str(e))
+
+
+@lru_cache()
+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
+
+
+@lru_cache()
+def get_trans(app=Depends(get_app), user: Optional[User] = Depends(get_user),
+              galaxy_session: Optional[model.GalaxySession] = Depends(get_session),
+              ) -> SessionRequestContext:
+    return SessionRequestContext(app=app, user=user, galaxy_session=galaxy_session)

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -9,24 +9,20 @@
 from functools import lru_cache
 
 from fastapi import (
-    Cookie,
     Depends,
 )
 from fastapi_utils.cbv import cbv
 from fastapi_utils.inferring_router import InferringRouter as APIRouter
 from sqlalchemy import (
-    and_,
     or_,
-    true,
 )
-from sqlalchemy.orm import joinedload
 
 from galaxy import (
-    app as galaxy_app,
     exceptions,
     model,
     util,
 )
+from galaxy.app import UniverseApplication
 from galaxy.managers import hdas
 from galaxy.managers.jobs import (
     JobManager,
@@ -48,52 +44,29 @@
     SessionRequestContext,
     WorkRequestContext,
 )
+from .dependencies import (
+    get_app,
+    get_trans,
+)
 
 log = logging.getLogger(__name__)
 
 router = APIRouter()
 
 
 @lru_cache()
-def get_session(galaxysession: typing.Optional[str] = Cookie(None)) -> model.GalaxySession:
-    app = galaxy_app.app
-    session_key = app.security.decode_guid(galaxysession)
-    sa_session = app.model.session
-    if session_key:
-        # Retrieve the galaxy_session id via the unique session_key
-        galaxy_session = sa_session.query(app.model.GalaxySession).filter(
-            and_(
-                app.model.GalaxySession.table.c.session_key == session_key,
-                app.model.GalaxySession.table.c.is_valid == true())
-        ).options(joinedload("user")).first()
-        return galaxy_session
-
-
-@lru_cache()
-def get_user(galaxy_session: typing.Optional[model.GalaxySession] = Depends(get_session)):
-    return galaxy_session.user
-
-
-@lru_cache()
-def get_job_manager() -> JobManager:
-    return JobManager(galaxy_app.app)
-
-
-@lru_cache()
-def get_job_search() -> JobSearch:
-    return JobSearch(galaxy_app.app)
+def get_job_manager(app: UniverseApplication = Depends(get_app)) -> JobManager:
+    return JobManager(app=app)
 
 
 @lru_cache()
-def get_hda_manager() -> hdas.HDAManager:
-    return hdas.HDAManager(galaxy_app.app)
+def get_job_search(app: UniverseApplication = Depends(get_app)) -> JobSearch:
+    return JobSearch(app=app)
 
 
 @lru_cache()
-def get_trans(user: typing.Optional[model.User] = Depends(get_user),
-              galaxy_session: typing.Optional[model.GalaxySession] = Depends(get_session),
-              ) -> SessionRequestContext:
-    return SessionRequestContext(app=galaxy_app.app, user=user, galaxy_session=galaxy_session)
+def get_hda_manager(app: UniverseApplication = Depends(get_app)) -> hdas.HDAManager:
+    return hdas.HDAManager(app=app)
 
 
 @cbv(router)