"""
All database transactions for the ESS, PPM and Reports go here.
We also include related templates to not separate them from their parent things.
"""
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
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 (
Measurements,
Reports,
Samples,
TemplateSamples,
TemplateMeasurements,
)
from .basics import Transaction
# =============================================================================
# measurement.py - related to P/M/M (measurement) management
# =============================================================================
[docs]def create_measurement(database: SQLAlchemy, info_measurement: dict) -> int:
"""
Create a new measurement.
We will also update the sample updated time to mark the change to the sample
related information.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
info_measurement: dict.
Entry of Measurements to be created.
Returns
-------
int
ID of the created measurement.
"""
new_measurement = Measurements(
creator_id=info_measurement["creator_id"],
short_dis=info_measurement["short_dis"],
long_dis=info_measurement["long_dis"],
sample_id=info_measurement["sample_id"],
mtype_id=info_measurement["mtype_id"],
creator=info_measurement["creator"],
)
if "instrument_id" in info_measurement:
new_measurement.instrument_id = info_measurement["instrument_id"]
if "created" in info_measurement:
try:
new_measurement.created = datetime.strptime(
info_measurement["created"], "%a, %d %b %Y %H:%M:%S %Z"
)
except ValueError:
new_measurement.created = datetime.utcnow()
with Transaction(database) as db_session:
db_session.add(new_measurement)
with Transaction(database) as db_session:
db_session.refresh(new_measurement)
new_measurement.measurement_sample.updated = datetime.utcnow()
return new_measurement.id
[docs]def update_measurement(database: SQLAlchemy, measurement: Measurements, info_measurement: dict) -> None:
"""
Update Measurement record in database.
Parameters
----------
database: FlaskSQLAlchemy class
measurement: Measurements
Entry of Measurements to be updated.
info_measurement: Dict
Returns
-------
None
"""
with Transaction(database):
measurement.short_dis = info_measurement["short_dis"]
measurement.long_dis = info_measurement["long_dis"]
measurement.mtype_id = info_measurement["mtype"]
measurement.creator = info_measurement["creator"]
measurement.measurement_sample.updated = datetime.utcnow()
if "instrument_id" in info_measurement:
measurement.instrument_id = info_measurement["instrument_id"]
else:
measurement.instrument_id = None
# =============================================================================
# report.py - related to Report management
# =============================================================================
[docs]def create_report(database: SQLAlchemy, info_report: dict) -> int:
"""
Add a report to the database and updates relevant timestamps.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
info_report : dict
Entry to Reports to be added to the database.
Returns
-------
int.
ID of the new created report.
"""
new_report = Reports(
creator_id=info_report["creator_id"],
title=info_report["title"],
long_dis=info_report["long_dis"],
creator=info_report["creator"],
sample_id=info_report["sample_id"],
measurement_id=info_report["measurement_id"],
)
if "created" in info_report:
try:
new_report.created = datetime.strptime(info_report["created"], "%a, %d %b %Y %H:%M:%S %Z")
except ValueError:
new_report.created = datetime.utcnow()
with Transaction(database) as db_session:
db_session.add(new_report)
with Transaction(database) as db_session:
db_session.refresh(new_report)
new_report.reports_sample.updated = datetime.utcnow()
new_report.report_measurement.updated = datetime.utcnow()
return new_report.id
[docs]def update_report(database: SQLAlchemy, report: Reports, report_data: dict) -> None:
"""
Update the title, long description, and timestamps of a report.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
report : Reports
Entry in Reports to be updated.
report_data: Dict.
Dict that contains the updated content of the report.
Returns
-------
None.
"""
with Transaction(database):
report.title = report_data["title"]
report.long_dis = report_data["long_dis"]
report.reports_sample.updated = datetime.utcnow()
report.report_measurement.updated = datetime.utcnow()
# =============================================================================
# sample.py - related to E/S/S (sample) management
# =============================================================================
[docs]def create_sample(database: SQLAlchemy, sample_data: dict) -> int:
"""
Create a new ESS entry.
Parameters
----------
database: SQLAlchemy
The database class we want the sample created in.
sample_data: dict
Record to contain info for sample creation.
Returns
-------
int
ID of the newly created sample.
"""
new_sample = Samples(
creator_id=sample_data["creator_id"],
group_id=sample_data["group_id"],
identifier=secure_filename(sample_data["identifier"]),
origin=sample_data["origin"],
short_dis=sample_data["short_dis"],
long_dis=sample_data["long_dis"],
creator=sample_data["creator"],
location_id=1,
)
if "created" in sample_data:
try:
new_sample.created = datetime.strptime(sample_data["created"], "%a, %d %b %Y %H:%M:%S %Z")
except ValueError:
new_sample.created = datetime.utcnow()
with Transaction(database) as db_session:
db_session.add(new_sample)
return new_sample.id
[docs]def update_sample(database: SQLAlchemy, sample: Samples, sample_data: dict) -> None:
"""
Update the metadata associated with a sample.
Pass a dict with the needed information (identifier, origin, long
description, and short description).
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
sample: Sample
Entry of Sample object to be updated.
sample_data: dict
Containing the updated fields.
Returns
-------
None.
"""
with Transaction(database):
sample.short_dis = sample_data["short_dis"]
sample.long_dis = sample_data["long_dis"]
sample.identifier = secure_filename(sample_data["identifier"])
sample.origin = sample_data["origin"]
sample.creator = sample_data["creator"]
# =============================================================================
# template.py - template management
# =============================================================================
[docs]def create_template_sample(database: SQLAlchemy, temp_info: dict) -> None:
"""
Create a template sample based on the data of a web form
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
temp_info: dict
Information needed to initialize a TemplateSamples object
Returns
-------
None.
"""
sample_template = TemplateSamples(
tname=temp_info["tname"],
identifier=temp_info["identifier"],
origin=temp_info["origin"],
short_dis=temp_info["short_dis"],
long_dis=temp_info["long_dis"],
creator_id=temp_info["creator_id"],
group_id=temp_info["group_id"],
)
with Transaction(database) as db_session:
db_session.add(sample_template)
[docs]def update_template_sample(database: SQLAlchemy, template: TemplateSamples, temp_info: dict) -> None:
"""
Update a template sample based on the data of a web form.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
template : TemplateSamples
temp_info: Dict
Returns
-------
None.
"""
with Transaction(database):
template.tname = temp_info["tname"]
template.short_dis = temp_info["short_dis"]
template.long_dis = temp_info["long_dis"]
template.identifier = temp_info["identifier"]
template.origin = temp_info["origin"]
[docs]def create_template_measurement(database: SQLAlchemy, temp_info: dict) -> None:
"""
Creates a template measurement based on the data of a web form.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
temp_info: Dict
Information needed to initialize a TemplateMeasurements object
Returns
-------
None.
"""
measurement_template = TemplateMeasurements(
tname=temp_info["tname"],
short_dis=temp_info["short_dis"],
long_dis=temp_info["long_dis"],
creator_id=temp_info["creator_id"],
group_id=temp_info["group_id"],
creator=temp_info["creator"],
mtype_id=temp_info["mtype_id"],
)
with Transaction(database) as db_session:
db_session.add(measurement_template)
[docs]def update_template_measurement(
database: SQLAlchemy, template: TemplateMeasurements, temp_info: dict
) -> None:
"""
Update a template measurement based on the data of a web form.
Parameters
----------
database: FlaskSQLAlchemy class
Database handler that the Flask APP is using.
template : TemplateMeasurements
temp_info: Dict
Returns
-------
None.
"""
with Transaction(database):
template.tname = temp_info["tname"]
template.short_dis = temp_info["short_dis"]
template.long_dis = temp_info["long_dis"]
template.creator = temp_info["creator"]
template.mtype_id = temp_info["mtype_id"]
template.group_id = temp_info["group_id"]