Verwenden des SQLAlchemy ORM in einer Alembic-Migration: Wie kann ich?

Question

Ich habe derzeit eine Spalte, die HTML-Markup enthält. Innerhalb dieses Markups gibt es einen Zeitstempel, den ich in einer neuen Spalte speichern möchte (so kann ich dagegen abfragen). Meine Idee war, die folgenden in einer einzigen Migration zu tun:

1. Erstellen Sie eine neue Spalte von -zulässige für die Daten
2. Verwenden Sie die ORM die HTML-I
3. Für jede Zeile
   analysieren müssen zurück ziehen
   1. analysieren den HTML-Code den Zeitstempel ziehen
   2. Update das ORM-Objekt

Aber wenn ich versuche, meine Migration auszuführen, scheint es in einer Endlosschleife stecken zu bleiben. Hier ist, was ich bisher habe:

def _extract_publication_date(html): 
    root = html5lib.parse(html, treebuilder='lxml', namespaceHTMLElements=False) 
    publication_date_string = root.xpath("//a/@data-datetime")[0] 
    return parse_date(publication_date) 


def _update_tip(tip): 
    tip.publication_date = _extract_publication_date(tip.rendered_html) 
    tip.save() 


def upgrade(): 
    op.add_column('tip', sa.Column('publication_date', sa.DateTime(timezone=True))) 
    tips = Tip.query.all() 
    map(tips, _update_tip) 


def downgrade(): 
    op.drop_column('tip', 'publication_date') 

Answer 1

weiter aus den Kommentaren, können Sie so etwas wie dies versuchen:

import sqlalchemy as sa 


tip = sa.sql.table(
    'tip', 
    sa.sql.column('id', sa.Integer), 
    sa.sql.column('publication_date', sa.DateTime(timezone=True)), 
) 


def upgrade(): 
    mappings = [ 
     (x.id, _extract_publication_date(x.rendered_html)) 
     for x in Tip.query 
    ] 

    op.add_column('tip', sa.Column('publication_date', sa.DateTime(timezone=True))) 

    exp = sa.sql.case(value=tip.c.id, whens=(
     (op.inline_literal(id), op.inline_literal(publication_date)) 
     for id, publication_date in mappings.iteritems() 
    )) 

    op.execute(tip.update().values({'publication_date': exp})) 


def downgrade(): 
    op.drop_column('tip', 'publication_date') 

Answer 2

Was für mich gearbeitet ist, eine Sitzung zu erhalten, indem Sie folgendermaßen vorgehen:

Answer 3

Nach ein wenig Experimentieren mit @ velochy's Antwort, entschied ich mich für etwas wie das folgende Muster für die Verwendung von SqlAlchemy in Alembic. Dies funktionierte gut für mich und könnte wahrscheinlich als eine allgemeine Lösung für die Frage des OP dienen:

from sqlalchemy.orm.session import Session 
from alembic import op 

def upgrade(): 
    # Attach a sqlalchemy Session to the env connection 
    session = Session(bind=op.get_bind()) 

    # Perform arbitrarily-complex ORM logic 
    instance1 = Model1(foo='bar') 
    instance2 = Model2(monkey='banana') 

    # Add models to Session so they're tracked 
    session.add(instance1) 
    session.add(instance2) 

def downgrade(): 
    # Attach a sqlalchemy Session to the env connection 
    session = Session(bind=op.get_bind()) 

    # Perform ORM logic in downgrade (e.g. clear tables) 
    session.query(Model2).delete() 
    session.query(Model1).delete() 

Dieser Ansatz scheint Transaktionen ordnungsgemäß zu behandeln. Während ich daran arbeitete, würde ich häufig DB-Exceptions erzeugen und sie würden die Dinge wie erwartet zurückrollen.