import re
from typing import List
from datetime import datetime, UTC
import sqlalchemy as sql
from sqlalchemy.orm import (
Mapped,
mapped_column,
relationship,
)
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy_utc.sqltypes import UtcDateTime
from app.model.orm.orm_base import OrmBase
[docs]
class Study(OrmBase):
[docs]
__tablename__ = 'Studies'
# A relationship representing ownership of these records. Clearing them out
# should directly delete them so they can be replaced.
[docs]
owner_relationship = lambda: relationship(
back_populates='study',
cascade='all, delete-orphan',
)
[docs]
studyUniqueID: Mapped[str] = mapped_column(sql.String(100), primary_key=True)
[docs]
ownerUniqueID: Mapped[str] = mapped_column(sql.ForeignKey('Users.uuid'))
[docs]
owner: Mapped['User'] = relationship(back_populates='ownedStudies')
[docs]
studyId: Mapped[str] = mapped_column(sql.String(100))
[docs]
studyName: Mapped[str] = mapped_column(sql.String(100))
[docs]
studyDescription: Mapped[str] = mapped_column(sql.String, nullable=True)
[docs]
studyURL: Mapped[str] = mapped_column(sql.String, nullable=True)
[docs]
timeUnits: Mapped[str] = mapped_column(sql.String(100))
[docs]
projectUniqueID: Mapped[str] = mapped_column(sql.ForeignKey('Projects.projectUniqueID'))
[docs]
project: Mapped['Project'] = relationship(back_populates="studies")
[docs]
createdAt: Mapped[datetime] = mapped_column(UtcDateTime, server_default=sql.FetchedValue())
[docs]
updatedAt: Mapped[datetime] = mapped_column(UtcDateTime, server_default=sql.FetchedValue())
[docs]
publishableAt: Mapped[datetime] = mapped_column(UtcDateTime, nullable=True)
[docs]
publishedAt: Mapped[datetime] = mapped_column(UtcDateTime, nullable=True)
[docs]
embargoExpiresAt: Mapped[datetime] = mapped_column(UtcDateTime, nullable=True)
[docs]
studyUsers: Mapped[List['StudyUser']] = owner_relationship()
[docs]
experiments: Mapped[List['Experiment']] = owner_relationship()
[docs]
strains: Mapped[List['Strain']] = owner_relationship()
[docs]
communities: Mapped[List['Community']] = owner_relationship()
[docs]
compartments: Mapped[List['Compartment']] = owner_relationship()
[docs]
measurementTechniques: Mapped[List['MeasurementTechnique']] = owner_relationship()
[docs]
measurementContexts: Mapped[List['MeasurementContext']] = owner_relationship()
[docs]
modelingRequests: Mapped[List['ModelingRequest']] = owner_relationship()
[docs]
experimentCompartments: Mapped[List['ExperimentCompartment']] = owner_relationship()
[docs]
bioreplicates: Mapped[List['Bioreplicate']] = owner_relationship()
[docs]
perturbations: Mapped[List['Perturbation']] = owner_relationship()
[docs]
measurements: Mapped[List['Measurement']] = relationship(
order_by='Measurement.timeInSeconds',
secondary='MeasurementContexts',
viewonly=True,
)
[docs]
modelingResults: Mapped[List['ModelingResult']] = relationship(
secondary='ModelingRequests',
viewonly=True,
)
@hybrid_property
[docs]
def uuid(self):
return self.studyUniqueID
@hybrid_property
[docs]
def publicId(self):
return self.studyId
@hybrid_property
[docs]
def name(self):
return self.studyName
@hybrid_property
[docs]
def description(self):
return self.studyDescription
@hybrid_property
[docs]
def isPublished(self):
return self.publishedAt
@hybrid_property
[docs]
def isPublishable(self):
return self.publishableAt and self.publishableAt <= datetime.now(UTC)
[docs]
def visible_to_user(self, user):
if self.isPublished:
return True
elif not user or not user.uuid:
return False
else:
return user.uuid in self.managerUuids
[docs]
def manageable_by_user(self, user):
if not user or not user.uuid:
return False
else:
return user.uuid in self.managerUuids
[docs]
def find_last_submission(self, db_session):
from app.model.orm import Submission
return db_session.scalars(
sql.select(Submission)
.where(Submission.studyUniqueID == self.uuid)
.order_by(Submission.updatedAt.desc())
.limit(1)
).one_or_none()
@property
[docs]
def managerUuids(self):
return {su.userUniqueID for su in self.studyUsers}
[docs]
def publish(self):
if not self.isPublishable:
return False
else:
self.publishedAt = datetime.now(UTC)
return True
@staticmethod
[docs]
def generate_public_id(db_session):
last_string_id = db_session.scalars(
sql.select(Study.publicId)
.order_by(Study.publicId.desc())
.limit(1)
).one_or_none()
if last_string_id:
last_numeric_id = int(re.sub(r'SMGDB0*', '', last_string_id))
else:
last_numeric_id = 0
return "SMGDB{:08d}".format(last_numeric_id + 1)