galaxy/lib/tool_shed/context.py

import abc

from sqlalchemy.orm import scoped_session
from typing_extensions import Protocol

from galaxy.security.idencoding import IdEncodingHelper
from galaxy.work.context import (
    GalaxyAbstractRequest,
    GalaxyAbstractResponse,
)
from tool_shed.structured_app import ToolShedApp
from tool_shed.webapp.model import (
    GalaxySession,
    User,
)
from tool_shed.webapp.model.mapping import ToolShedModelMapping


class ProvidesAppContext(Protocol):
    """For transaction-like objects to provide the shed convenience layer for
    database and event handling.

    Mixed in class must provide `app` property.
    """

    @property
    @abc.abstractmethod
    def app(self) -> ToolShedApp:
        """Provide access to the shed ``app`` object."""

    @property
    def security(self) -> IdEncodingHelper:
        return self.app.security

    @property
    def sa_session(self) -> scoped_session:
        """Provide access to Galaxy's SQLAlchemy session.

        :rtype: galaxy.model.scoped_session.galaxy_scoped_session
        """
        return self.model.session

    @property
    def model(self) -> ToolShedModelMapping:
        """Provide access to Tool Shed's model mapping class."""
        return self.app.model


class ProvidesUserContext(ProvidesAppContext, Protocol):
    """For transaction-like objects to provide Galaxy convenience layer for
    reasoning about users.

    Mixed in class must provide `user` and `app`
    properties.
    """

    @property
    @abc.abstractmethod
    def user(self) -> User | None:
        """Provide access to the user object."""

    @property
    def anonymous(self) -> bool:
        return self.user is None

    @property
    def user_is_admin(self) -> bool:
        return self.app.config.is_admin_user(self.user)

    @property
    def user_is_bootstrap_admin(self) -> bool:
        """Master key provided so there is no real user"""
        user = self.user
        return not self.anonymous and user is not None and user.bootstrap_admin_user


class ProvidesRepositoriesContext(ProvidesUserContext, Protocol):
    @property
    @abc.abstractmethod
    def repositories_hostname(self) -> str:
        """Provide access to hostname used by target mercurial server."""


class SessionRequestContext(ProvidesRepositoriesContext, Protocol):
    @abc.abstractmethod
    def get_galaxy_session(self) -> GalaxySession | None: ...

    @abc.abstractmethod
    def set_galaxy_session(self, galaxy_session: GalaxySession): ...

    @property
    @abc.abstractmethod
    def request(self) -> GalaxyAbstractRequest: ...

    @property
    @abc.abstractmethod
    def response(self) -> GalaxyAbstractResponse: ...

    @abc.abstractmethod
    def url_builder(self): ...

    @property
    @abc.abstractmethod
    def session_csrf_token(self) -> str: ...


class SessionRequestContextImpl(SessionRequestContext):
    _app: ToolShedApp
    _user: User | None
    _galaxy_session: GalaxySession | None

    def __init__(
        self,
        app: ToolShedApp,
        request: GalaxyAbstractRequest,
        response: GalaxyAbstractResponse,
        user: User | None = None,
        galaxy_session: GalaxySession | None = None,
        url_builder=None,
    ):
        self._app = app
        self._user = user
        self._galaxy_session = galaxy_session
        self._url_builder = url_builder
        self.__request = request
        self.__response = response

    @property
    def app(self) -> ToolShedApp:
        return self._app

    @property
    def url_builder(self):
        return self._url_builder

    @property
    def user(self) -> User | None:
        return self._user

    def get_galaxy_session(self) -> GalaxySession | None:
        return self._galaxy_session

    def set_galaxy_session(self, galaxy_session: GalaxySession):
        self._galaxy_session = galaxy_session
        if galaxy_session.user:
            self._user = galaxy_session.user

    @property
    def repositories_hostname(self) -> str:
        # Use configured tool_shed_url if available
        if tool_shed_url := self.app.config.tool_shed_url:
            return tool_shed_url.rstrip("/")
        # Fall back to request-based URL
        return str(self.request.base).rstrip("/")

    @property
    def host(self):
        return self.__request.host

    @property
    def request(self) -> GalaxyAbstractRequest:
        return self.__request

    @property
    def response(self) -> GalaxyAbstractResponse:
        return self.__response

    # Following three things added v2.0 frontend
    @property
    def session_csrf_token(self):
        token = ""
        if self._galaxy_session:
            token = self.security.encode_id(self._galaxy_session.id, kind="csrf")
        return token

    @property
    def galaxy_session(self) -> GalaxySession | None:
        return self._galaxy_session

    def log_event(self, str):
        pass