Source code for galaxy.managers.api_keys

from sqlalchemy import (
from typing_extensions import Protocol

from galaxy.model.base import transaction
from galaxy.structured_app import BasicSharedApp

[docs]class IsUserModel(Protocol): id: int
[docs]class ApiKeyManager:
[docs] def __init__(self, app: BasicSharedApp): = app self.session =
[docs] def get_api_key(self, user: IsUserModel): APIKeys = stmt = select(APIKeys).filter_by(, deleted=False).order_by(APIKeys.create_time.desc()).limit(1) return self.session.scalars(stmt).first()
[docs] def create_api_key(self, user: IsUserModel): guid = new_key = new_key.user_id = new_key.key = guid self.session.add(new_key) with transaction(self.session): self.session.commit() return new_key
[docs] def get_or_create_api_key(self, user: IsUserModel) -> str: # Logic Galaxy has always used - but it would appear to have a race # condition. Worth fixing? Would kind of need a message queue to fix # in multiple process mode. api_key = self.get_api_key(user) key = api_key.key if api_key else self.create_api_key(user).key return key
[docs] def delete_api_key(self, user: IsUserModel) -> None: """Marks the current user API key as deleted.""" # Before it was possible to create multiple API keys for the same user although they were not considered valid # So all non-deleted keys are marked as deleted for backward compatibility self._mark_all_api_keys_as_deleted( with transaction(self.session): self.session.commit()
def _mark_all_api_keys_as_deleted(self, user_id: int): APIKeys = stmt = ( update(APIKeys) .where(APIKeys.user_id == user_id) .where(APIKeys.deleted == false()) .values(deleted=True) .execution_options(synchronize_session="evaluate") ) return self.session.execute(stmt)