catalog_entry

Test Catalog Entry CRUD endpoints.

create_catalog_entry(*, catalog_id, entry, current_user)

Create a MLTE catalog entry.

Parameters:

Name	Type	Description	Default
entry	CatalogEntry	The catalog entry to create	required

Returns:

Type	Description
CatalogEntry	The created catalog entry

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.post("/{catalog_id}/entry")
def create_catalog_entry(
    *,
    catalog_id: str,
    entry: CatalogEntry,
    current_user: AuthorizedUser,
) -> CatalogEntry:
    """
    Create a MLTE catalog entry.
    :param entry: The catalog entry to create
    :return: The created catalog entry
    """
    catalog_id = url_utils.revert_valid_url_part(catalog_id)
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            if not entry.header.catalog_id:
                entry.header.catalog_id = catalog_id

            catalog_session = catalog_stores.get_session(catalog_id)
            if catalog_session.read_only:
                raise errors.ForbiddenError(
                    f"The catalog '{catalog_id}' is read only."
                )

            return catalog_session.entry_mapper.create_with_header(
                entry, user=current_user.username
            )
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except errors.ErrorAlreadyExists as e:
            raise HTTPException(
                status_code=codes.ALREADY_EXISTS, detail=f"Exists: {e}"
            )
        except errors.ForbiddenError as e:
            raise HTTPException(
                status_code=codes.FORBIDDEN, detail=f"Forbidden: {e}"
            )
        except Exception as e:
            raise_http_internal_error(e)

delete_catalog_entry(*, catalog_id, catalog_entry_id, current_user)

Delete a MLTE catalog entry.

Parameters:

Name	Type	Description	Default
catalog_entry_id	str	The entry id	required

Returns:

Type	Description
CatalogEntry	The deleted entry

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.delete("/{catalog_id}/entry/{catalog_entry_id}")
def delete_catalog_entry(
    *,
    catalog_id: str,
    catalog_entry_id: str,
    current_user: AuthorizedUser,
) -> CatalogEntry:
    """
    Delete a MLTE catalog entry.
    :param catalog_entry_id: The entry id
    :return: The deleted entry
    """
    catalog_id = url_utils.revert_valid_url_part(catalog_id)
    catalog_entry_id = url_utils.revert_valid_url_part(catalog_entry_id)
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            catalog_session = catalog_stores.get_session(catalog_id)
            if catalog_session.read_only:
                raise errors.ForbiddenError(
                    f"The catalog '{catalog_id}' is read only."
                )

            return catalog_session.entry_mapper.delete(catalog_entry_id)
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except errors.ForbiddenError as e:
            raise HTTPException(
                status_code=codes.FORBIDDEN, detail=f"Forbidden: {e}"
            )
        except Exception as e:
            raise_http_internal_error(e)

edit_catalog_entry(*, catalog_id, entry, current_user)

Edit a MLTE entry.

Parameters:

Name	Type	Description	Default
entry	CatalogEntry	The entry to edit	required

Returns:

Type	Description
CatalogEntry	The edited entry

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.put("/{catalog_id}/entry")
def edit_catalog_entry(
    *,
    catalog_id: str,
    entry: CatalogEntry,
    current_user: AuthorizedUser,
) -> CatalogEntry:
    """
    Edit a MLTE entry.
    :param entry: The entry to edit
    :return: The edited entry
    """
    catalog_id = url_utils.revert_valid_url_part(catalog_id)
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            if not entry.header.catalog_id:
                entry.header.catalog_id = catalog_id

            catalog_session = catalog_stores.get_session(catalog_id)
            if catalog_session.read_only:
                raise errors.ForbiddenError(
                    f"The catalog '{catalog_id}' is read only."
                )

            return catalog_session.entry_mapper.edit_with_header(
                entry, current_user.username
            )
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except errors.ForbiddenError as e:
            raise HTTPException(
                status_code=codes.FORBIDDEN, detail=f"Forbidden: {e}"
            )
        except Exception as e:
            raise_http_internal_error(e)

list_catalog_entries(*, catalog_id, current_user)

List MLTE catalog entries, with details for each entry.

Returns:

Type	Description
List[CatalogEntry]	A collection of entries with their details.

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.get("/{catalog_id}/entry")
def list_catalog_entries(
    *,
    catalog_id: str,
    current_user: AuthorizedUser,
) -> List[CatalogEntry]:
    """
    List MLTE catalog entries, with details for each entry.
    :return: A collection of entries with their details.
    """
    catalog_id = url_utils.revert_valid_url_part(catalog_id)
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            return catalog_stores.get_session(
                catalog_id
            ).entry_mapper.list_details()
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

list_catalog_entries_all_catalogs(*, current_user)

List MLTE catalog entries, with details for each entry.

Returns:

Type	Description
List[CatalogEntry]	A collection of entries with their details.

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.get("s/entry")
def list_catalog_entries_all_catalogs(
    *,
    current_user: AuthorizedUser,
) -> List[CatalogEntry]:
    """
    List MLTE catalog entries, with details for each entry.
    :return: A collection of entries with their details.
    """
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            return catalog_stores.list_details()
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

list_catalogs(*, current_user)

List MLTE catalogs, returning their ids.

Returns:

Type	Description
List[CatalogReply]	A collection of catalog ids.

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.get("s")
def list_catalogs(
    *,
    current_user: AuthorizedUser,
) -> List[CatalogReply]:
    """
    List MLTE catalogs, returning their ids.
    :return: A collection of catalog ids.
    """
    with state_stores.catalog_stores_session() as catalog_stores:
        catalog_stores.sessions
        try:
            return [
                CatalogReply(
                    id=catalog_id,
                    read_only=catalog.read_only,
                    type=catalog.get_uri().type.value,
                )
                for catalog_id, catalog in catalog_stores.sessions.items()
            ]
        except Exception as e:
            raise_http_internal_error(e)

read_catalog_entry(*, catalog_id, catalog_entry_id, current_user)

Read a MLTE entry.

Parameters:

Name	Type	Description	Default
catalog_entry_id	str	The entry name	required

Returns:

Type	Description
CatalogEntry	The read entry

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.get("/{catalog_id}/entry/{catalog_entry_id}")
def read_catalog_entry(
    *,
    catalog_id: str,
    catalog_entry_id: str,
    current_user: AuthorizedUser,
) -> CatalogEntry:
    """
    Read a MLTE entry.
    :param catalog_entry_id: The entry name
    :return: The read entry
    """
    catalog_id = url_utils.revert_valid_url_part(catalog_id)
    catalog_entry_id = url_utils.revert_valid_url_part(catalog_entry_id)
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            return catalog_stores.get_session(catalog_id).entry_mapper.read(
                catalog_entry_id
            )
        except errors.ErrorNotFound as e:
            raise HTTPException(
                status_code=codes.NOT_FOUND, detail=f"{e} not found."
            )
        except Exception as e:
            raise_http_internal_error(e)

search(*, query, current_user)

Search MLTE catalog entries, with details for each entry.

Parameters:

Name	Type	Description	Default
query	Query	The search query.	required

Returns:

Type	Description
List[CatalogEntry]	A collection of entries with their details for the provided query.

Source code in mlte/backend/api/endpoints/catalog_entry.py

@router.post("s/entry/search")
def search(
    *,
    query: Query,
    current_user: AuthorizedUser,
) -> List[CatalogEntry]:
    """
    Search MLTE catalog entries, with details for each entry.
    :param query: The search query.
    :return: A collection of entries with their details for the provided query.
    """
    with state_stores.catalog_stores_session() as catalog_stores:
        try:
            return catalog_stores.search(query=query)
        except errors.ErrorNotFound as e:
            raise HTTPException(status_code=codes.NOT_FOUND, detail=f"{e}")
        except Exception as e:
            raise_http_internal_error(e)