1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
|
"""Endpoints handling species."""
from pymonad.either import Left, Right, Either
from gn_libs.mysqldb import database_connection
from flask import (flash,
request,
url_for,
redirect,
Blueprint,
current_app as app)
from uploader.population import popbp
from uploader.platforms import platformsbp
from uploader.ui import make_template_renderer
from uploader.oauth2.client import oauth2_get, oauth2_post
from uploader.authorisation import require_login, require_token
from uploader.datautils import order_by_family, enumerate_sequence
from .models import (all_species,
save_species,
species_by_id,
update_species,
species_families)
speciesbp = Blueprint("species", __name__)
speciesbp.register_blueprint(popbp, url_prefix="/")
speciesbp.register_blueprint(platformsbp, url_prefix="/")
render_template = make_template_renderer("species")
@speciesbp.route("/", methods=["GET"])
@require_login
def list_species():
"""List and display all the species in the database."""
with database_connection(app.config["SQL_URI"]) as conn:
return render_template("species/list-species.html",
allspecies=enumerate_sequence(all_species(conn)))
@speciesbp.route("/<int:species_id>", methods=["GET"])
@require_login
def view_species(species_id: int):
"""View details of a particular species and menus to act upon it."""
with database_connection(app.config["SQL_URI"]) as conn:
species = species_by_id(conn, species_id)
if bool(species):
return render_template("species/view-species.html",
species=species,
activelink="view-species")
flash("Could not find a species with the given identifier.",
"alert-danger")
return redirect(url_for("species.view_species"))
@speciesbp.route("/create", methods=["GET", "POST"])
@require_login
def create_species():
"""Create a new species."""
# We can use uniprot's API to fetch the details with something like
# https://rest.uniprot.org/taxonomy/<taxonID> e.g.
# https://rest.uniprot.org/taxonomy/6239
with (database_connection(app.config["SQL_URI"]) as conn,
conn.cursor() as cursor):
if request.method == "GET":
return render_template("species/create-species.html",
families=species_families(conn),
activelink="create-species")
error = False
taxon_id = request.form.get("species_taxonomy_id", "").strip() or None
common_name = request.form.get("common_name", "").strip()
if not bool(common_name):
flash("The common species name MUST be provided.", "alert-danger")
error = True
scientific_name = request.form.get("scientific_name", "").strip()
if not bool(scientific_name):
flash("The species' scientific name MUST be provided.",
"alert-danger")
error = True
parts = tuple(name.strip() for name in scientific_name.split(" "))
if len(parts) != 2 or not all(bool(name) for name in parts):
flash("The scientific name you provided is invalid.", "alert-danger")
error = True
cursor.execute(
"SELECT * FROM Species WHERE FullName=%s", (scientific_name,))
res = cursor.fetchone()
if bool(res):
flash("A species already exists with the provided scientific name.",
"alert-danger")
error = True
family = request.form.get("species_family", "").strip()
if not bool(family):
flash("The species' family MUST be selected.", "alert-danger")
error = True
if bool(taxon_id):
cursor.execute(
"SELECT * FROM Species WHERE TaxonomyId=%s", (taxon_id,))
res = cursor.fetchone()
if bool(res):
flash("A species already exists with the provided scientific name.",
"alert-danger")
error = True
if error:
return redirect(url_for("species.create_species",
common_name=common_name,
scientific_name=scientific_name,
taxon_id=taxon_id))
species = save_species(
conn, common_name, scientific_name, family, taxon_id)
flash("Species saved successfully!", "alert-success")
return redirect(url_for("species.view_species", species_id=species["species_id"]))
@speciesbp.route("/<int:species_id>/edit-extra", methods=["GET", "POST"])
@require_login
@require_token
#def edit_species(species_id: int):
def edit_species_extra(token: dict, species_id: int):# pylint: disable=[unused-argument]
"""Edit a species' details.
Parameters
----------
token: A JWT token used for authorisation.
species_id: An identifier for the species being edited.
"""
def __failure__(res):
app.logger.debug(
"There was an error in the attempt to edit the species: %s", res)
flash(res, "alert-danger")
return redirect(url_for("species.view_species", species_id=species_id))
def __system_resource_uuid__(resources) -> Either:
sys_res = [
resource for resource in resources
if resource["resource_category"]["resource_category_key"] == "system"
]
if len(sys_res) != 1:
return Left("Could not find/identify a valid system resource.")
return Right(sys_res[0]["resource_id"])
def __check_privileges__(authorisations):
if len(authorisations.items()) != 1:
return Left("Got authorisations for more than a single resource!")
auths = tuple(authorisations.items())[0][1]
authorised = "system:species:edit-extra-info" in tuple(
privilege["privilege_id"]
for role in auths["roles"]
for privilege in role["privileges"])
if authorised:
return Right(authorised)
return Left("You are not authorised to edit species extra details.")
with database_connection(app.config["SQL_URI"]) as conn:
species = species_by_id(conn, species_id)
all_the_species = all_species(conn)
families = species_families(conn)
family_order = tuple(
item[0] for item in order_by_family(all_the_species)
if item[0][1] is not None)
if bool(species) and request.method == "GET":
return oauth2_get("auth/user/resources").then(
__system_resource_uuid__
).then(
lambda resource_id: oauth2_post(
"auth/resource/authorisation",
json={"resource-ids": [resource_id]})
).then(__check_privileges__).then(
lambda authorisations: render_template(
"species/edit-species.html",
species=species,
families=families,
family_order=family_order,
max_order_id = max(
row["OrderId"] for row in all_the_species
if row["OrderId"] is not None),
activelink="edit-species")
).either(__failure__, lambda res: res)
if bool(species) and request.method == "POST":
update_species(conn,
species_id,
request.form["species_name"],
request.form["species_fullname"],
request.form["species_family"],
int(request.form["species_familyorderid"]),
int(request.form["species_orderid"]))
flash("Updated species successfully.", "alert-success")
return redirect(url_for("species.edit_species_extra",
species_id=species_id))
flash("Species with the given identifier was not found!",
"alert-danger")
return redirect(url_for("species.list_species"))
|
