"""
Contains all database transaction functions for the basic moduls - User, Groups, Locations and init_db
"""
from copy import copy
from datetime import datetime
from typing import Optional, Type, List
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash
# from werkzeug.utils import secure_filename
# from flask import g # This should not be needed here and be always a parameter for the function
from ResearchNotes.database import (
Role,
Documents,
User,
Groups,
ParentLocations,
Locations,
MeasurementType,
)
# =============================================================================
# Database transaction context manager
# =============================================================================
[docs]class Transaction:
"""
Database transaction context manager.
Allows with as statement for our database operations.
"""
def __init__(self, database: SQLAlchemy):
self.db = database
def __enter__(self):
"""
Enter function to work with the with- construction.
Returns
-------
SQLAlchemy.session
Scoped Session of the database to write and read.
"""
return self.db.session
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""
Exit function for database to work with with-construction.
Parameters
----------
exc_val : Exception type
DESCRIPTION.
exc_val : Exception value
DESCRIPTION.
exc_tb : Exception traceback
DESCRIPTION.
Returns
-------
None.
"""
if not exc_val:
self.db.session.commit()
# =============================================================================
# Database transaction functions
# =============================================================================
# =============================================================================
# init function - most go to the cli interface
# =============================================================================
[docs]def create_roles(database: SQLAlchemy) -> None:
"""
Create the Admin, Supervisor, Student, and ExStudent roles. Also, location and Mtypes.
Some basic roles as well as needed basic entries for Location and Measurement Types) are
created.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
Returns
-------
None.
"""
with Transaction(database) as db_session:
db_session.add(Role(name="admin", description="Admin can configure and register users"))
db_session.add(
Role(
name="Supervisor",
description="Supervisor can see all samples of his group",
)
)
db_session.add(Role(name="Student", description="Standard user"))
db_session.add(
Role(
name="ExStudent",
description="User, who left group",
)
)
db_session.add(Role(name="StudentAdmin", description="Student that can do some administration"))
db_session.add(Groups(name="None"))
db_session.add(ParentLocations(name="Not defined", group_id=1))
db_session.add(Locations(name="None", parent_location_id=1, group_id=1))
db_session.add(MeasurementType(name="Other", group_id=1))
[docs]def create_admin(database: SQLAlchemy) -> None:
"""
Create a generic admin user.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
Returns
-------
None.
"""
with Transaction(database) as db_session:
admin_role = Role.query.filter_by(name="admin").first()
db_session.add(
User(
name="admin",
UserName="Joe Admin",
email="admin@nowhere.com",
password=generate_password_hash("admin"),
role_id=admin_role.id,
group_id=1,
)
)
[docs]def create_index(database: SQLAlchemy, label: str, user_info: dict) -> None:
"""
Create the first index of the Documents tree.
It is created for the group using the scheme group_name+'_index'. These pages are
protected through the database - mainly, one cannot rename them. As pages are
only deletable, without children, the main page cannot be deleted, when it has
some pages linked from it.
Parameters
----------
database: FlaskSQLAlchemy
Database handler that the Flask APP is using.
label: str
Label to be given to the document entry
user_info: dict
Information of user creating the index
Returns
-------
None.
"""
doc = Documents(
label=label,
title=f"Documents of the {user_info['group_name']} group",
body="Here some basic documentation of processes, measurement "
+ "setups or procedures as well as a documentation for machinery can be stored. "
+ "New entries "
+ "are created by creating a \\[\\[ WikiLink \\]\\] with two squared breaks "
+ "at the beginning and the end. A word is any combination of upper or lower "
+ "case letters, number, dashes, underscores and spaces surrounded by double "
+ "brackets.\n We recommend to make categories, but in the end, you are "
+ "free to order things as you wish.",
group_id=user_info["group"],
creator_id=user_info["user_id"],
updatetor_id=user_info["user_id"],
)
with Transaction(database) as db_session:
db_session.add(doc)
# =============================================================================
# auth.py - related to login and logout
# =============================================================================
# =============================================================================
# setup.py - user, group, places and measurement type management
# =============================================================================
[docs]def create_user(database: SQLAlchemy, user_info: dict) -> None:
"""
Create user entry.
Parameters
----------
database : FlaskSQLAlchemy class
Database handler that the Flask APP is using.
user_info : dict
Info dict containing the needed user information
"""
new_user = User(
name=user_info["name"],
UserName=user_info["UserName"],
email=user_info["email"],
password=generate_password_hash(user_info["password"]),
role_id=int(user_info["role_id"]),
group_id=int(user_info["group_id"]),
active=True,
)
with Transaction(database) as db_session:
db_session.add(new_user)
[docs]def update_user(database: SQLAlchemy, user: User, user_info: dict) -> None:
"""
Update user data based on information passed.
Parameters
----------
database: FlaskSQLAlchemy class.
Database handler that the Flask APP is using
user : User
Entry of User object to be updated.
user_info: Dict
Returns
-------
None.
"""
with Transaction(database):
user.name = user_info["name"]
user.UserName = user_info["UserName"]
user.email = user_info["email"]
user.role_id = int(user_info["role_id"])
user.group_id = int(user_info["group_id"])
user.active = True
[docs]def deactivate_user(database: SQLAlchemy, user: User) -> None:
"""
Deactivate user.
Registers a user deactivation in the database, by changing its role to
ExStudent, updating its "active" status, and resetting its password
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
user : User
Entry to User whose deactivation is registered.
Returns
-------
None.
"""
with Transaction(database):
user.role_id = Role.query.filter_by(name="ExStudent").first().id
user.active = False
user.password = "password"
[docs]def activate_user(database: SQLAlchemy, user: User) -> None:
"""
Reactivates user, if it was inactive (and makes it a Student).
Registers a user activation in the database, by changing its role to
Student and updating its "active" status
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
user : User
Entry to User whose activation is registered.
Returns
-------
None.
"""
with Transaction(database):
user.active = True
user.role_id = Role.query.filter_by(name="Student").first().id
[docs]def rename_group(database: SQLAlchemy, group: Groups, name: str) -> None:
"""
Renames a group
Changes the name of a group and relabels documents indexed to it
Parameters
----------
database : SQLAlchemy
Database handler that the Flask APP is using.
group : Groups.
Group to update name.
name : str
New name for group.
Returns
-------
None.
"""
doc = Documents.query.filter_by(label=group.name + "_index", group_id=group.id).first()
with Transaction(database):
group.name = name
if doc:
doc.label = name + "_index"
doc.title = f"Documents of the {name} group"
[docs]def delete_group(database: SQLAlchemy, group: Groups) -> None:
"""
Deletes a group
Removes the group itself, without removing any information associated with it.
Instead, all its members, samples, documents, mtypes, and locations are moved
to the orphan group None (gid = 1)
Parameters
----------
database : SQLAlchemy
database
group : Groups
group
Returns
--------
None
"""
# move members, samples, documents, mtypes, and locations to orphan group
change_gid: List = [
group.members,
group.samples,
group.documents,
group.mtypes,
group.locations,
]
orphan_group_id: int = 1
for models in change_gid:
for model in models:
with Transaction(database):
model.group_id = orphan_group_id
# then delete group
with Transaction(database) as db_session:
db_session.delete(group)
[docs]def create_parent_location(database: SQLAlchemy, location_info: dict) -> None:
"""
Creates a location in the database
Parameters
----------
database : SQLAlchemy
database
location_info: dict
Information needed to initialize a Locations object
Returns
-------
None.
"""
new_location = ParentLocations(
name=location_info["name"],
group_id=location_info["group_id"],
)
with Transaction(database) as db_session:
db_session.add(new_location)
create_location(
database,
{
"name": "None",
"group_id": location_info["group_id"],
"parent_location_id": new_location.pl_id,
},
)
[docs]def create_location(database: SQLAlchemy, location_info: dict) -> None:
"""
Creates a location in the database
Parameters
----------
database : SQLAlchemy
database
location_info: dict
Information needed to initialize a Locations object
Returns
-------
None.
"""
new_location = Locations(
name=location_info["name"],
group_id=location_info["group_id"],
parent_location_id=location_info.get("parent_location_id", 1),
)
with Transaction(database) as db_session:
db_session.add(new_location)
[docs]def create_mtype(database: SQLAlchemy, mtype_info: dict) -> None:
"""
Creates a measurement in the database
Parameters
----------
database : SQLAlchemy
The database/session object to write to.
mtype_info: dict
Information needed to initialize a Measurement Type object.
Returns
-------
None.
"""
new_mtype = MeasurementType(
name=mtype_info["name"],
group_id=mtype_info["group_id"],
)
with Transaction(database) as db_session:
db_session.add(new_mtype)