import os
import astropy.units as u
from astropy.coordinates import SkyCoord
from datetime import datetime, timezone
from sqlalchemy import (
Boolean,
Column,
DateTime,
Float,
ForeignKey,
Index,
Integer,
String,
Table,
Text,
UniqueConstraint,
Enum as SQLAlchemyEnum,
BigInteger,
)
from enum import Enum
from .config.config import ccat_ops_db_settings
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from typing import Optional
from .ccat_ops_db import Base
from .types import JSONB_VARIANT
[docs]
class LocationType(Enum):
"""Location type enum
The location type determines the role of the location in the data transfer system.
SOURCE: Telescope instrument computers
BUFFER: Input/transit buffers
LONG_TERM_ARCHIVE: Permanent storage
PROCESSING: Temporary processing areas
"""
SOURCE = "source" # Telescope instrument computers
BUFFER = "buffer" # Input/processing buffers
LONG_TERM_ARCHIVE = "long_term_archive" # Permanent storage
PROCESSING = "processing" # Temporary processing areas
[docs]
class StorageType(Enum):
"""Storage type enum
The storage type determines the physical storage medium of the location.
DISK: Traditional disk storage
S3: Object storage (AWS S3 or compatible)
TAPE: Tape-based archival storage
"""
DISK = "disk"
S3 = "s3"
TAPE = "tape"
[docs]
class RouteType(Enum):
"""Route type enum
The route type determines the route of the data transfer.
DIRECT:
RELAY: Route through intermediate site
CUSTOM: Custom location-to-location override
"""
DIRECT = "direct"
RELAY = "relay" # Route through intermediate site
CUSTOM = "custom" # Custom location-to-location override
[docs]
class Status(Enum):
"""Status enum
These Status labels are used to make the Status checks uniform accross the entire
data transfer system.
These are used accros many concepts such as RawDataPackage, DataTransferPackage,
DataTransfer, etc.
PENDING: Is used to mark a concept as waiting for execution e.g. a RawDataPackage in
PENDING will be scheduled to be build.
SCHEDULED: Marks a concept as scheduled for execution e.g. a RawDataPackage in
IN_PROGRESS: Marks that a celery job is working on this concept.
COMPLETED: Succesfully completed the concept.
FAILED: Failed to execute the concept.
"""
PENDING = ccat_ops_db_settings.status.PENDING
SCHEDULED = ccat_ops_db_settings.status.SCHEDULED
IN_PROGRESS = ccat_ops_db_settings.status.IN_PROGRESS
COMPLETED = ccat_ops_db_settings.status.COMPLETED
FAILED = ccat_ops_db_settings.status.FAILED
[docs]
class PackageState(Enum):
"""Package state enum
The package state determines the state of the package.
WAITING: Waiting for transfer
TRANSFERRING: Transferring
ARCHIVED: Archived
FAILED: Failed
"""
WAITING = "waiting" # Yellow hourglass - only in primary archive
TRANSFERRING = "transferring" # Blue rotating circle - part of DataTransferPackage
ARCHIVED = "archived" # Green checkmark - archived with all statuses completed
FAILED = "failed" # Red cross - any status failed
[docs]
class PhysicalCopyStatus(Enum):
PRESENT = "present"
STAGED = (
"staged" # Package staged and unpacked, physical RawDataPackage file removed
)
DELETION_POSSIBLE = "deletion_possible"
DELETION_PENDING = "deletion_pending"
DELETION_SCHEDULED = "deletion_scheduled"
DELETION_IN_PROGRESS = "deletion_in_progress"
DELETION_FAILED = "deletion_failed"
DELETED = "deleted"
# Terminal state for copies whose bytes are gone and unrecoverable (e.g. a dead
# host that was decommissioned). Distinct from DELETED, which denotes an orderly,
# intentional deletion. Provenance records (package/file) are kept; only the copy
# is marked lost. See DataLocation.decommissioned_at.
LOST = "lost"
[docs]
class TriggerType(Enum):
CONTINUOUS = "continuous"
CRON = "cron"
MANUAL = "manual"
[docs]
class RunStatus(Enum):
PENDING = "pending"
STAGING_DATA = "staging_data"
SUBMITTED = "submitted"
RUNNING = "running"
COLLECTING_RESULTS = "collecting_results"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
[docs]
class DataProductType(Enum):
SCIENCE = "science"
QA_METRIC = "qa_metric"
PLOT = "plot"
LOG = "log"
STATISTICS = "statistics"
INTERMEDIATE = "intermediate"
[docs]
class OperationKind(str, Enum):
"""Identity axis of a pipeline operation and the polymorphic discriminator
for the uniform Operation model (ops-db #94, ADR-0003).
The values are BYTE-IDENTICAL to the frozen breadcrumb strings already
stored in ``OperationFailureEvent.operation_type`` and used by recovery,
the circuit breaker, and routing. Keeping them equal makes this a code
refactor rather than a data migration, and makes the operation-kind drift
class (recovery's ``archive`` vs the task's ``long_term_archive``)
unrepresentable: this enum is the single source of truth for identity.
``str`` subclassing lets a member compare and serialize as its raw string,
so it can be stored directly in the discriminator column and matched
against the legacy breadcrumb strings without translation.
Note the deliberate name/value asymmetry on two members: PACKAGING ==
``raw_data_package`` and ARCHIVE == ``long_term_archive`` -- the names follow
the stage, the values follow the frozen strings.
"""
PACKAGING = "raw_data_package"
BUNDLING = "data_transfer_package"
TRANSFER = "transfer"
UNPACK = "unpack"
ARCHIVE = "long_term_archive"
STAGING = "staging"
[docs]
class Site(Base):
"""Represents a physical or logical site where data can be stored or processed."""
__tablename__ = "site"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False) # "CCAT", "Cologne", "Cornell"
short_name = Column(String(20), nullable=False) # "ccat", "cologne", "us"
site_location = Column(String(250), nullable=True) # "Atacama", "Germany", "USA"
long_term_archive_transfers = relationship(
"LongTermArchiveTransfer", back_populates="site"
)
locations = relationship("DataLocation", back_populates="site")
__table_args__ = (UniqueConstraint("short_name", name="uix_site_short_name"),)
[docs]
class DataLocation(Base):
"""Base class for all data storage locations with polymorphic storage types."""
__tablename__ = "data_location"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False) # "ccat_chai_computer", "fyst_buffer_1"
location_type = Column(SQLAlchemyEnum(LocationType, values_callable=lambda x: [e.value for e in x]), nullable=False)
site_id = Column(Integer, ForeignKey("site.id"), nullable=False)
site = relationship("Site", back_populates="locations")
# Buffer hierarchy and failover
active = Column(Boolean, default=True, nullable=False)
priority = Column(Integer, default=0, nullable=False) # Lower = higher priority
# Decommissioning: a non-null decommissioned_at retires the location permanently
# (e.g. a dead host whose data is unrecoverable). The row is KEPT so historical
# packages/copies still resolve to a known location; it is hidden from active
# operational views. active=False is a reversible pause; decommissioned_at is the
# terminal retirement marker. decommission_reason records why (audited separately
# in OperationalConfigLog).
decommissioned_at = Column(DateTime(timezone=True), nullable=True)
decommission_reason = Column(Text, nullable=True)
# Polymorphic setup
storage_type = Column(
SQLAlchemyEnum(StorageType, values_callable=lambda x: [e.value for e in x]), nullable=False
) # "disk", "s3", "tape"
__mapper_args__ = {"polymorphic_identity": None, "polymorphic_on": storage_type}
__table_args__ = (
UniqueConstraint("site_id", "name", name="uix_data_location_name"),
)
[docs]
class DiskDataLocation(DataLocation):
"""Disk-based storage location."""
__tablename__ = "disk_data_location"
id = Column(Integer, ForeignKey("data_location.id"), primary_key=True)
path = Column(String(500), nullable=False)
host = Column(String(250))
user = Column(String(250))
__mapper_args__ = {"polymorphic_identity": StorageType.DISK}
[docs]
class S3DataLocation(DataLocation):
"""S3-compatible object storage location."""
__tablename__ = "s3_data_location"
id = Column(Integer, ForeignKey("data_location.id"), primary_key=True)
bucket_name = Column(String(250), nullable=False)
region = Column(String(100))
# Self-serve connection config (non-secret), so an S3/LTA store can be
# commissioned from the admin UI without a data-transfer config redeploy.
# All nullable: data-transfer falls back to its dynaconf settings when unset
# (S3_ENDPOINTS / S3_METHOD / COSCINE_*), so existing deployments are
# unaffected until a location is populated. Secrets (S3 keys, Coscine token)
# stay in the environment, keyed per-location (see get_s3_credentials /
# get_coscine_api_token). The single endpoint_url assumes one URL reachable
# from every worker that routes to this location.
endpoint_url = Column(String(500), nullable=True)
# Access method for this location: "s3" (boto3) or "coscine". NULL => use the
# deployment-wide S3_METHOD. Lets one LTA be Coscine and another plain S3.
access_method = Column(String(20), nullable=True)
# boto3 S3 addressing style for this endpoint:
# True => path-style (https://endpoint/bucket/key) e.g. RDS, MinIO
# False => virtual-host (https://bucket.endpoint/key) e.g. NRW Datastorage
# NULL => boto3 default ("auto") / deployment-wide S3_PATH_STYLE
# Different S3 backends require different styles, so it belongs on the location.
path_style = Column(Boolean, nullable=True)
# Coscine target identifiers (not secrets) used when access_method == "coscine".
coscine_project = Column(String(250), nullable=True)
coscine_resource = Column(String(250), nullable=True)
__mapper_args__ = {"polymorphic_identity": StorageType.S3}
[docs]
def get_s3_credentials(self, site_name: str) -> tuple[str, str]:
"""Resolve S3 credentials from the environment for this location.
Credentials are looked up (via dynaconf) under the per-location name::
{Site.short_name}_{DataLocation.name}_S3_ACCESS_KEY_ID
{Site.short_name}_{DataLocation.name}_S3_SECRET_ACCESS_KEY
Because data-transfer's dynaconf prefix is ``CCAT_DATA_TRANSFER``, the
actual environment variables carry that prefix. For the Cologne site
(``short_name="cologne"``) with a ``long_term_archive`` location::
CCAT_DATA_TRANSFER_COLOGNE_LONG_TERM_ARCHIVE_S3_ACCESS_KEY_ID=...
CCAT_DATA_TRANSFER_COLOGNE_LONG_TERM_ARCHIVE_S3_SECRET_ACCESS_KEY=...
Do NOT embed the site name in the location name: the convention already
prefixes ``site.short_name``, so a location literally named
``cologne_long_term_archive`` yields the doubled, easy-to-miss
``..._COLOGNE_COLOGNE_LONG_TERM_ARCHIVE_...``. Name the location just
``long_term_archive``.
If the per-location variable is unset, this falls back to the global
``S3_ACCESS_KEY_ID`` / ``S3_SECRET_ACCESS_KEY`` (default placeholder
``"MUST_SET_VIA_ENV"``). A misspelled/doubled name therefore surfaces at
the S3 layer as ``InvalidAccessKeyId`` rather than a config error — the
data-transfer worker logs the resolved env-var name and a masked key
fingerprint (``utils.get_s3_client``) to make this diagnosable.
Parameters
----------
site_name : str
The site's ``short_name`` (passed by the pipeline).
Returns
-------
tuple[str, str]
(access_key_id, secret_access_key)
"""
from ccat_data_transfer.config.config import ccat_data_transfer_settings
# Construct environment variable names
access_key_var = f"{site_name}_{self.name}_S3_ACCESS_KEY_ID"
secret_key_var = f"{site_name}_{self.name}_S3_SECRET_ACCESS_KEY"
# Get credentials from DynaConf settings
access_key_id = getattr(ccat_data_transfer_settings, access_key_var, None)
secret_access_key = getattr(ccat_data_transfer_settings, secret_key_var, None)
# Fall back to global settings if location-specific ones aren't set
if not access_key_id:
access_key_id = ccat_data_transfer_settings.s3_access_key_id
if not secret_access_key:
secret_access_key = ccat_data_transfer_settings.s3_secret_access_key
return access_key_id, secret_access_key
[docs]
def get_coscine_api_token(self, site_name: str) -> str:
"""Get the Coscine API token for this location (secret, env-resolved).
Mirrors :meth:`get_s3_credentials`: looks for a per-location token
``{Site.name}_{DataLocation.name}_COSCINE_API_TOKEN`` and falls back to
the deployment-wide ``COSCINE_API_TOKEN``. The non-secret project/resource
live on the location (``coscine_project`` / ``coscine_resource``).
"""
from ccat_data_transfer.config.config import ccat_data_transfer_settings
token_var = f"{site_name}_{self.name}_COSCINE_API_TOKEN"
token = getattr(ccat_data_transfer_settings, token_var, None)
if not token:
token = ccat_data_transfer_settings.coscine_api_token
return token
[docs]
class TapeDataLocation(DataLocation):
"""Tape-based storage location."""
__tablename__ = "tape_data_location"
id = Column(Integer, ForeignKey("data_location.id"), primary_key=True)
library_name = Column(String(250))
mount_path = Column(String(500))
__mapper_args__ = {"polymorphic_identity": StorageType.TAPE}
raw_data_package_staging_job_association = Table(
"raw_data_package_staging_job_association",
Base.metadata,
Column("raw_data_package_id", Integer, ForeignKey("raw_data_package.id")),
Column("staging_job_id", Integer, ForeignKey("staging_job.id")),
)
# Copy-anchored lineage for the uniform Operation model (ops-db #94, ADR-0003).
# An Operation consumes input PhysicalCopy(s) and produces output PhysicalCopy(s).
# Lineage is traced through the artifact/copy graph rather than operation-to-
# operation FKs, because the pipeline's fan-in (bundling) and fan-out (unpack)
# live between artifacts. The consume/produce split is two tables, not one with a
# role column, so each direction is independently queryable from either side.
operation_consumes_physical_copy = Table(
"operation_consumes_physical_copy",
Base.metadata,
Column(
"operation_id",
Integer,
ForeignKey("operation.id", ondelete="CASCADE"),
primary_key=True,
),
Column(
"physical_copy_id",
Integer,
ForeignKey("physical_copy.id", ondelete="CASCADE"),
primary_key=True,
),
)
operation_produces_physical_copy = Table(
"operation_produces_physical_copy",
Base.metadata,
Column(
"operation_id",
Integer,
ForeignKey("operation.id", ondelete="CASCADE"),
primary_key=True,
),
Column(
"physical_copy_id",
Integer,
ForeignKey("physical_copy.id", ondelete="CASCADE"),
primary_key=True,
),
)
user_role_association = Table(
"user_role_association",
Base.metadata,
Column("user_id", Integer, ForeignKey("user.id"), primary_key=True),
Column("role_id", Integer, ForeignKey("role.id"), primary_key=True),
)
instrument_observing_program_association = Table(
"instrument_observing_program",
Base.metadata,
Column("instrument_id", Integer, ForeignKey("instrument.id"), primary_key=True),
Column(
"observing_program_id",
Integer,
ForeignKey("observing_program.id"),
primary_key=True,
),
)
obs_unit_instrument_module_configuration_association = Table(
"obs_unit_instrument_module_configuration_association",
Base.metadata,
Column("obs_unit_id", Integer, ForeignKey("obs_unit.id")),
Column(
"instrument_module_configuration_id",
Integer,
ForeignKey("instrument_module_configuration.id"),
),
)
[docs]
class ObservingProgram(Base):
__tablename__ = "observing_program"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
short_name = Column(String(20), nullable=False,
doc="short name without spaces to be used in selection")
description = Column(Text)
lead_id = Column(Integer, ForeignKey("user.id"), nullable=True)
lead = relationship("User", back_populates="observing_programs")
sub_observing_programs = relationship(
"SubObservingProgram",
back_populates="observing_program",
)
instruments = relationship(
"Instrument",
secondary=instrument_observing_program_association,
back_populates="observing_programs",
)
obs_units = relationship("ObsUnit", back_populates="observing_program")
[docs]
class SubObservingProgram(Base):
__tablename__ = "sub_observing_program"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
short_name = Column(String(20), nullable=False,
doc="short name without spaces to be used in selection")
description = Column(Text)
observing_program_id = Column(
Integer,
ForeignKey("observing_program.id"),
nullable=False,
)
observing_program = relationship(
"ObservingProgram",
back_populates="sub_observing_programs",
)
obs_units = relationship("ObsUnit", back_populates="sub_observing_program")
# The following tables implement a data transfer system from one side to the other with
# handshaking and deletion on the origin uncommented
# the Source Table needs a sqlalchemy column that resembles a SkyCoord object read from
# the ra_deg and dec_deg columns as well as the coordinate_system column and the epoch
# column
[docs]
class Source(Base):
"""A source is a celestial object
This class serves as a base class for various types of sources. It is a polymorphic
class in SQLAlchemy, not instantiated directly, but used to provide common
attributes. Subclasses, implemented as separate database tables, inherit from the
source class and can have additional specific attributes.
See classes that are based on this class for more information on the implemented
types of sources.
"""
__tablename__ = "source"
id = Column(Integer, primary_key=True)
name = Column(
String(250), nullable=False, unique=True, doc="The name of the source"
)
version = Column(
Integer, nullable=False, doc="Version number of the current parameters")
history = Column(JSONB_VARIANT, nullable=True, doc="Change history")
type = Column(String(250), nullable=False, doc="The polymorphic identity")
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "source",
}
# Add relationships
obs_units = relationship("ObsUnit", back_populates="source")
[docs]
class FixedSource(Source):
"""A source that has a fixed coordinates
This class is a subclass of the Source class and inherits all attributes from the
Source class.
The class implements the skycoord property that returns a SkyCoord object from the
ra_deg and dec_deg columns.
"""
__tablename__ = "fixed_source"
__mapper_args__ = {"polymorphic_identity": "fixed_source"}
id = Column(Integer, ForeignKey("source.id"), primary_key=True)
ra_deg = Column(Float, nullable=False, doc="The right ascension in degrees (ICRS)")
dec_deg = Column(Float, nullable=False, doc="The declination in degrees (ICRS)")
slam = Column(String(50), doc="Longitude string of the original input")
sbet = Column(String(50), doc="Latitude string of the original input")
vlsr = Column(Float, doc="The local standard of rest velocity in km/s")
frame = Column(String(50), doc="The frame of the coordinates of the original input")
@property
def skycoord(self):
"""Return a SkyCoord object from the ra_deg and dec_deg columns"""
if self.ra_deg is not None and self.dec_deg is not None:
return SkyCoord(
ra=self.ra_deg * u.deg,
dec=self.dec_deg * u.deg,
frame="icrs",
)
return None
[docs]
class SolarSystemObject(Source):
"""A source that is a solar system object
This class is a subclass of the Source class and inherits all attributes from the
Source class.
"""
__tablename__ = "solar_system_object"
__mapper_args__ = {"polymorphic_identity": "solar_system_object"}
id = Column(Integer, ForeignKey("source.id"), primary_key=True)
eph_name = Column(
String(50),
nullable=True,
doc="Standard ephemeris name of the source",
)
# https://naif.jpl.nasa.gov/pub/naif/toolkit_docs/FORTRAN/req/naif_ids.html#NAIF%20Integer%20ID%20codes
naif_id = Column(Integer, nullable=True, doc="The NAIF ID of the source")
[docs]
class ConstantElevationSource(Source):
"""A source that is observed at a constant elevation
This class is a subclass of the Source class and inherits all attributes from the
Source class.
"""
__tablename__ = "constant_elevation_source"
__mapper_args__ = {"polymorphic_identity": "constant_elevation_source"}
id = Column(Integer, ForeignKey("source.id"), primary_key=True)
ra_deg_min = Column(
Float, nullable=False, doc="The minimum right ascension of the area in degrees"
)
ra_deg_max = Column(
Float, nullable=False, doc="The maximum right ascension of the area in degrees"
)
dec_deg_min = Column(
Float, nullable=False, doc="The minimum declination of the area in degrees"
)
dec_deg_max = Column(
Float, nullable=False, doc="The maximum declination of the area in degrees"
)
slam_min = Column(
String(50), doc="The minimum longitude string of the original input"
)
slam_max = Column(
String(50), doc="The maximum longitude string of the original input"
)
sbet_min = Column(
String(50), doc="The minimum latitude string of the original input"
)
sbet_max = Column(
String(50), doc="The maximum latitude string of the original input"
)
vlsr = Column(Float, doc="The local standard of rest velocity in km/s")
frame = Column(String(50), doc="The frame of the coordinates of the original input")
[docs]
class Line(Base):
"""A spectral line that is observed by an instrument"""
__tablename__ = "line"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False, doc="The name of the spectral line")
rest_frequency = Column(Float, nullable=False, doc="The rest frequency of the line")
side_band = Column(String(250), nullable=False, doc="The side band of the line")
available = Column(Boolean, nullable=False,
doc="Whether this line is ready to be scheduled")
comment = Column(String(250), nullable=True)
# chai_array_configurations = relationship(
# "ChaiArrayConfiguration", back_populates="line"
# )
# To be able to also include data from other observatories such as GREAT/SOFIA we
# include a specific observatory table
[docs]
class Observatory(Base):
__tablename__ = "observatory"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False, unique=True)
description = Column(String(250), nullable=False)
telescopes = relationship("Telescope", back_populates="observatory")
# To prepare for the possibility of multiple telescopes in an observatory we include a
# telescope table
[docs]
class Telescope(Base):
__tablename__ = "telescope"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
description = Column(String(250), nullable=False)
lon_deg = Column(Float, nullable=False)
lat_deg = Column(Float, nullable=False)
alt_m = Column(Float, nullable=False)
instruments = relationship("Instrument", back_populates="telescope")
observatory_id = Column(Integer, ForeignKey("observatory.id"), nullable=False)
observatory = relationship("Observatory", back_populates="telescopes")
# An instrument runs on a specific telescope
[docs]
class Instrument(Base):
__tablename__ = "instrument"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
instrument_type = Column(String(250), nullable=False)
description = Column(String(250))
telescope_id = Column(Integer, ForeignKey("telescope.id"), nullable=False)
telescope = relationship("Telescope", back_populates="instruments")
modules = relationship("InstrumentModule", back_populates="instrument")
observing_programs = relationship(
"ObservingProgram",
secondary=instrument_observing_program_association,
back_populates="instruments",
)
available = Column(Boolean, nullable=False)
[docs]
class InstrumentModule(Base):
__tablename__ = "instrument_module"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
description = Column(String(250))
instrument_id = Column(Integer, ForeignKey("instrument.id"), nullable=False)
instrument = relationship("Instrument", back_populates="modules")
instrument_module_configurations = relationship(
"InstrumentModuleConfiguration", back_populates="instrument_module"
)
available = Column(Boolean, nullable=False)
raw_data_packages = relationship(
"RawDataPackage", back_populates="instrument_module"
)
[docs]
class InstrumentModuleConfiguration(Base):
__tablename__ = "instrument_module_configuration"
id = Column(Integer, primary_key=True)
type = Column(String)
instrument_module_id = Column(Integer, ForeignKey("instrument_module.id"))
instrument_module = relationship(
"InstrumentModule", back_populates="instrument_module_configurations"
)
raw_data_files = relationship(
"RawDataFile", back_populates="instrument_module_configuration"
)
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "instrument_module_configuration",
} # Polymorphic mapping
[docs]
class ChaiModuleConfiguration(InstrumentModuleConfiguration):
__tablename__ = "chai_module_configuration"
__mapper_args__ = {"polymorphic_identity": "chai_module_configuration"}
id = Column(
Integer, ForeignKey("instrument_module_configuration.id"), primary_key=True
)
line_id = Column(Integer, ForeignKey("line.id"), nullable=False)
line = relationship("Line")
if_ghz = Column(
Float,
doc="Intermediate frequency (IF) in GHz at the center of the signal sideband",
)
config_parameters = Column(
JSONB_VARIANT,
doc="List of instrument configuration parameters",
)
[docs]
class PrimeCamModuleConfiguration(InstrumentModuleConfiguration):
__tablename__ = "prime_cam_module_configuration"
__mapper_args__ = {
"polymorphic_identity": "prime_cam_module_configuration"
} # Polymorphic mapping
id = Column(
Integer, ForeignKey("instrument_module_configuration.id"), primary_key=True
)
config_parameters = Column(
JSONB_VARIANT,
doc="List of instrument configuration parameters",
)
[docs]
class ObservationConfiguration(Base):
__tablename__ = "observation_configuration"
id = Column(Integer, primary_key=True)
type = Column(String)
obs_units = relationship("ObsUnit", back_populates="observation_configuration")
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "observation_configuration",
}
azimuth_range = Column(
JSONB_VARIANT,
nullable=True,
doc="Azimuth range lookup table for constant elevation scans",
)
[docs]
class ChaiObservationConfiguration(ObservationConfiguration):
__tablename__ = "chai_observation_configuration"
__mapper_args__ = {"polymorphic_identity": "chai_observation_configuration"}
id = Column(Integer, ForeignKey("observation_configuration.id"), primary_key=True)
chai_tilings = relationship(
"ChaiTiling", back_populates="chai_observation_configuration"
)
ntilelines = Column(
Integer,
nullable=True,
doc="Number of tile lines to be grouped (for socring in scheduler)",
)
[docs]
class PrimeCamObservationConfiguration(ObservationConfiguration):
__tablename__ = "prime_cam_observation_configuration"
__mapper_args__ = {
"polymorphic_identity": "prime_cam_observation_configuration"
} # Polymorphic mapping
id = Column(Integer, ForeignKey("observation_configuration.id"), primary_key=True)
version = Column(
Integer, nullable=False, doc="Version number of the current parameters")
history = Column(JSONB_VARIANT, nullable=True, doc="Change history")
mapping_parameters = Column(
JSONB_VARIANT,
nullable=True,
doc="List of mapping parameters",
)
[docs]
class ObsMode(Base):
__tablename__ = "obs_mode"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
description = Column(String(250))
obs_units = relationship("ObsUnit", back_populates="obs_mode")
[docs]
class ObsUnit(Base):
__tablename__ = "obs_unit"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
version = Column(
Integer, nullable=False, doc="Version number of the current parameters")
history = Column(JSONB_VARIANT, nullable=True, doc="Change history")
phase = Column(
String(20),
nullable=False,
doc="Common name of campaigns that share the observing requests.",
)
group = Column(
String(250),
nullable=True,
doc="List of other ObsUnits that belong to the same group",
)
group_type = Column(
String(250),
nullable=True,
doc="Reason of grouping. 'equal' indicates balanced scheduling, 'either' indicates resetting cadence once one of them are observed.",
)
equal_tolerance = Column(
Float,
nullable=True,
doc="Tolerance of inbalanced schedule when group_type is equal",
)
min_alt = Column(
Float, nullable=True, doc="Minimum allowed altitude (elevation) [deg]")
max_alt = Column(
Float, nullable=True, doc="Maximum allowed altitude (elevation) [deg]")
min_rotang = Column(
Float, nullable=True, doc="Minimum allowed rotation angle [deg]")
max_rotang = Column(
Float, nullable=True, doc="Maximum allowed rotation angle [deg]")
nominal_alt = Column(
String(250),
nullable=True,
doc="Either fixed altitude in [deg] or 'var', which allows a flexible choice",
)
min_lsa = Column(Float, nullable=True, doc="Minimum LSA [deg]")
max_lsa = Column(Float, nullable=True, doc="Maximum LSA [deg]")
lsa_margin = Column(
Float, nullable=True, doc="Allowed deviation from min_lsa and max_lsa [deg]")
cadence = Column(
Float, nullable=True, doc="Requested cadence for scheduling [day]")
requested_time_h = Column(
Float,
doc="Requested total observing hours. Mandatory for PrimeCam. For CHAI it is calculated by contents in ChaiTiling",
nullable=True,
)
unit_duration_h = Column(Float, nullable=False, doc="Observing duration [h]")
trans_ref = Column(Float, nullable=True, doc="Reference transmission")
# Note for priorities: due to and enhaced usage in the scheduler,
# this needs to be float, not integer.
priorities = Column(Float, nullable=False, doc="Science priority")
available = Column(
Boolean, nullable=False, doc="Whether this ObsUnit is ready to be scheduled"
)
pre_scheduled_basis = Column(
Boolean,
doc="Whether this Obsunit is executed on pre-scheduled basis, meaning that it has a very low priority outside of the pre-scheduled slots.",
)
additional_parameters = Column(
JSONB_VARIANT, nullable=True, doc="Any special parameters or constraints to be stored"
)
source_id = Column(Integer, ForeignKey("source.id"), nullable=True)
source = relationship("Source", back_populates="obs_units")
observing_program_id = Column(
Integer,
ForeignKey("observing_program.id"),
nullable=False,
)
observing_program = relationship("ObservingProgram", back_populates="obs_units")
sub_observing_program_id = Column(Integer, ForeignKey("sub_observing_program.id"))
sub_observing_program = relationship(
"SubObservingProgram",
back_populates="obs_units",
)
obs_mode_id = Column(Integer, ForeignKey("obs_mode.id"), nullable=True)
obs_mode = relationship("ObsMode", back_populates="obs_units")
primary_instrument_module_configuration_id = Column(
Integer, ForeignKey("instrument_module_configuration.id"), nullable=False
)
primary_instrument_module_configuration = relationship(
"InstrumentModuleConfiguration",
backref="primary_obs_units",
)
# Note: it is a bit redundant but the following list includes
# primary_instrument_module_configuration as well
# so that it is easier to retrieve all instrument module configuration
# that are associated to this obsunit
instrument_module_configurations = relationship(
"InstrumentModuleConfiguration",
secondary=obs_unit_instrument_module_configuration_association,
backref="obs_units",
)
observation_configuration_id = Column(
Integer, ForeignKey("observation_configuration.id"), nullable=True
)
observation_configuration = relationship(
"ObservationConfiguration", back_populates="obs_units"
)
executed_obs_units = relationship("ExecutedObsUnit", back_populates="obs_unit")
pre_scheduled_slots = relationship("PreScheduledSlot", back_populates="obs_unit")
raw_data_packages = relationship("RawDataPackage", back_populates="obs_unit")
[docs]
class PreScheduledSlot(Base):
__tablename__ = "pre_scheduled_slot"
id = Column(Integer, primary_key=True)
start_time = Column(DateTime(timezone=True), nullable=False)
end_time = Column(DateTime(timezone=True), nullable=False)
obs_unit_id = Column(Integer, ForeignKey("obs_unit.id"), nullable=False)
obs_unit = relationship("ObsUnit", back_populates="pre_scheduled_slots")
[docs]
class ExecutedObsUnit(Base):
__tablename__ = "executed_obs_unit"
id = Column(UUID(as_uuid=True), primary_key=True)
start_time = Column(DateTime(timezone=True), nullable=False, index=True)
end_time = Column(DateTime(timezone=True))
status = Column(
String(20), nullable=True,
doc="Status of this observation. 'running' indicates currently running, '*success*' or '*completed*' will be counted as a successfully completed observation.",
)
mean_pwv = Column(Float, nullable=True, doc="Mean pwv [mm] during observation")
mean_elevation = Column(Float, nullable=True,
doc="Mean elevation [deg] during observation")
achievement_factor = Column(
Float,
nullable=True,
doc="Correction factor for the achieved time compared to the reference transmission",
)
quality = Column(String(20), nullable=True,
doc="If *discard*, not considered in counting achievement")
obs_unit_id = Column(Integer, ForeignKey("obs_unit.id"), nullable=False)
obs_unit = relationship("ObsUnit", back_populates="executed_obs_units")
obs_info = Column(
JSONB_VARIANT,
nullable=True,
doc="Static ancillary infomation (e.g. ObsUnit version, tiling ID for CHAI)",
)
obs_progress = Column(JSONB_VARIANT, nullable=True, doc="Field for progress tracking")
raw_data_packages = relationship(
"RawDataPackage", back_populates="executed_obs_unit"
)
raw_data_files = relationship("RawDataFile", back_populates="executed_obs_unit")
[docs]
class ChaiTiling(Base):
__tablename__ = "chai_tiling"
id = Column(Integer, primary_key=True)
version = Column(
Integer, nullable=False, doc="Version number of the current parameters")
history = Column(JSONB_VARIANT, nullable=True, doc="Change history")
priority_in_tiling = Column(Integer, nullable=False)
tile_id = Column(String(10), nullable=False, doc="CHAI internal tile ID")
tile_offset_x = Column(Float, nullable=False, doc="tile position along x-axis")
tile_offset_y = Column(Float, nullable=False, doc="tile position along y-axis")
x_or_y = Column(String(2), doc="Scan direction")
tile_unit_scaling_x = Column(
Float,
doc="Scaling factor for the script-defined tiling unit for x-direction",
)
tile_unit_scaling_y = Column(
Float,
doc="Scaling factor for the script-defined tiling unit for y-direction",
)
edge = Column(
String(10), doc="Edge name if half of the tile should be observed")
goal_ncycle = Column(Integer, nullable=False, doc="Goal number of cycle")
# tiling is connected to obsunit through observation configuration
chai_observation_configuration_id = Column(
Integer, ForeignKey("chai_observation_configuration.id")
)
chai_observation_configuration = relationship(
"ChaiObservationConfiguration", back_populates="chai_tilings"
)
chai_inpar_parameter_id = Column(Integer, ForeignKey("chai_inpar_parameter.id"))
chai_inpar_parameter = relationship(
"ChaiInparParameter", back_populates="chai_tilings"
)
[docs]
class ChaiInparParameter(Base):
__tablename__ = "chai_inpar_parameter"
id = Column(Integer, primary_key=True)
# nullable=True is default
name = Column(String(100), nullable=False)
version = Column(
Integer, nullable=False, doc="Version number of the current parameters")
history = Column(JSONB_VARIANT, doc="Change history")
chai_tilings = relationship("ChaiTiling", back_populates="chai_inpar_parameter")
lam = Column(Float, nullable=False, doc="Map center offset in longitude [arcsec]")
bet = Column(Float, nullable=False, doc="Map center offset in latitude [arcsec]")
cormap = Column(String(10), nullable=False, doc="Map cood. system")
line_range = Column(
String(50),
doc='Range of velocities [km/s] affected by lines (e.g. "-70:-30:v,-10:10:v")',
)
goal_resolution = Column(
Float,
doc="Spectral resolution [km/s] with which the scientific analysis will be done",
)
refname = Column(
String(20),
doc="Reference name when using the absolute positions. Set NAN to use relative",
)
refoffl = Column(
Float, doc="Reference position relative to on in longitude [arcsec]"
)
refoffb = Column(
Float, doc="Reference position relative to on in latitude [arcesc]"
)
corref = Column(String(10), doc="Reference cood. system")
mode = Column(String(10), doc="Observing mode; otfl or otfb")
otfpattern = Column(
String(50), doc="OTF pattern file name specified in kosma_software"
)
ton = Column(Float, doc="On integration time [sec]")
toff = Column(
Float,
doc="Forced off time [s] (-1 to use the default calculation = default)",
)
repetition = Column(Integer, doc="Repetition number")
offononoff = Column(
Integer,
doc="Sequence of OFF and ON (1 = OFF-ON-ON-OFF (default), 0 = OFF-ON)",
)
stepl = Column(Float, doc="Step size in longitude [arcsec]")
stepb = Column(Float, doc="Step size in latitude [arcsec]")
mapsizel = Column(Float, doc="Size of map in longitude [arcsec]")
mapsizeb = Column(Float, doc="Size of map in latitude [arcsec]")
nmapl = Column(Integer, doc="number of raster positions along longitude")
nmapb = Column(Integer, doc="number of raster positions along latitude")
mapangle = Column(Float, doc="Position angle of the map (counter-clock) [degree]")
crosssizel = Column(Float, doc="Size of cross in longitude [arcsec]")
crosssizeb = Column(Float, doc="Size of cross in latitude [arcsec]")
reverseflg = Column(
Integer,
doc=" OTF scan direction in cross : 0 = (+x,+y), 1 = (-x,-y), 2 = (-x,+x,-y,+y)",
)
scan_dir = Column(Integer, doc="OTF scan direction (1 is +x or +y, -1 is -x or -y)")
scan_order = Column(
Integer, doc="Order of OTF lines (1 is +x or +y, -1 is -x or -y)"
)
evendump = Column(
Integer,
doc="If 1 allow even number of dump positions without hitting the center. Default = 0",
)
novertical = Column(
Integer,
doc="If 1 skip the second (vertical) scan for cross observations. Default = 0",
)
pointingflg = Column(
Integer,
doc="Flag to indicate that it is a pointing session. Default = 0",
)
offperload = Column(Integer, doc="Define how many off for one load measurment")
onperload = Column(
Integer, doc="Define how many on positions for one load measurement"
)
repperload = Column(
Integer, doc="Define how many repetition for one load measurment"
)
lineperoff = Column(
Integer, doc="Define how many otf scan lines for one off measurement"
)
onperoff = Column(
Integer, doc="Define how many on positions for one off measurement"
)
offperpattern = Column(Integer, doc="Define how many off measurement per pattern")
refpoint = Column(String(20), doc="Reference point (as is used in setpoint)")
act_pixflg = Column(
Integer,
doc="Flag to use the actual pixel position of refpoint. Default = 0",
)
nextflg = Column(
String(1),
doc="If Y, enable telescope moving when writing data. Default = N",
)
[docs]
class Role(Base):
__tablename__ = "role"
id = Column(Integer, primary_key=True)
name = Column(String(150), unique=True)
description = Column(String(150), nullable=True)
# GitHub team mappings for automatic role assignment
github_team_mappings = Column(
JSONB_VARIANT, nullable=True, doc="GitHub teams that map to this role"
)
# Permission scopes for this role
permissions = Column(
JSONB_VARIANT, nullable=True, doc="List of permissions granted to this role"
)
users = relationship(
"User", secondary=user_role_association, back_populates="roles"
)
[docs]
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
email = Column(String(150))
username = Column(String(150))
first_name = Column(String(150), nullable=True)
last_name = Column(String(150), nullable=True)
title = Column(String(150), nullable=True)
affiliation = Column(Text, nullable=True)
password = Column(String(150))
# GitHub integration fields
github_id = Column(String(50), nullable=True, unique=True, doc="GitHub user ID")
github_username = Column(String(150), nullable=True, doc="GitHub username")
# User preferences stored as JSON
preferences = Column(JSONB_VARIANT, nullable=True, doc="User preferences and settings")
roles = relationship(
"Role", secondary=user_role_association, back_populates="users"
)
# Clean up duplicate fields (removing duplicates)
last_login_at = Column(DateTime())
current_login_at = Column(DateTime())
last_login_ip = Column(String(100))
current_login_ip = Column(String(100))
login_count = Column(Integer)
active = Column(Boolean())
confirmed_at = Column(DateTime())
# Relationships
observing_programs = relationship("ObservingProgram", back_populates="lead")
api_tokens = relationship(
"ApiToken", back_populates="user", cascade="all, delete-orphan"
)
[docs]
class ApiToken(Base):
"""API tokens for programmatic access to the API"""
__tablename__ = "api_token"
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
user = relationship("User", back_populates="api_tokens")
# Token identification
name = Column(String(100), nullable=False, doc="Human-readable name for the token")
token_hash = Column(
String(255), nullable=False, unique=True, doc="Hashed token value"
)
token_prefix = Column(
String(10), nullable=False, doc="First few characters for identification"
)
# Token permissions and scopes
scopes = Column(JSONB_VARIANT, nullable=True, doc="List of permission scopes for this token")
# Token lifecycle
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
expires_at = Column(
DateTime(timezone=True), nullable=True, doc="Token expiration time"
)
last_used_at = Column(
DateTime(timezone=True), nullable=True, doc="Last time token was used"
)
active = Column(
Boolean, default=True, nullable=False, doc="Whether token is active"
)
# Usage tracking
usage_count = Column(
Integer, default=0, nullable=False, doc="Number of times token was used"
)
last_used_ip = Column(String(100), nullable=True, doc="IP address of last usage")
[docs]
def is_expired(self):
"""Check if token is expired"""
if self.expires_at is None:
return False
return datetime.now(timezone.utc) > self.expires_at
[docs]
def is_valid(self):
"""Check if token is valid (active and not expired)"""
return self.active and not self.is_expired()
[docs]
class DataTransferRoute(Base):
"""Defines routes for data transfer between sites and locations.
The database implements a flexible routing system that supports:
1. Direct routes between specific locations
2. Relay routes through intermediate sites
3. Custom location-to-location overrides
"""
__tablename__ = "data_transfer_route"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
# Site-level routing
origin_site_id = Column(Integer, ForeignKey("site.id"), nullable=False)
origin_site = relationship("Site", foreign_keys=[origin_site_id])
destination_site_id = Column(Integer, ForeignKey("site.id"), nullable=False)
destination_site = relationship("Site", foreign_keys=[destination_site_id])
# Route configuration
route_type = Column(SQLAlchemyEnum(RouteType, values_callable=lambda x: [e.value for e in x]), nullable=False)
transfer_method = Column(String(250), nullable=False) # "bbcp", "s3", "cp"
# Optional location-specific overrides
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=True)
origin_location = relationship("DataLocation", foreign_keys=[origin_location_id])
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=True
)
destination_location = relationship(
"DataLocation", foreign_keys=[destination_location_id]
)
# Relay configuration
relay_site_id = Column(Integer, ForeignKey("site.id"), nullable=True)
relay_site = relationship("Site", foreign_keys=[relay_site_id])
__table_args__ = (
UniqueConstraint(
"origin_site_id",
"destination_site_id",
"route_type",
name="uix_data_transfer_route",
),
)
data_transfer_package_files_association = Table(
"data_transfer_package_files",
Base.metadata,
Column(
"data_transfer_package_id",
Integer,
ForeignKey("data_transfer_package.id"),
primary_key=True,
),
Column(
"file_id", UUID(as_uuid=True), ForeignKey("raw_data_file.id"), primary_key=True
),
)
[docs]
class RawDataFile(Base):
"""Represents a raw data file from an instrument."""
__tablename__ = "raw_data_file"
id = Column(UUID(as_uuid=True), primary_key=True)
name = Column(String(250), nullable=False)
relative_path = Column(String(250), nullable=False)
created_at = Column(
DateTime(timezone=True),
nullable=False,
default=lambda: datetime.now(timezone.utc),
)
instrument_module_configuration_id = Column(
Integer, ForeignKey("instrument_module_configuration.id"), nullable=False
)
instrument_module_configuration = relationship(
"InstrumentModuleConfiguration", back_populates="raw_data_files"
)
file_type = Column(String(250), nullable=False)
size = Column(BigInteger, nullable=False)
checksum = Column(String(250), nullable=False)
description = Column(Text, nullable=True)
# Source location tracking
source_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=False)
source_location = relationship("DataLocation")
# Package relationships
raw_data_package_id = Column(
Integer, ForeignKey("raw_data_package.id", ondelete="CASCADE"), nullable=True
)
raw_data_package = relationship("RawDataPackage", back_populates="raw_data_files")
data_transfer_package_id = Column(
Integer, ForeignKey("data_transfer_package.id"), nullable=True
)
data_transfer_package = relationship(
"DataTransferPackage", back_populates="raw_data_files"
)
executed_obs_unit_id = Column(
UUID(as_uuid=True), ForeignKey("executed_obs_unit.id"), nullable=False
)
executed_obs_unit = relationship("ExecutedObsUnit", back_populates="raw_data_files")
# Physical copies
state = Column(SQLAlchemyEnum(PackageState, values_callable=lambda x: [e.value for e in x]), nullable=True)
physical_copies = relationship(
"RawDataFilePhysicalCopy",
back_populates="raw_data_file",
cascade="all, delete-orphan",
)
[docs]
class RawDataPackage(Base):
"""A raw data package is a bundle of raw data files that were observed in an
observation unit. But they should never be larger than 50GB in size.
"""
__tablename__ = "raw_data_package"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
relative_path = Column(
String(250),
nullable=False,
doc="""
This is the relative path to the raw data package. The absolute path of the location
of this file for each archive is stored in the DataLocation table. The path is
relative to the raw_data_path of the DataLocation.
""",
)
size = Column(BigInteger, nullable=False)
executed_obs_unit_id = Column(
UUID(as_uuid=True),
ForeignKey("executed_obs_unit.id", ondelete="CASCADE"),
nullable=False,
)
executed_obs_unit = relationship(
"ExecutedObsUnit", back_populates="raw_data_packages"
)
instrument_module_id = Column(
Integer, ForeignKey("instrument_module.id"), nullable=False
)
instrument_module = relationship(
"InstrumentModule", back_populates="raw_data_packages"
)
obs_unit_id = Column(
Integer, ForeignKey("obs_unit.id", ondelete="CASCADE"), nullable=False
)
obs_unit = relationship("ObsUnit", back_populates="raw_data_packages")
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
checksum = Column(String(250), nullable=False)
raw_data_files = relationship("RawDataFile", back_populates="raw_data_package")
data_transfer_package_id = Column(
Integer,
ForeignKey("data_transfer_package.id", ondelete="CASCADE"),
nullable=True,
)
data_transfer_package = relationship(
"DataTransferPackage", back_populates="raw_data_packages"
)
long_term_archive_transfers = relationship(
"LongTermArchiveTransfer",
back_populates="raw_data_package",
cascade="all, delete-orphan",
)
analyze_status = Column(
SQLAlchemyEnum(Status, values_callable=lambda x: [e.value for e in x]), nullable=False, default=Status.PENDING
)
physical_copies = relationship(
"RawDataPackagePhysicalCopy",
back_populates="raw_data_package",
cascade="all, delete-orphan",
)
package_metadata = relationship(
"RawDataPackageMetadata",
back_populates="raw_data_package",
uselist=False,
)
state = Column(
SQLAlchemyEnum(PackageState, values_callable=lambda x: [e.value for e in x]),
nullable=True,
default=PackageState.WAITING,
)
staging_jobs = relationship(
"StagingJob",
secondary=raw_data_package_staging_job_association,
back_populates="raw_data_packages",
)
[docs]
class DataTransferPackage(Base):
__tablename__ = "data_transfer_package"
id = Column(Integer, primary_key=True)
hash_id = Column(String(250), nullable=False)
file_name = Column(String(250), nullable=False)
size = Column(BigInteger, nullable=True)
checksum = Column(String(250), nullable=True)
relative_path = Column(String(250), nullable=False)
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=False)
origin_location = relationship("DataLocation", foreign_keys=[origin_location_id])
raw_data_files = relationship(
"RawDataFile",
back_populates="data_transfer_package",
)
raw_data_packages = relationship(
"RawDataPackage",
back_populates="data_transfer_package",
)
data_transfers = relationship(
"DataTransfer", back_populates="data_transfer_package"
)
physical_copies = relationship(
"DataTransferPackagePhysicalCopy",
back_populates="data_transfer_package",
cascade="all, delete-orphan",
)
[docs]
class DataTransfer(Base):
"""Tracks data transfers between locations."""
__tablename__ = "data_transfer"
id = Column(Integer, primary_key=True)
process_id = Column(String(250))
start_time = Column(DateTime)
end_time = Column(DateTime)
data_transfer_method = Column(String(250), default="bbcp")
transfer_program_log = Column(Text)
# Location references
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=False)
origin_location = relationship("DataLocation", foreign_keys=[origin_location_id])
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=False
)
destination_location = relationship(
"DataLocation", foreign_keys=[destination_location_id]
)
# Package reference
data_transfer_package_id = Column(
Integer, ForeignKey("data_transfer_package.id"), nullable=False
)
data_transfer_package = relationship(
"DataTransferPackage", back_populates="data_transfers"
)
unpack_start_time = Column(DateTime)
unpack_end_time = Column(DateTime)
unpack_log = Column(Text)
logs = relationship("DataTransferLog", back_populates="data_transfer")
[docs]
class SystemLog(Base):
__tablename__ = "system_log"
id = Column(Integer, primary_key=True)
type = Column(String(250), nullable=False)
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "system_log",
}
[docs]
class DataTransferLog(SystemLog):
"""Log entries for data transfers with references to log files.
This model implements a lightweight approach where:
- Basic status info and log file path are stored in the database
- Full command outputs are stored in files
- Detailed metrics are stored in InfluxDB
"""
__tablename__ = "data_transfer_log"
__mapper_args__ = {
"polymorphic_identity": "data_transfer_log",
}
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
data_transfer_id = Column(Integer, ForeignKey("data_transfer.id"), nullable=False)
data_transfer = relationship("DataTransfer", back_populates="logs")
# Basic info
timestamp = Column(DateTime, nullable=False)
status = Column(
String(50),
nullable=False,
doc="Status of this transfer attempt (success/failure)",
)
# Path to the log file
log_path = Column(
String(500), nullable=False, doc="Path to file containing full command output"
)
@property
def content(self) -> Optional[str]:
"""Read and return the full log content if file exists."""
if self.log_path and os.path.exists(self.log_path):
with open(self.log_path, "r") as f:
return f.read()
return None
# def __repr__(self):
# return (
# f"<DataTransferLog(id={self.id}, "
# f"data_transfer_id={self.data_transfer_id}, "
# f"timestamp={self.timestamp}, "
# f"status={self.status})>"
# )
[docs]
class OperationFailureEvent(SystemLog):
"""Append-only failure-history record for any pipeline operation (#85).
One row per failure occurrence (retryable and permanent), never erased by a
reset/retry — the durable trail behind the denormalized per-row
``error_context`` cache.
Generic across operation types, so it keys on ``(operation_type,
operation_id)`` — the pair the system already correlates on (Celery task
state, recovery handlers, circuit breaker, routing) — and carries NO foreign
key to the operation row (it can't reference one of several tables).
Precedent for the bare integer id: ``PhysicalCopy.deletion_task_id``.
"""
__tablename__ = "operation_failure_event"
__mapper_args__ = {
"polymorphic_identity": "operation_failure_event",
}
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
operation_type = Column(String(50), nullable=False)
operation_id = Column(Integer, nullable=False)
error_type = Column(String(250), nullable=True)
error_context = Column(JSONB_VARIANT, nullable=True)
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
__table_args__ = (
Index(
"ix_operation_failure_event_operation",
"operation_type",
"operation_id",
),
)
[docs]
class DataArchive(Base):
__tablename__ = "data_archive"
id = Column(Integer, primary_key=True)
name = Column(String(250), nullable=False)
[docs]
class PhysicalCopy(Base):
"""Base class for tracking physical copies of files across different storage locations.
This class implements a polymorphic pattern to track physical copies of different types
of files (RawDataFile, RawDataPackage, DataTransferPackage) across different storage
locations and types. Each physical copy represents an actual file on disk or other
storage medium.
"""
__tablename__ = "physical_copy"
id = Column(Integer, primary_key=True)
type = Column(String(50), nullable=False)
# Location reference
data_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=False)
data_location = relationship("DataLocation")
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
verified_at = Column(DateTime(timezone=True), nullable=True)
status = Column(
SQLAlchemyEnum(PhysicalCopyStatus, values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=PhysicalCopyStatus.PRESENT,
)
checksum = Column(String(250), nullable=True)
deletion_task_id = Column(Integer, nullable=True)
deleted_at = Column(DateTime(timezone=True), nullable=True)
# Copy-anchored lineage back-references (ops-db #94): the operations that
# consumed this copy as input and that produced it as output.
consumed_by_operations = relationship(
"Operation",
secondary=operation_consumes_physical_copy,
back_populates="consumed_copies",
)
produced_by_operations = relationship(
"Operation",
secondary=operation_produces_physical_copy,
back_populates="produced_copies",
)
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "physical_copy",
}
[docs]
class RawDataFilePhysicalCopy(PhysicalCopy):
"""Tracks physical copies of individual raw data files."""
__tablename__ = "raw_data_file_physical_copy"
__mapper_args__ = {"polymorphic_identity": "raw_data_file_physical_copy"}
id = Column(Integer, ForeignKey("physical_copy.id"), primary_key=True)
raw_data_file_id = Column(
UUID(as_uuid=True), ForeignKey("raw_data_file.id"), nullable=False
)
raw_data_file = relationship("RawDataFile", back_populates="physical_copies")
@property
def full_path(self):
"""Get the full path/S3 key for this physical copy."""
if isinstance(self.data_location, DiskDataLocation):
return os.path.join(
self.data_location.path, self.raw_data_file.relative_path
)
elif isinstance(self.data_location, S3DataLocation):
# For S3, return just the relative path - the actual S3 key construction
# should be handled by the data-transfer package using the shared utility functions
return self.raw_data_file.relative_path
elif isinstance(self.data_location, TapeDataLocation):
return os.path.join(
self.data_location.mount_path, self.raw_data_file.relative_path
)
return self.raw_data_file.relative_path
[docs]
class RawDataPackagePhysicalCopy(PhysicalCopy):
"""Tracks physical copies of raw data packages."""
__tablename__ = "raw_data_package_physical_copy"
__mapper_args__ = {"polymorphic_identity": "raw_data_package_physical_copy"}
id = Column(Integer, ForeignKey("physical_copy.id"), primary_key=True)
raw_data_package_id = Column(
Integer, ForeignKey("raw_data_package.id", ondelete="CASCADE"), nullable=False
)
raw_data_package = relationship("RawDataPackage", back_populates="physical_copies")
@property
def full_path(self):
"""Get the full path/S3 key for this physical copy."""
if isinstance(self.data_location, DiskDataLocation):
return os.path.join(
self.data_location.path, self.raw_data_package.relative_path
)
elif isinstance(self.data_location, S3DataLocation):
# For S3, return just the relative path - the actual S3 key construction
# should be handled by the data-transfer package using the shared utility functions
return self.raw_data_package.relative_path
elif isinstance(self.data_location, TapeDataLocation):
return os.path.join(
self.data_location.mount_path, self.raw_data_package.relative_path
)
return self.raw_data_package.relative_path
[docs]
class DataTransferPackagePhysicalCopy(PhysicalCopy):
"""Tracks physical copies of data transfer packages."""
__tablename__ = "data_transfer_package_physical_copy"
__mapper_args__ = {"polymorphic_identity": "data_transfer_package_physical_copy"}
id = Column(Integer, ForeignKey("physical_copy.id"), primary_key=True)
data_transfer_package_id = Column(
Integer,
ForeignKey("data_transfer_package.id", ondelete="CASCADE"),
nullable=False,
)
data_transfer_package = relationship(
"DataTransferPackage", back_populates="physical_copies"
)
@property
def full_path(self):
"""Get the full path/S3 key for this physical copy."""
if isinstance(self.data_location, DiskDataLocation):
return os.path.join(
self.data_location.path, self.data_transfer_package.relative_path
)
elif isinstance(self.data_location, S3DataLocation):
# For S3, return just the relative path - the actual S3 key construction
# should be handled by the data-transfer package using the shared utility functions
return self.data_transfer_package.relative_path
elif isinstance(self.data_location, TapeDataLocation):
return os.path.join(
self.data_location.mount_path, self.data_transfer_package.relative_path
)
return self.data_transfer_package.relative_path
[docs]
class LongTermArchiveTransfer(Base):
__tablename__ = "long_term_archive_transfer"
id = Column(Integer, primary_key=True)
site_id = Column(Integer, ForeignKey("site.id"))
site = relationship("Site", back_populates="long_term_archive_transfers")
origin_data_location_id = Column(Integer, ForeignKey("data_location.id"))
origin_data_location = relationship(
"DataLocation", foreign_keys=[origin_data_location_id]
)
destination_data_location_id = Column(Integer, ForeignKey("data_location.id"))
destination_data_location = relationship(
"DataLocation", foreign_keys=[destination_data_location_id]
)
raw_data_package_id = Column(Integer, ForeignKey("raw_data_package.id"))
raw_data_package = relationship(
"RawDataPackage", back_populates="long_term_archive_transfers"
)
# Store logs directly in the LongTermArchiveTransfer model as a text field
logs = relationship(
"LongTermArchiveTransferLog", back_populates="long_term_archive_transfer"
)
start_time = Column(DateTime, nullable=True)
end_time = Column(DateTime, nullable=True)
# Add fields for tracking attempts
last_attempt_time = Column(DateTime, nullable=True)
# Add a field to store error messages
error_message = Column(Text, nullable=True)
__table_args__ = (
UniqueConstraint(
"raw_data_package_id",
"origin_data_location_id",
"destination_data_location_id",
name="uix_long_term_archive_transfer_location",
),
)
[docs]
class LongTermArchiveTransferLog(SystemLog):
__tablename__ = "long_term_archive_transfer_log"
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
long_term_archive_transfer_id = Column(
Integer, ForeignKey("long_term_archive_transfer.id"), nullable=False
)
long_term_archive_transfer = relationship(
"LongTermArchiveTransfer", back_populates="logs"
)
log = Column(Text, nullable=False)
timestamp = Column(DateTime, nullable=False)
__mapper_args__ = {
"polymorphic_identity": "long_term_archive_transfer_log",
}
[docs]
class StagingJob(Base):
__tablename__ = "staging_job"
id = Column(Integer, primary_key=True)
status = Column(SQLAlchemyEnum(Status, values_callable=lambda x: [e.value for e in x]), nullable=False, default=Status.PENDING)
failure_error_message = Column(Text, nullable=True)
start_time = Column(DateTime, nullable=True)
active = Column(Boolean, nullable=False, default=True)
end_time = Column(DateTime, nullable=True)
retry_count = Column(Integer, nullable=False, default=0)
raw_data_packages = relationship(
"RawDataPackage",
secondary=raw_data_package_staging_job_association,
back_populates="staging_jobs",
)
logs = relationship("StagingJobLog", back_populates="staging_job")
origin_data_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=False
)
origin_data_location = relationship(
"DataLocation", foreign_keys=[origin_data_location_id]
)
destination_data_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=False
)
destination_data_location = relationship(
"DataLocation", foreign_keys=[destination_data_location_id]
)
[docs]
class StagingJobLog(SystemLog):
__tablename__ = "staging_job_log"
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
staging_job_id = Column(Integer, ForeignKey("staging_job.id"), nullable=False)
staging_job = relationship("StagingJob", back_populates="logs")
log = Column(Text, nullable=False)
timestamp = Column(DateTime, nullable=False)
__mapper_args__ = {
"polymorphic_identity": "staging_job_log",
}
# ---------------------------------------------------------------------------
# Uniform Operation model (ops-db #94, ADR-0003)
#
# Every pipeline stage becomes a uniform, independently addressable operation:
# one polymorphic Operation base (joined-table inheritance, same pattern as
# SystemLog / PhysicalCopy) plus six stage subclasses. This is ADDITIVE -- the
# legacy status/counter columns on RawDataPackage, DataTransferPackage,
# DataTransfer, and LongTermArchiveTransfer are retained so the current chain
# keeps working until a later cleanup migrates onto these tables.
# ---------------------------------------------------------------------------
[docs]
class OperationGroup(Base):
"""Polymorphic parent that bundles N co-created operations (ADR-0003).
Its status is DERIVED from its child operations and is never set
independently -- modelled as a read-only Python property, not a mapped
column, so the derivation rule is the single source of truth and the group
cannot drift from its children. Built minimal/claim-free here; the retention
claim lives on the StagingOperationGroup specialization.
"""
__tablename__ = "operation_group"
id = Column(Integer, primary_key=True)
type = Column(String(50), nullable=False)
created_at = Column(
DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc),
nullable=False,
)
operations = relationship("Operation", back_populates="operation_group")
__mapper_args__ = {
"polymorphic_on": type,
"polymorphic_identity": "operation_group",
}
@property
def status(self) -> "Status":
"""Derive the group status from the child operations.
Rule (in precedence order):
- any child FAILED -> FAILED
- all children COMPLETED -> COMPLETED
- any child past PENDING -> IN_PROGRESS (work is in flight)
- otherwise (empty, or all PENDING/SCHEDULED) -> PENDING
An empty group has nothing to derive from and reads as PENDING
(not-yet-started), which is also the safe default before children are
attached. SCHEDULED is folded into the pre-start bucket: a group with
only SCHEDULED children has dispatched nothing yet, so it is not in
flight; the moment any child reaches IN_PROGRESS (or beyond, while
others lag) the group reads IN_PROGRESS.
"""
children = self.operations
if not children:
return Status.PENDING
statuses = [op.status for op in children]
if any(s == Status.FAILED for s in statuses):
return Status.FAILED
if all(s == Status.COMPLETED for s in statuses):
return Status.COMPLETED
if all(s in (Status.PENDING, Status.SCHEDULED) for s in statuses):
return Status.PENDING
return Status.IN_PROGRESS
[docs]
class StagingOperationGroup(OperationGroup):
"""Staging specialization of OperationGroup carrying the retention claim.
This is the uniform-model successor to the legacy ``StagingJob`` (which is
retained, untouched, during the additive transition). ``active`` is the
retention claim that holds the staged packages until released -- the
deletion manager treats an active staging group as a hold on its copies.
The job's status is still derived from its child staging operations on the
OperationGroup base; only the claim is added here.
"""
__tablename__ = "staging_operation_group"
id = Column(Integer, ForeignKey("operation_group.id"), primary_key=True)
active = Column(Boolean, nullable=False, default=True)
__mapper_args__ = {
"polymorphic_identity": "staging_operation_group",
}
[docs]
class Operation(Base):
"""Polymorphic base for every pipeline operation (joined-table inheritance).
Uniform fields shared by all stages live here; stage-specific fields live on
the subclasses. ``operation_kind`` is the polymorphic discriminator and
holds an ``OperationKind`` value (== the frozen breadcrumb string).
"""
__tablename__ = "operation"
id = Column(Integer, primary_key=True)
# Discriminator / identity axis -- the OperationKind value (frozen string).
operation_kind = Column(
SQLAlchemyEnum(OperationKind, values_callable=lambda x: [e.value for e in x]),
nullable=False,
)
# Uniform operation fields (one canonical state machine + one retry counter).
status = Column(
SQLAlchemyEnum(Status, values_callable=lambda x: [e.value for e in x]),
nullable=False,
default=Status.PENDING,
)
retry_count = Column(Integer, nullable=False, default=0)
failure_error_message = Column(Text, nullable=True)
# Structured latest-failure breadcrumb; JSONB_VARIANT so SELECT DISTINCT over
# an operation row carries an equality operator on Postgres (ADR-0002).
error_context = Column(JSONB_VARIANT, nullable=True)
start_time = Column(DateTime(timezone=True), nullable=True)
end_time = Column(DateTime(timezone=True), nullable=True)
# Optional membership in a batched group (status derived there).
operation_group_id = Column(
Integer, ForeignKey("operation_group.id"), nullable=True
)
operation_group = relationship("OperationGroup", back_populates="operations")
# Copy-anchored consume/produce lineage (ADR-0003). Identical contract for
# every kind: an operation consumes input copies and produces output copies.
consumed_copies = relationship(
"PhysicalCopy",
secondary=operation_consumes_physical_copy,
back_populates="consumed_by_operations",
)
produced_copies = relationship(
"PhysicalCopy",
secondary=operation_produces_physical_copy,
back_populates="produced_by_operations",
)
__mapper_args__ = {
"polymorphic_on": operation_kind,
# No base polymorphic_identity: Operation is abstract in practice -- every
# row is one of the six concrete kinds (mirrors DataLocation's None base).
"polymorphic_identity": None,
}
[docs]
class PackagingOperation(Operation):
"""Packaging stage: assembles raw data files into a RawDataPackage."""
__tablename__ = "packaging_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
raw_data_package_id = Column(
Integer, ForeignKey("raw_data_package.id", ondelete="CASCADE"), nullable=True
)
__mapper_args__ = {"polymorphic_identity": OperationKind.PACKAGING}
[docs]
class BundlingOperation(Operation):
"""Bundling stage: assembles packages into a DataTransferPackage."""
__tablename__ = "bundling_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
data_transfer_package_id = Column(
Integer,
ForeignKey("data_transfer_package.id", ondelete="CASCADE"),
nullable=True,
)
__mapper_args__ = {"polymorphic_identity": OperationKind.BUNDLING}
[docs]
class TransferOperation(Operation):
"""Transfer stage: moves a DataTransferPackage between locations.
Sibling of UnpackOperation under the same DataTransferPackage (ADR-0003) --
the transfer/unpack unicorn record splits into two independent rows.
"""
__tablename__ = "transfer_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
data_transfer_package_id = Column(
Integer,
ForeignKey("data_transfer_package.id", ondelete="CASCADE"),
nullable=True,
)
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=True)
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=True
)
transfer_method = Column(String(250), nullable=True, default="bbcp")
__mapper_args__ = {"polymorphic_identity": OperationKind.TRANSFER}
[docs]
class UnpackOperation(Operation):
"""Unpack stage: verifies and unpacks a transferred DataTransferPackage.
Sibling of TransferOperation under the same DataTransferPackage (ADR-0003).
Anchored per (data_transfer_package_id, destination_location_id): a package
transferred to several destinations is unpacked/verified independently at
each, so there is one UnpackOperation per destination, mirroring the sibling
TransferOperation's destination_location_id. This makes the transfer+unpack
pair 1:1 per destination. destination_location_id is nullable for the
additive transition (no backfill).
"""
__tablename__ = "unpack_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
data_transfer_package_id = Column(
Integer,
ForeignKey("data_transfer_package.id", ondelete="CASCADE"),
nullable=True,
)
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=True
)
__mapper_args__ = {"polymorphic_identity": OperationKind.UNPACK}
[docs]
class ArchiveOperation(Operation):
"""Archive stage: copies a RawDataPackage to long-term storage."""
__tablename__ = "archive_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
raw_data_package_id = Column(
Integer, ForeignKey("raw_data_package.id"), nullable=True
)
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=True)
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=True
)
__mapper_args__ = {"polymorphic_identity": OperationKind.ARCHIVE}
[docs]
class StagingOperation(Operation):
"""Staging stage: downloads + unpacks a package into a processing area.
Typically grouped under a StagingOperationGroup (the retention-claim parent).
"""
__tablename__ = "staging_operation"
id = Column(Integer, ForeignKey("operation.id"), primary_key=True)
raw_data_package_id = Column(
Integer, ForeignKey("raw_data_package.id"), nullable=True
)
origin_location_id = Column(Integer, ForeignKey("data_location.id"), nullable=True)
destination_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=True
)
__mapper_args__ = {"polymorphic_identity": OperationKind.STAGING}
# ---------------------------------------------------------------------------
# Workflow Manager Models
# ---------------------------------------------------------------------------
# Association tables for workflow manager
executed_step_input_packages = Table(
"executed_step_input_packages",
Base.metadata,
Column("executed_reduction_step_id", Integer, ForeignKey("executed_reduction_step.id")),
Column("raw_data_package_id", Integer, ForeignKey("raw_data_package.id")),
Column("role", String, default="science"),
)
executed_step_input_products = Table(
"executed_step_input_products",
Base.metadata,
Column("executed_reduction_step_id", Integer, ForeignKey("executed_reduction_step.id")),
Column("data_product_id", Integer, ForeignKey("data_product.id")),
Column("role", String, default="science"),
)
data_product_lineage = Table(
"data_product_lineage",
Base.metadata,
Column("data_product_id", Integer, ForeignKey("data_product.id")),
Column("raw_data_package_id", Integer, ForeignKey("raw_data_package.id")),
)
reduction_step_dependency = Table(
"reduction_step_dependency",
Base.metadata,
Column(
"upstream_step_id",
Integer,
ForeignKey("reduction_step.id"),
primary_key=True,
),
Column(
"downstream_step_id",
Integer,
ForeignKey("reduction_step.id"),
primary_key=True,
),
)
[docs]
class ReductionSoftware(Base):
__tablename__ = "reduction_software"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True, nullable=False)
description = Column(Text)
ghcr_image_url = Column(String)
github_repo_url = Column(String)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
active = Column(Boolean, default=True)
versions = relationship(
"ReductionSoftwareVersion", back_populates="reduction_software"
)
reduction_steps = relationship(
"ReductionStep", back_populates="reduction_software"
)
[docs]
class ReductionSoftwareVersion(Base):
__tablename__ = "reduction_software_version"
id = Column(Integer, primary_key=True)
reduction_software_id = Column(
Integer, ForeignKey("reduction_software.id"), nullable=False
)
version_tag = Column(String, nullable=False)
image_digest = Column(String)
build_date = Column(DateTime)
changelog = Column(Text)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
is_latest = Column(Boolean, default=False)
reduction_software = relationship(
"ReductionSoftware", back_populates="versions"
)
executed_steps = relationship(
"ExecutedReductionStep", back_populates="reduction_software_version"
)
__table_args__ = (
UniqueConstraint("reduction_software_id", "version_tag"),
)
[docs]
class ReductionStepConfig(Base):
__tablename__ = "reduction_step_config"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
version = Column(String)
description = Column(Text)
config_parameters = Column(JSONB_VARIANT)
environment_variables = Column(JSONB_VARIANT)
command_template = Column(String)
resource_requirements = Column(JSONB_VARIANT)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
reduction_steps = relationship(
"ReductionStep", back_populates="reduction_step_config"
)
[docs]
class DataGrouping(Base):
__tablename__ = "data_grouping"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
description = Column(Text)
instrument_module_id = Column(
Integer, ForeignKey("instrument_module.id"), nullable=True
)
filter_rules = Column(JSONB_VARIANT)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
active = Column(Boolean, default=True)
instrument_module = relationship("InstrumentModule")
pipelines = relationship("Pipeline", back_populates="data_grouping")
[docs]
class Pipeline(Base):
__tablename__ = "pipeline"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
description = Column(Text)
data_grouping_id = Column(
Integer, ForeignKey("data_grouping.id"), nullable=False
)
processing_location_id = Column(
Integer, ForeignKey("data_location.id"), nullable=False
)
trigger_type = Column(
SQLAlchemyEnum(TriggerType), default=TriggerType.MANUAL
)
trigger_config = Column(JSONB_VARIANT)
enabled = Column(Boolean, default=True)
priority = Column(Integer, default=0)
created_by_user_id = Column(Integer, nullable=True)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
updated_at = Column(DateTime, onupdate=lambda: datetime.now(timezone.utc))
data_grouping = relationship("DataGrouping", back_populates="pipelines")
processing_location = relationship("DataLocation")
reduction_steps = relationship(
"ReductionStep",
back_populates="pipeline",
order_by="ReductionStep.step_order",
)
[docs]
class ReductionStep(Base):
__tablename__ = "reduction_step"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
description = Column(Text)
pipeline_id = Column(
Integer, ForeignKey("pipeline.id"), nullable=False
)
step_order = Column(Integer, nullable=False)
reduction_software_id = Column(
Integer, ForeignKey("reduction_software.id"), nullable=False
)
pinned_version_id = Column(
Integer, ForeignKey("reduction_software_version.id"), nullable=True
)
reduction_step_config_id = Column(
Integer, ForeignKey("reduction_step_config.id"), nullable=False
)
group_by = Column(JSONB_VARIANT)
cooldown_seconds = Column(Integer, default=0)
schedule = Column(String, nullable=True)
max_concurrent_runs = Column(Integer, default=1)
version_policy = Column(String, default="compatible")
max_retries = Column(Integer, default=3)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
pipeline = relationship("Pipeline", back_populates="reduction_steps")
reduction_software = relationship(
"ReductionSoftware", back_populates="reduction_steps"
)
pinned_version = relationship("ReductionSoftwareVersion")
reduction_step_config = relationship(
"ReductionStepConfig", back_populates="reduction_steps"
)
executed_steps = relationship(
"ExecutedReductionStep", back_populates="reduction_step"
)
upstream_dependencies = relationship(
"ReductionStep",
secondary=reduction_step_dependency,
primaryjoin=id == reduction_step_dependency.c.downstream_step_id,
secondaryjoin=id == reduction_step_dependency.c.upstream_step_id,
backref="downstream_dependencies",
)
[docs]
class ExecutedReductionStep(Base):
__tablename__ = "executed_reduction_step"
id = Column(Integer, primary_key=True)
reduction_step_id = Column(
Integer, ForeignKey("reduction_step.id"), nullable=False
)
reduction_software_version_id = Column(
Integer,
ForeignKey("reduction_software_version.id"),
nullable=False,
)
status = Column(
SQLAlchemyEnum(RunStatus),
default=RunStatus.PENDING,
nullable=False,
)
sub_group_key = Column(String)
sub_group_metadata = Column(JSONB_VARIANT)
execution_command = Column(Text)
hpc_job_id = Column(String)
hpc_queue = Column(String)
staging_job_id = Column(
Integer, ForeignKey("staging_job.id"), nullable=True
)
start_time = Column(DateTime)
end_time = Column(DateTime)
processing_time_s = Column(Float, nullable=True)
peak_memory_gb = Column(Float, nullable=True)
cpu_hours = Column(Float, nullable=True)
logs = Column(Text)
failure_error_message = Column(Text)
retry_count = Column(Integer, default=0)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
trigger_reason = Column(String)
__table_args__ = (
Index("ix_executed_reduction_step_status", "status"),
Index("ix_executed_reduction_step_step_id", "reduction_step_id"),
Index("ix_executed_reduction_step_sub_group", "sub_group_key"),
)
reduction_step = relationship(
"ReductionStep", back_populates="executed_steps"
)
reduction_software_version = relationship(
"ReductionSoftwareVersion", back_populates="executed_steps"
)
staging_job = relationship("StagingJob")
input_packages = relationship(
"RawDataPackage", secondary=executed_step_input_packages
)
input_products = relationship(
"DataProduct",
secondary=executed_step_input_products,
back_populates="consumed_by_steps",
)
data_products = relationship(
"DataProduct", back_populates="executed_reduction_step"
)
step_logs = relationship(
"ExecutedReductionStepLog", back_populates="executed_reduction_step"
)
[docs]
class DataProduct(Base):
__tablename__ = "data_product"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
relative_path = Column(String)
size = Column(BigInteger)
checksum = Column(String)
file_type = Column(String)
product_type = Column(
SQLAlchemyEnum(DataProductType), default=DataProductType.SCIENCE
)
executed_reduction_step_id = Column(
Integer, ForeignKey("executed_reduction_step.id"), nullable=False
)
data_grouping_id = Column(
Integer, ForeignKey("data_grouping.id"), nullable=True
)
quality_metrics = Column(JSONB_VARIANT)
metadata_ = Column("metadata", JSONB_VARIANT)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
executed_reduction_step = relationship(
"ExecutedReductionStep", back_populates="data_products"
)
data_grouping = relationship("DataGrouping")
raw_data_lineage = relationship(
"RawDataPackage", secondary=data_product_lineage
)
physical_copies = relationship(
"DataProductPhysicalCopy",
back_populates="data_product",
cascade="all, delete-orphan",
)
consumed_by_steps = relationship(
"ExecutedReductionStep",
secondary=executed_step_input_products,
back_populates="input_products",
)
[docs]
class DataProductPhysicalCopy(PhysicalCopy):
__tablename__ = "data_product_physical_copy"
id = Column(Integer, ForeignKey("physical_copy.id"), primary_key=True)
data_product_id = Column(
Integer, ForeignKey("data_product.id"), nullable=False
)
data_product = relationship(
"DataProduct", back_populates="physical_copies"
)
__mapper_args__ = {
"polymorphic_identity": "data_product_physical_copy",
}
@property
def full_path(self):
if isinstance(self.data_location, DiskDataLocation):
return f"{self.data_location.path}/{self.data_product.relative_path}"
elif isinstance(self.data_location, S3DataLocation):
return f"s3://{self.data_location.bucket_name}/{self.data_product.relative_path}"
return None
[docs]
class ExecutedReductionStepLog(SystemLog):
__tablename__ = "executed_reduction_step_log"
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
executed_reduction_step_id = Column(
Integer, ForeignKey("executed_reduction_step.id"), nullable=False
)
executed_reduction_step = relationship(
"ExecutedReductionStep", back_populates="step_logs"
)
log = Column(Text, nullable=False)
timestamp = Column(DateTime, nullable=False)
__mapper_args__ = {
"polymorphic_identity": "executed_reduction_step_log",
}
# ---------------------------------------------------------------------------
# Operational Config Models
# ---------------------------------------------------------------------------
[docs]
class SystemSettings(Base):
"""Key-value store for operational configuration settings.
Used by data-transfer SettingsManager to read runtime-tunable parameters
(poll intervals, buffer thresholds, transfer limits, etc.).
"""
__tablename__ = "system_settings"
id = Column(Integer, primary_key=True)
key = Column(String(250), unique=True, index=True, nullable=False)
value = Column(JSONB_VARIANT, nullable=False)
description = Column(Text, nullable=True)
updated_by = Column(String(150), nullable=True)
updated_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
created_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
[docs]
class OperationalConfigLog(SystemLog):
"""Audit log for operational configuration changes."""
__tablename__ = "operational_config_log"
id = Column(Integer, ForeignKey("system_log.id"), primary_key=True)
setting_key = Column(String(250), nullable=False, index=True)
old_value = Column(JSONB_VARIANT, nullable=True)
new_value = Column(JSONB_VARIANT, nullable=True)
action = Column(String(50), nullable=False)
changed_by = Column(String(150), nullable=True)
changed_at = Column(
DateTime, default=lambda: datetime.now(timezone.utc), nullable=False
)
__mapper_args__ = {
"polymorphic_identity": "operational_config",
}
