@@ -464,6 +464,141 @@ async def _migrate_add_admin_flag(conn):
464464 _log .info ("Migration v4_admin_flag complete" )
465465
466466
467+ async def _migrate_add_job_name (conn ):
468+ """v5: Add name column to jobs table. Idempotent."""
469+ try :
470+ r = await conn .execute (
471+ text ("SELECT 1 FROM schema_versions WHERE version = 'v5_job_name'" )
472+ )
473+ if r .scalar ():
474+ return
475+ except Exception :
476+ pass
477+
478+ _log .info ("Migration v5: job name — starting" )
479+
480+ try :
481+ r = await conn .execute (text (
482+ "SELECT column_name FROM information_schema.columns "
483+ "WHERE table_name = 'jobs' AND column_name = 'name'"
484+ ))
485+ if not r .scalar ():
486+ await conn .execute (text (
487+ "ALTER TABLE jobs ADD COLUMN name VARCHAR(255)"
488+ ))
489+ _log .info (" Added jobs.name column" )
490+ except Exception :
491+ pass
492+
493+ await conn .execute (text (
494+ "INSERT INTO schema_versions (version, applied_at) "
495+ "VALUES ('v5_job_name', CURRENT_TIMESTAMP) ON CONFLICT DO NOTHING"
496+ ))
497+ _log .info ("Migration v5_job_name complete" )
498+
499+
500+ async def _migrate_add_job_user_id (conn ):
501+ """v6: Add user_id FK to jobs table for tracking who launched each job."""
502+ try :
503+ r = await conn .execute (
504+ text ("SELECT 1 FROM schema_versions WHERE version = 'v6_job_user_id'" )
505+ )
506+ if r .scalar ():
507+ return
508+ except Exception :
509+ pass
510+
511+ _log .info ("Migration v6: job user_id — starting" )
512+
513+ try :
514+ r = await conn .execute (text (
515+ "SELECT column_name FROM information_schema.columns "
516+ "WHERE table_name = 'jobs' AND column_name = 'user_id'"
517+ ))
518+ if not r .scalar ():
519+ await conn .execute (text (
520+ "ALTER TABLE jobs ADD COLUMN user_id VARCHAR(12) REFERENCES users(id) ON DELETE SET NULL"
521+ ))
522+ _log .info (" Added jobs.user_id column" )
523+ except Exception :
524+ pass
525+
526+ await conn .execute (text (
527+ "INSERT INTO schema_versions (version, applied_at) "
528+ "VALUES ('v6_job_user_id', CURRENT_TIMESTAMP) ON CONFLICT DO NOTHING"
529+ ))
530+ _log .info ("Migration v6_job_user_id complete" )
531+
532+
533+ async def _migrate_add_job_config_hash (conn ):
534+ """v7: Add config_hash column to jobs for duplicate detection."""
535+ try :
536+ r = await conn .execute (
537+ text ("SELECT 1 FROM schema_versions WHERE version = 'v7_job_config_hash'" )
538+ )
539+ if r .scalar ():
540+ return
541+ except Exception :
542+ pass
543+
544+ _log .info ("Migration v7: job config_hash — starting" )
545+
546+ try :
547+ r = await conn .execute (text (
548+ "SELECT column_name FROM information_schema.columns "
549+ "WHERE table_name = 'jobs' AND column_name = 'config_hash'"
550+ ))
551+ if not r .scalar ():
552+ await conn .execute (text (
553+ "ALTER TABLE jobs ADD COLUMN config_hash VARCHAR(32)"
554+ ))
555+ await conn .execute (text (
556+ "CREATE INDEX IF NOT EXISTS ix_jobs_config_hash ON jobs (config_hash)"
557+ ))
558+ _log .info (" Added jobs.config_hash column with index" )
559+ except Exception :
560+ pass
561+
562+ await conn .execute (text (
563+ "INSERT INTO schema_versions (version, applied_at) "
564+ "VALUES ('v7_job_config_hash', CURRENT_TIMESTAMP) ON CONFLICT DO NOTHING"
565+ ))
566+ _log .info ("Migration v7_job_config_hash complete" )
567+
568+
569+ async def _migrate_add_job_disk_size (conn ):
570+ """v8: Add disk_size_bytes column to jobs for cached disk usage."""
571+ try :
572+ r = await conn .execute (
573+ text ("SELECT 1 FROM schema_versions WHERE version = 'v8_job_disk_size'" )
574+ )
575+ if r .scalar ():
576+ return
577+ except Exception :
578+ pass
579+
580+ _log .info ("Migration v8: job disk_size_bytes — starting" )
581+
582+ try :
583+ r = await conn .execute (text (
584+ "SELECT column_name FROM information_schema.columns "
585+ "WHERE table_name = 'jobs' AND column_name = 'disk_size_bytes'"
586+ ))
587+ if not r .scalar ():
588+ await conn .execute (text (
589+ "ALTER TABLE jobs ADD COLUMN disk_size_bytes BIGINT"
590+ ))
591+ _log .info (" Added jobs.disk_size_bytes column" )
592+ except Exception :
593+ pass
594+
595+ await conn .execute (text (
596+ "INSERT INTO schema_versions (version, applied_at) "
597+ "VALUES ('v8_job_disk_size', CURRENT_TIMESTAMP) ON CONFLICT DO NOTHING"
598+ ))
599+ _log .info ("Migration v8_job_disk_size complete" )
600+
601+
467602@asynccontextmanager
468603async def lifespan (app : FastAPI ):
469604 """Startup/shutdown events."""
@@ -478,6 +613,14 @@ async def lifespan(app: FastAPI):
478613 await _migrate_to_composite_datasets (conn )
479614 async with engine .begin () as conn :
480615 await _migrate_add_admin_flag (conn )
616+ async with engine .begin () as conn :
617+ await _migrate_add_job_name (conn )
618+ async with engine .begin () as conn :
619+ await _migrate_add_job_user_id (conn )
620+ async with engine .begin () as conn :
621+ await _migrate_add_job_config_hash (conn )
622+ async with engine .begin () as conn :
623+ await _migrate_add_job_disk_size (conn )
481624 _log .info ("PredomicsApp started — data_dir=%s" , settings .data_dir )
482625 yield
483626
0 commit comments