from typing import (
Optional,
TYPE_CHECKING,
)
from galaxy.model import User
from galaxy.structured_app import BasicSharedApp
if TYPE_CHECKING:
from galaxy.model import APIKeys
[docs]class ApiKeyManager:
[docs] def __init__(self, app: BasicSharedApp):
self.app = app
[docs] def get_api_key(self, user: User) -> Optional["APIKeys"]:
sa_session = self.app.model.context
api_key = (
sa_session.query(self.app.model.APIKeys)
.filter_by(user_id=user.id, deleted=False)
.order_by(self.app.model.APIKeys.create_time.desc())
.first()
)
return api_key
[docs] def create_api_key(self, user: User) -> "APIKeys":
guid = self.app.security.get_new_guid()
new_key = self.app.model.APIKeys()
new_key.user_id = user.id
new_key.key = guid
sa_session = self.app.model.context
sa_session.add(new_key)
sa_session.flush()
return new_key
[docs] def get_or_create_api_key(self, user: User) -> 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: User) -> None:
"""Marks the current user API key as deleted."""
sa_session = self.app.model.context
# 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
api_keys = sa_session.query(self.app.model.APIKeys).filter_by(user_id=user.id, deleted=False)
for api_key in api_keys:
api_key.deleted = True
sa_session.add(api_key)
sa_session.flush()