rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fixing alembic revisions to not use external models.

Browse source
This commit is contained in:
Roland 2025-12-01 15:33:02 -07:00 • committed by lrnselfreliance
parent 1d8738ace2
commit 2a1ed5ab2b
2 changed files with 43 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

19
alembic/versions/53f6780aa599_drop_urls.py
View file

@ -8,11 +8,10 @@ Create Date: 2022-01-03 12:25:51.575505
import os
from alembic import op
from sqlalchemy import text
from sqlalchemy.orm import Session
# revision identifiers, used by Alembic.
from modules.archive.models import Archive
revision = '53f6780aa599'
down_revision = '31896cc29b71'
branch_labels = None
@ -25,14 +24,16 @@ def upgrade():
bind = op.get_bind()
session = Session(bind=bind)
urls = {i['id']: i['url'] for i in session.execute('SELECT id, url FROM url')}
archives = {i['id']: i['url_id'] for i in session.execute('SELECT id, url_id FROM archive')}
urls = {i['id']: i['url'] for i in session.execute(text('SELECT id, url FROM url'))}
archives = {i['id']: i['url_id'] for i in session.execute(text('SELECT id, url_id FROM archive'))}
session.execute('ALTER TABLE archive ADD COLUMN url TEXT')
session.execute(text('ALTER TABLE archive ADD COLUMN url TEXT'))
for archive_id, url_id in archives.items():
archive = session.query(Archive).filter_by(id=archive_id).one()
archive.url = urls[url_id]
session.execute(
text('UPDATE archive SET url = :url WHERE id = :id'),
{'url': urls[url_id], 'id': archive_id}
)
session.execute('ALTER TABLE archive DROP COLUMN url_id')
session.execute('DROP TABLE url')
session.execute(text('ALTER TABLE archive DROP COLUMN url_id'))
session.execute(text('DROP TABLE url'))

40
alembic/versions/migrate_domains_to_collections.py
View file

@ -15,7 +15,8 @@ import os
from typing import Dict, List
from alembic import op
import sqlalchemy as sa
from sqlalchemy import text
from sqlalchemy import text, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
# revision identifiers, used by Alembic.
@ -26,6 +27,32 @@ depends_on = None
DOCKERIZED = True if os.environ.get('DOCKER', '').lower().startswith('t') else False
# Local model definitions for migration stability.
# These are minimal copies of the Collection and CollectionItem models
# that contain only the columns needed for this migration.
# This ensures the migration will work regardless of future changes to the actual models.
MBase = declarative_base()
class MCollection(MBase):
"""Migration-stable Collection model with minimum required columns."""
__tablename__ = 'collection'
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
kind = Column(String, nullable=False, server_default='channel', default='channel')
directory = Column(String, nullable=True)
class MCollectionItem(MBase):
"""Migration-stable CollectionItem model with minimum required columns."""
__tablename__ = 'collection_item'
id = Column(Integer, primary_key=True)
collection_id = Column(Integer, ForeignKey('collection.id', ondelete='CASCADE'), nullable=False)
file_group_id = Column(Integer, ForeignKey('file_group.id', ondelete='CASCADE'), nullable=False)
position = Column(Integer, nullable=False, default=0)
class DomainMigrationStats:
"""Track migration statistics."""
@ -119,9 +146,8 @@ def perform_domain_migration(session: Session, verbose: bool = True) -> DomainMi
Migrate all Domain records to Collection records.
This function is integrated into the alembic migration.
Uses local MCollection and MCollectionItem models for migration stability.
"""
from wrolpi.collections import Collection, CollectionItem
stats = DomainMigrationStats()
# Step 1: Fetch all domains
@ -151,7 +177,7 @@ def perform_domain_migration(session: Session, verbose: bool = True) -> DomainMi
continue
# Check if Collection already exists
existing = session.query(Collection).filter_by(
existing = session.query(MCollection).filter_by(
name=domain_name,
kind='domain'
).first()
@ -163,7 +189,7 @@ def perform_domain_migration(session: Session, verbose: bool = True) -> DomainMi
continue
# Create Collection
collection = Collection(
collection = MCollection(
name=domain_name,
kind='domain',
directory=None, # Domains are unrestricted
@ -235,7 +261,7 @@ def perform_domain_migration(session: Session, verbose: bool = True) -> DomainMi
archive_id = archive['archive_id']
# Check if CollectionItem already exists
existing_item = session.query(CollectionItem).filter_by(
existing_item = session.query(MCollectionItem).filter_by(
collection_id=collection_id,
file_group_id=file_group_id
).first()
@ -246,7 +272,7 @@ def perform_domain_migration(session: Session, verbose: bool = True) -> DomainMi
continue
# Create CollectionItem
item = CollectionItem(
item = MCollectionItem(
collection_id=collection_id,
file_group_id=file_group_id,
position=idx