Source code for galaxy.webapps.galaxy.api.libraries

"""
API operations on a data library.
"""
import logging
from typing import (
    List,
    Optional,
    Union,
)

from fastapi import (
    Body,
    Path,
    Query,
)

from galaxy.managers.context import ProvidesUserContext
from galaxy.schema.fields import EncodedDatabaseIdField
from galaxy.schema.schema import (
    CreateLibrariesFromStore,
    CreateLibraryPayload,
    DeleteLibraryPayload,
    LegacyLibraryPermissionsPayload,
    LibraryAvailablePermissions,
    LibraryCurrentPermissions,
    LibraryLegacySummary,
    LibraryPermissionAction,
    LibraryPermissionScope,
    LibraryPermissionsPayload,
    LibrarySummary,
    LibrarySummaryList,
    UpdateLibraryPayload,
)
from galaxy.webapps.galaxy.services.libraries import LibrariesService
from . import (
    depends,
    DependsOnTrans,
    Router,
)

log = logging.getLogger(__name__)

router = Router(tags=["libraries"])

DeletedQueryParam: Optional[bool] = Query(
    default=None, title="Display deleted", description="Whether to include deleted libraries in the result."
)

LibraryIdPathParam: EncodedDatabaseIdField = Path(
    ..., title="Library ID", description="The encoded identifier of the Library."
)

UndeleteQueryParam: Optional[bool] = Query(
    default=None, title="Undelete", description="Whether to restore a deleted library."
)


[docs]@router.cbv
class FastAPILibraries:
    service: LibrariesService = depends(LibrariesService)

[docs]    @router.get(
        "/api/libraries",
        summary="Returns a list of summary data for all libraries.",
    )
    def index(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        deleted: Optional[bool] = DeletedQueryParam,
    ) -> LibrarySummaryList:
        """Returns a list of summary data for all libraries."""
        return self.service.index(trans, deleted)

[docs]    @router.get(
        "/api/libraries/deleted",
        summary="Returns a list of summary data for all libraries marked as deleted.",
    )
    def index_deleted(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
    ) -> LibrarySummaryList:
        """Returns a list of summary data for all libraries marked as deleted."""
        return self.service.index(trans, True)

[docs]    @router.get(
        "/api/libraries/{id}",
        summary="Returns summary information about a particular library.",
    )
    def show(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        id: EncodedDatabaseIdField = LibraryIdPathParam,
    ) -> LibrarySummary:
        """Returns summary information about a particular library."""
        return self.service.show(trans, id)

[docs]    @router.post(
        "/api/libraries",
        summary="Creates a new library and returns its summary information.",
        require_admin=True,
    )
    def create(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        payload: CreateLibraryPayload = Body(...),
    ) -> LibrarySummary:
        """Creates a new library and returns its summary information. Currently, only admin users can create libraries."""
        return self.service.create(trans, payload)

[docs]    @router.post(
        "/api/libraries/from_store",
        summary="Create libraries from a model store.",
        require_admin=True,
    )
    def create_from_store(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        payload: CreateLibrariesFromStore = Body(...),
    ) -> List[LibrarySummary]:
        return self.service.create_from_store(trans, payload)

[docs]    @router.patch(
        "/api/libraries/{id}",
        summary="Updates the information of an existing library.",
    )
    def update(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        id: EncodedDatabaseIdField = LibraryIdPathParam,
        payload: UpdateLibraryPayload = Body(...),
    ) -> LibrarySummary:
        """
        Updates the information of an existing library.
        """
        return self.service.update(trans, id, payload)

[docs]    @router.delete(
        "/api/libraries/{id}",
        summary="Marks the specified library as deleted (or undeleted).",
        require_admin=True,
    )
    def delete(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        id: EncodedDatabaseIdField = LibraryIdPathParam,
        undelete: Optional[bool] = UndeleteQueryParam,
        payload: Optional[DeleteLibraryPayload] = Body(default=None),
    ) -> LibrarySummary:
        """Marks the specified library as deleted (or undeleted).
        Currently, only admin users can delete or restore libraries."""
        if payload:
            undelete = payload.undelete
        return self.service.delete(trans, id, undelete)

[docs]    @router.get(
        "/api/libraries/{id}/permissions",
        summary="Gets the current or available permissions of a particular library.",
    )
    def get_permissions(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        id: EncodedDatabaseIdField = LibraryIdPathParam,
        scope: Optional[LibraryPermissionScope] = Query(
            None,
            title="Scope",
            description="The scope of the permissions to retrieve. Either the `current` permissions or the `available`.",
        ),
        is_library_access: Optional[bool] = Query(
            None,
            title="Is Library Access",
            description="Indicates whether the roles available for the library access are requested.",
        ),
        page: Optional[int] = Query(
            default=1, title="Page", description="The page number to retrieve when paginating the available roles."
        ),
        page_limit: Optional[int] = Query(
            default=10, title="Page Limit", description="The maximum number of permissions per page when paginating."
        ),
        q: Optional[str] = Query(
            None, title="Query", description="Optional search text to retrieve only the roles matching this query."
        ),
    ) -> Union[LibraryCurrentPermissions, LibraryAvailablePermissions]:
        """Gets the current or available permissions of a particular library.
        The results can be paginated and additionally filtered by a query."""
        return self.service.get_permissions(
            trans,
            id,
            scope,
            is_library_access,
            page,
            page_limit,
            q,
        )

[docs]    @router.post(
        "/api/libraries/{id}/permissions",
        summary="Sets the permissions to access and manipulate a library.",
    )
    def set_permissions(
        self,
        trans: ProvidesUserContext = DependsOnTrans,
        id: EncodedDatabaseIdField = LibraryIdPathParam,
        action: Optional[LibraryPermissionAction] = Query(
            default=None,
            title="Action",
            description="Indicates what action should be performed on the Library.",
        ),
        payload: Union[
            LibraryPermissionsPayload,
            LegacyLibraryPermissionsPayload,
        ] = Body(...),
    ) -> Union[LibraryLegacySummary, LibraryCurrentPermissions]:  # Old legacy response
        """Sets the permissions to access and manipulate a library."""
        payload_dict = payload.dict(by_alias=True)
        if isinstance(payload, LibraryPermissionsPayload) and action is not None:
            payload_dict["action"] = action
        return self.service.set_permissions(trans, id, payload_dict)