import logging
from sqlalchemy import (
and_,
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
galaxy_session = (
self.sa_session.query(self.model.GalaxySession)
.filter(
and_(
self.model.GalaxySession.table.c.session_key == session_key,
self.model.GalaxySession.table.c.is_valid == true(),
)
)
.options(joinedload(self.model.GalaxySession.user))
.first()
)
return galaxy_session