import logging
from sqlalchemy import (
select,
true,
)
from sqlalchemy.orm import joinedload
from galaxy.model.base import SharedModelMapping
log = logging.getLogger(__name__)
[docs]
class GalaxySessionManager:
"""Manages GalaxySession."""
[docs]
def __init__(self, model: SharedModelMapping):
self.model = model
self.sa_session = model.context
[docs]
def get_session_from_session_key(self, session_key: str):
"""Returns GalaxySession if session_key is valid."""
# going through self.model since this can be used by Galaxy or Toolshed despite
# type annotations
stmt = (
select(self.model.GalaxySession)
.where(self.model.GalaxySession.session_key == session_key)
.where(self.model.GalaxySession.is_valid == true())
.options(joinedload(self.model.GalaxySession.user))
.limit(1)
)
return self.sa_session.scalars(stmt).first()