This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.model.migrate.versions.0060_history_archive_import

Migration script to create column and table for importing histories from
file archives.

import logging

from sqlalchemy import Boolean, Column, ForeignKey, Integer, MetaData, Table, TEXT

from galaxy.model.migrate.versions.util import engine_false

log = logging.getLogger(__name__)
metadata = MetaData()

# Columns to add.

importing_col = Column("importing", Boolean, index=True, default=False)
ldda_parent_col = Column("ldda_parent_id", Integer, ForeignKey("library_dataset_dataset_association.id"), index=True)

# Table to add.

JobImportHistoryArchive_table = Table("job_import_history_archive", metadata,
                                      Column("id", Integer, primary_key=True),
                                      Column("job_id", Integer, ForeignKey("job.id"), index=True),
                                      Column("history_id", Integer, ForeignKey("history.id"), index=True),
                                      Column("archive_dir", TEXT))

[docs]def upgrade(migrate_engine): print(__doc__) metadata.bind = migrate_engine metadata.reflect() # Add column to history table and initialize. try: History_table = Table("history", metadata, autoload=True) importing_col.create(History_table, index_name="ix_history_importing") assert importing_col is History_table.c.importing # Initialize column to false. migrate_engine.execute(f"UPDATE history SET importing={engine_false(migrate_engine)}") except Exception: log.exception("Adding column 'importing' to history table failed.") # Create job_import_history_archive table. try: JobImportHistoryArchive_table.create() except Exception: log.exception("Creating job_import_history_archive table failed.")
[docs]def downgrade(migrate_engine): metadata.bind = migrate_engine metadata.reflect() # Drop 'importing' column from history table. try: History_table = Table("history", metadata, autoload=True) importing_col = History_table.c.importing importing_col.drop() except Exception: log.exception("Dropping column 'importing' from history table failed.") # Drop job_import_history_archive table. try: JobImportHistoryArchive_table.drop() except Exception: log.exception("Dropping job_import_history_archive table failed.")