"""
Module dealing with the entry types - might be merged to entry.py
"""
import typing
from flask import (
Blueprint,
flash,
g,
redirect,
render_template,
url_for,
current_app,
abort,
)
from werkzeug.wrappers.response import Response
import ResearchNotes.database_transactions as dbt
from ResearchNotes.auth import role_required
from ResearchNotes.database import EntryType, db, MeasurementType, ItemType
from ResearchNotes.form import ConfigCreateMType
from ResearchNotes.url_security import token_decode
# from ResearchNotes.instruments import check_permission
bp = Blueprint("type_management", __name__, url_prefix="/type_management")
# =============================================================================
# entry types management for instruments
# =============================================================================
[docs]@bp.route("/<int:iid>/create_etype", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def create_etype(iid: int) -> str | Response:
"""
Register a new instrumentation journal entry type.
Parameters
----------
iid: int
ID of the instrument that the journal entry type is associated with
Returns
-------
str|Response
"""
if not (
iid in [i.id for i in g.user.group_member.owned_instruments]
or iid in [i.id for i in g.user.shared_instruments]
):
abort(
403,
description=f"instrument.get_instrument: Authorization failure. {g.user} "
+ f"tried to load instrument {iid}",
)
else:
existing_etypes = db.session.execute(
db.select(EntryType).order_by("name").filter_by(instrument_id=iid)
).scalars()
form = ConfigCreateMType()
if form.validate_on_submit():
name = form.mtype.data
if (
db.session.execute(db.select(EntryType).filter_by(name=name, instrument_id=iid)).scalar()
is not None
):
flash(f"Entry type {name} is already registered.", "alert-warning")
else:
dbt.create_etype(
db,
{
"name": name,
"instrument_id": iid,
},
)
flash("Entry type created.", "alert-success")
return redirect(url_for("instruments.instrument_view", iid=iid))
return render_template(
"setup/registertype.html", form=form, existing=existing_etypes, what="Entry Type"
)
[docs]@bp.route("/<int:etid>/update_etype", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def update_etype(etid: int) -> typing.Union[str, Response]:
"""
Update an existing instrumentation journal entry type.
Parameters
----------
etid : int
EntryType ID to update.
Returns
-------
str|Response
"""
etype = db.get_or_404(EntryType, etid)
if not (
etype.instrument_id in [i.id for i in g.user.group_member.owned_instruments]
or etype.instrument_id in [i.id for i in g.user.shared_instruments]
):
abort(
403,
description=f"instrument.get_instrument: Authorization failure. {g.user} "
+ f"tried to load instrument {etype.instrument_id}",
)
else:
existing_etypes = db.session.execute(
db.select(EntryType).order_by("name").filter_by(instrument_id=etype.instrument_id)
).scalars()
form = ConfigCreateMType(data={"mtype": etype.name})
if form.validate_on_submit():
name = form.mtype.data
if (
db.session.execute(
db.select(EntryType).filter_by(name=name, instrument_id=etype.instrument_id)
).scalar()
is not None
):
flash(f"Entry type {name} is already registered.", "alert-warning")
else:
dbt.update_etype(
db,
etype,
{
"name": name,
},
)
flash(f"Entry type {name} updated", "alert-info")
return redirect(url_for("instruments.instrument_view", iid=etype.instrument_id))
return render_template(
"setup/registertype.html", form=form, existing=existing_etypes, what="Entry Type"
)
[docs]@bp.route("/delete_etype/<string:token>")
@role_required(["Supervisor", "StudentAdmin"])
def delete_etype(token: str) -> Response:
"""
Delete a location where samples are stored.
Parameters
----------
token : str
Signed string that encodes the id of entry type to delete
Returns
-------
Flask.Response
Redirect to instrument view
"""
etid = token_decode(token, current_app.config["SEC_SESSION_KEY"], g.salt)
etype = db.session.get(EntryType, etid)
for entry in etype.entries:
flash(f'{entry.identifier} assigned to "None" type', "alert-info")
with dbt.Transaction(db):
entry.etype_id = None
with dbt.Transaction(db) as db_session:
db_session.delete(etype)
flash(f"{etype.name} successfully deleted", "alert-success")
return redirect(url_for("instruments.instrument_view", iid=etype.instrument_id))
# =============================================================================
# measurement types management for PPMs
# =============================================================================
[docs]@bp.route("/register_type", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def register_type() -> typing.Union[str, Response]:
"""
Register a new measurement type.
Returns
-------
str|Response
"""
exists = db.session.execute(
db.select(MeasurementType).order_by("name").filter_by(group_id=g.user.group_id)
).scalars()
form = ConfigCreateMType()
if form.validate_on_submit():
if not db.session.execute(
db.select(MeasurementType).filter_by(name=form.mtype.data, group_id=g.user.group_id)
).scalar():
dbt.create_mtype(db, {"name": form.mtype.data, "group_id": g.user.group_id})
flash("Measurement type created.", "alert-success")
return redirect(url_for("conf.index"))
flash(f"Measurement Type {form.mtype.data} is already registered.", "alert-warning")
return render_template(
"setup/registertype.html", form=form, existing=exists, what="Measurement Types"
)
[docs]@bp.route("/update_type/<int:mtype_id>", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def update_type(mtype_id: int) -> typing.Union[str, Response]:
"""
Update a measurement type name.
Parameters
----------
mtype_id : int
MeasurementType ID to change name for.
Returns
-------
str|Response
"""
exists = db.session.execute(
db.select(MeasurementType).order_by("name").filter_by(group_id=g.user.group_id)
).scalars()
mtype = db.get_or_404(MeasurementType, mtype_id)
form = ConfigCreateMType(data={"mtype": mtype.name})
if form.validate_on_submit():
typename = form.mtype.data
if (
db.session.execute(
db.select(MeasurementType).filter_by(name=typename, group_id=g.user.group_id)
).scalar()
is None
):
with dbt.Transaction(db):
mtype.name = typename
flash("Measurement type updated.", "alert-success")
return redirect(url_for("conf.index"))
flash(f"Measurement Type {typename} is already registered.", "alert-warning")
return render_template(
"setup/registertype.html", form=form, existing=exists, what="Measurement Types"
)
[docs]@bp.route("/delete_type/<string:token>")
@role_required(["Supervisor", "StudentAdmin"])
def delete_type(token: str) -> Response:
"""
Delete a type of measurement.
Parameters
----------
token : str
Signed string that encodes the id of the type of measurement to delete
Returns
-------
Flask.Response
Redirect to admin settings page
"""
gid = token_decode(token, current_app.config["SEC_SESSION_KEY"], g.salt)
mtype = db.session.get(MeasurementType, gid)
for member in mtype.measurements:
flash(f"P/P/M type {member.mtype_id} will become id 1", "alert-info")
with dbt.Transaction(db):
member.mtype_id = 1
with dbt.Transaction(db) as db_session:
db_session.delete(mtype)
flash(f"{mtype.name} was deleted", "alert-success")
return redirect(url_for("conf.index"))
# =============================================================================
# item types management for inventory
# =============================================================================
[docs]@bp.route("/register_itype", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def register_itype() -> typing.Union[str, Response]:
"""
Register a new measurement type.
Returns
-------
str|Response
"""
exists = db.session.execute(
db.select(ItemType).order_by("name").filter_by(group_id=g.user.group_id)
).scalars()
form = ConfigCreateMType()
if form.validate_on_submit():
if not db.session.execute(
db.select(ItemType).filter_by(name=form.mtype.data, group_id=g.user.group_id)
).scalar():
dbt.create_itemtype(db, {"name": form.mtype.data, "group_id": g.user.group_id})
flash("Item type created.", "alert-success")
return redirect(url_for("inventory.itemtypes"))
flash(f"Item Type {form.mtype.data} is already registered.", "alert-warning")
return render_template("setup/registertype.html", form=form, existing=exists, what="Item Type")
[docs]@bp.route("/update_itype/<int:it_id>", methods=("GET", "POST"))
@role_required(["Supervisor", "StudentAdmin"])
def update_itype(it_id: int) -> str | Response:
"""
Update an item type.
Returns
-------
str|Response
"""
exists = db.session.execute(
db.select(ItemType).order_by("name").filter_by(group_id=g.user.group_id)
).scalars()
itype = db.get_or_404(ItemType, it_id)
form = ConfigCreateMType(data={"mtype": itype.name})
if form.validate_on_submit():
if not db.session.execute(
db.select(ItemType).filter_by(name=form.mtype.data, group_id=g.user.group_id)
).scalar():
with dbt.Transaction(db):
itype.name = form.mtype.data
flash("Item type updated.", "alert-success")
return redirect(url_for("inventory.itemtypes"))
flash(f"Item Type {form.mtype.data} is already registered.", "alert-warning")
return render_template("setup/registertype.html", form=form, existing=exists, what="Item Type")
[docs]@bp.route("/htmx_delete_itype/<string:token>", methods=["DELETE"])
@role_required(["Supervisor", "StudentAdmin"])
def htmx_delete_itype(token: str) -> str:
"""
Delete an item type.
Parameters
----------
token : str
Signed string that encodes the id of the type of measurement to delete
Returns
-------
str
Render item type list
"""
it_id = token_decode(token, current_app.config["SEC_SESSION_KEY"], g.salt)
itype = db.get_or_404(ItemType, it_id)
current_app.logger.debug(f" htmx : Delete item type {itype.name}")
for item in itype.items:
flash(f"For item {item.name} the item type will become 'None'", "alert-info")
with dbt.Transaction(db):
item.itype_id = None
with dbt.Transaction(db) as db_session:
db_session.delete(itype)
current_app.logger.debug(" htmx : Item type deleted")
return render_template("inventory/itemtype_list.html")