Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
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,
Query,
)
from galaxy.managers.context import ProvidesUserContext
from galaxy.schema.schema import (
CreateLibrariesFromStore,
CreateLibraryPayload,
DeleteLibraryPayload,
LegacyLibraryPermissionsPayload,
LibraryAvailablePermissions,
LibraryCurrentPermissions,
LibraryLegacySummary,
LibraryPermissionAction,
LibraryPermissionScope,
LibraryPermissionsPayload,
LibrarySummary,
LibrarySummaryList,
UpdateLibraryPayload,
)
from galaxy.webapps.galaxy.api import (
depends,
DependsOnTrans,
Router,
)
from galaxy.webapps.galaxy.api.common import LibraryIdPathParam
from galaxy.webapps.galaxy.services.libraries import LibrariesService
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."
)
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,
id: LibraryIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> 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,
id: LibraryIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
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,
id: LibraryIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
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,
id: LibraryIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
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: int = Query(
default=1, title="Page", description="The page number to retrieve when paginating the available roles."
),
page_limit: 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,
id: LibraryIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
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.model_dump(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)