diff options
Diffstat (limited to 'uploader/publications')
| -rw-r--r-- | uploader/publications/datatables.py | 2 | ||||
| -rw-r--r-- | uploader/publications/misc.py | 4 | ||||
| -rw-r--r-- | uploader/publications/models.py | 27 | ||||
| -rw-r--r-- | uploader/publications/pubmed.py | 10 | ||||
| -rw-r--r-- | uploader/publications/views.py | 119 |
5 files changed, 137 insertions, 25 deletions
diff --git a/uploader/publications/datatables.py b/uploader/publications/datatables.py index e07fafd..8b3d4a0 100644 --- a/uploader/publications/datatables.py +++ b/uploader/publications/datatables.py @@ -13,7 +13,7 @@ def fetch_publications( search: Optional[str] = None, offset: int = 0, limit: int = -1 -) -> tuple[dict, int, int, int]: +) -> tuple[tuple[dict, ...], int, int, int]: """Fetch publications from the database.""" _query = "SELECT * FROM Publication" _count_query = "SELECT COUNT(*) FROM Publication" diff --git a/uploader/publications/misc.py b/uploader/publications/misc.py index fca6f71..f0ff9c7 100644 --- a/uploader/publications/misc.py +++ b/uploader/publications/misc.py @@ -4,10 +4,10 @@ def publications_differences( filedata: tuple[dict, ...], dbdata: tuple[dict, ...], - pubmedid2pubidmap: tuple[dict, ...] + pubmedid2pubidmap: dict[int, int] ) -> tuple[dict, ...]: """Compute the differences between file data and db data""" - diff = tuple() + diff: tuple[dict, ...] = tuple() for filerow, dbrow in zip( sorted(filedata, key=lambda item: ( item["phenotype_id"], item["xref_id"])), diff --git a/uploader/publications/models.py b/uploader/publications/models.py index b199991..dcfa02b 100644 --- a/uploader/publications/models.py +++ b/uploader/publications/models.py @@ -1,6 +1,6 @@ """Module to handle persistence and retrieval of publication to/from MariaDB""" import logging -from typing import Iterable, Optional +from typing import Iterable from MySQLdb.cursors import DictCursor @@ -30,6 +30,7 @@ def create_new_publications( conn: Connection, publications: tuple[dict, ...] ) -> tuple[dict, ...]: + """Create new publications in the database.""" if len(publications) > 0: with conn.cursor(cursorclass=DictCursor) as cursor: cursor.executemany( @@ -47,7 +48,8 @@ def create_new_publications( return tuple({ **row, "publication_id": row["Id"] } for row in cursor.fetchall()) - return tuple() + + return tuple() def update_publications(conn: Connection , publications: tuple[dict, ...]) -> tuple[dict, ...]: @@ -69,6 +71,27 @@ def update_publications(conn: Connection , publications: tuple[dict, ...]) -> tu return tuple() +def delete_publications(conn: Connection , publications: tuple[dict, ...]): + """Delete multiple publications""" + publications = tuple(pub for pub in publications if bool(pub)) + if len(publications) > 0: + _pub_ids = tuple(pub["Id"] for pub in publications) + _paramstr = ", ".join(["%s"] * len(_pub_ids)) + _phenos_query = ( + "SELECT PublicationId, COUNT(PhenotypeId) FROM PublishXRef " + f"WHERE PublicationId IN ({_paramstr}) GROUP BY PublicationId;") + + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute(_phenos_query, _pub_ids) + _linked_phenos = cursor.fetchall() + if len(_linked_phenos) > 0: + raise Exception(# pylint: disable=[broad-exception-raised] + "Cannot delete publications with linked phenotypes.") + + cursor.execute( + f"DELETE FROM Publication WHERE Id IN ({_paramstr})", _pub_ids) + + def fetch_publication_by_id(conn: Connection, publication_id: int) -> dict: """Fetch a specific publication from the database.""" with conn.cursor(cursorclass=DictCursor) as cursor: diff --git a/uploader/publications/pubmed.py b/uploader/publications/pubmed.py index ed9b652..15bf701 100644 --- a/uploader/publications/pubmed.py +++ b/uploader/publications/pubmed.py @@ -1,5 +1,6 @@ """Module to interact with NCBI's PubMed""" import logging +from typing import Optional import requests from lxml import etree @@ -29,9 +30,7 @@ def __journal__(journal: etree.Element) -> dict: } def __author__(author: etree.Element) -> str: - return "%s %s" % ( - author.find("LastName").text, - author.find("Initials").text) + return f'{author.find("LastName").text} {author.find("Initials").text}' def __pages__(pagination: etree.Element) -> str: @@ -42,7 +41,7 @@ def __pages__(pagination: etree.Element) -> str: )) if start is not None else "" -def __abstract__(article: etree.Element) -> str: +def __abstract__(article: etree.Element) -> Optional[str]: abstract = article.find("Abstract/AbstractText") return abstract.text if abstract is not None else None @@ -88,7 +87,8 @@ def fetch_publications(pubmed_ids: tuple[int, ...]) -> tuple[dict, ...]: "db": "pubmed", "retmode": "xml", "id": ",".join(str(item) for item in pubmed_ids) - }) + }, + timeout=(9.13, 20)) if response.status_code == 200: return __process_pubmed_publication_data__(response.text) diff --git a/uploader/publications/views.py b/uploader/publications/views.py index 137052b..11732db 100644 --- a/uploader/publications/views.py +++ b/uploader/publications/views.py @@ -1,28 +1,30 @@ """Endpoints for publications""" import json -from MySQLdb.cursors import DictCursor from gn_libs.mysqldb import database_connection from flask import ( flash, request, - url_for, redirect, Blueprint, render_template, current_app as app) +from uploader.sui import sui_template + +from uploader.flask_extensions import url_for from uploader.authorisation import require_login +from uploader.route_utils import redirect_to_next from .models import ( + delete_publications, + update_publications, fetch_publication_by_id, create_new_publications, fetch_publication_phenotypes) from .datatables import fetch_publications -from gn_libs.debug import __pk__ - pubbp = Blueprint("publications", __name__) @@ -30,21 +32,21 @@ pubbp = Blueprint("publications", __name__) @require_login def index(): """Index page for publications.""" - with database_connection(app.config["SQL_URI"]) as conn: - return render_template("publications/index.html") + return render_template(sui_template("publications/index.html")) @pubbp.route("/list", methods=["GET"]) @require_login def list_publications(): + """Fetch publications that fulfill a specific search, or all of them, if + there is no search term.""" # request breakdown: # https://datatables.net/manual/server-side _page = int(request.args.get("draw")) _length = int(request.args.get("length") or '-1') _start = int(request.args.get("start") or '0') _search = request.args["search[value]"] - with (database_connection(app.config["SQL_URI"]) as conn, - conn.cursor(cursorclass=DictCursor) as cursor): + with database_connection(app.config["SQL_URI"]) as conn: _publications, _current_rows, _totalfiltered, _totalrows = fetch_publications( conn, _search, @@ -65,9 +67,15 @@ def list_publications(): def view_publication(publication_id: int): """View more details on a particular publication.""" with database_connection(app.config["SQL_URI"]) as conn: + publication = fetch_publication_by_id(conn, publication_id) + + if not bool(publication): + flash("Requested publication was not found!", "alert-warning") + return redirect(url_for('publications.index')) + return render_template( - "publications/view-publication.html", - publication=fetch_publication_by_id(conn, publication_id), + sui_template("publications/view-publication.html"), + publication=publication, linked_phenotypes=tuple(fetch_publication_phenotypes( conn, publication_id))) @@ -76,17 +84,25 @@ def view_publication(publication_id: int): @require_login def create_publication(): """Create a new publication.""" - if(request.method == "GET"): - return render_template("publications/create-publication.html") + _get_args = { + key: request.args[key] + for key in ("species_id", "population_id", "dataset_id", "return_to") + if bool(request.args.get(key)) + } + + if request.method == "GET": + return render_template( + sui_template("publications/create-publication.html"), + get_args=_get_args) form = request.form authors = form.get("publication-authors").encode("utf8") if authors is None or authors == "": flash("The publication's author(s) MUST be provided!", "alert alert-danger") - return redirect(url_for("publications.create", **request.args)) + return redirect(url_for("publications.create")) with database_connection(app.config["SQL_URI"]) as conn: publications = create_new_publications(conn, ({ - "pubmed_id": form.get("pubmed-id"), + "pubmed_id": form.get("pubmed-id") or None, "abstract": form.get("publication-abstract").encode("utf8") or None, "authors": authors, "title": form.get("publication-title").encode("utf8") or None, @@ -100,8 +116,81 @@ def create_publication(): return redirect(url_for( request.args.get("return_to") or "publications.view_publication", publication_id=publications[0]["publication_id"], - **request.args)) + **_get_args)) flash("Publication creation failed!", "alert alert-danger") app.logger.debug("Failed to create the new publication.", exc_info=True) return redirect(url_for("publications.create_publication")) + + +@pubbp.route("/edit/<int:publication_id>", methods=["GET", "POST"]) +@require_login +def edit_publication(publication_id: int): + """Edit a publication's details.""" + with database_connection(app.config["SQL_URI"]) as conn: + if request.method == "GET": + return render_template( + sui_template("publications/edit-publication.html"), + publication=fetch_publication_by_id(conn, publication_id), + linked_phenotypes=tuple(fetch_publication_phenotypes( + conn, publication_id)), + publication_id=publication_id) + + form = request.form + _pub = update_publications(conn, ({ + "publication_id": publication_id, + "pubmed_id": form.get("pubmed-id") or None, + "abstract": (form.get("publication-abstract") or "").encode("utf8") or None, + "authors": (form.get("publication-authors") or "").encode("utf8"), + "title": (form.get("publication-title") or "").encode("utf8") or None, + "journal": (form.get("publication-journal") or "").encode("utf8") or None, + "volume": (form.get("publication-volume") or "").encode("utf8") or None, + "pages": (form.get("publication-pages") or "").encode("utf8") or None, + "month": (form.get("publication-month") or "").encode("utf8").capitalize() or None, + "year": (form.get("publication-year") or "").encode("utf8") or None + },)) + + if not _pub: + flash("There was an error updating the publication details.", + "alert-danger") + return redirect(url_for( + "publications.edit_publication", publication_id=publication_id)) + + flash("Successfully updated the publication details.", + "alert-success") + return redirect_to_next({ + "uri": "publications.view_publication", + "publication_id": publication_id + }) + + +@pubbp.route("/delete/<int:publication_id>", methods=["GET", "POST"]) +@require_login +def delete_publication(publication_id: int): + """Delete a particular publication.""" + with database_connection(app.config["SQL_URI"]) as conn: + publication = fetch_publication_by_id(conn, publication_id) + linked_phenotypes=tuple(fetch_publication_phenotypes( + conn, publication_id)) + + if not bool(publication): + flash("Requested publication was not found!", "alert-warning") + return redirect(url_for('publications.index')) + + if len(linked_phenotypes) > 0: + flash("Cannot delete publication with linked phenotypes!", + "alert-warning") + return redirect(url_for( + sui_template("publications.view_publication"), + publication_id=publication_id)) + + if request.method == "GET": + return render_template( + sui_template("publications/delete-publication.html"), + publication=publication, + linked_phenotypes=linked_phenotypes, + publication_id=publication_id) + + delete_publications(conn, (publication,)) + flash("Deleted the publication successfully.", "alert-success") + return redirect(url_for("publications.index")) |