Comment faire un upsert avec SqlAlchemy?

J'ai un dossier que je veux exister dans la base de données si elle n'est pas là, et si il y est déjà (clé primaire existe) je veux que les champs à mettre à jour l'état actuel. Ceci est souvent appelé un upsert.

La suite incomplète extrait de code montre ce que sera le travail, mais il semble trop maladroit (surtout si il y avait beaucoup plus de colonnes). Ce qui est le mieux/meilleur moyen?

Base = declarative_base()
class Template(Base):
    __tablename__ = 'templates'
    id = Column(Integer, primary_key = True)
    name = Column(String(80), unique = True, index = True)
    template = Column(String(80), unique = True)
    description = Column(String(200))
    def __init__(self, Name, Template, Desc):
        self.name = Name
        self.template = Template
        self.description = Desc

def UpsertDefaultTemplate():
    sess = Session()
    desired_default = Template("default", "AABBCC", "This is the default template")
    try:
        q = sess.query(Template).filter_by(name = desiredDefault.name)
        existing_default = q.one()
    except sqlalchemy.orm.exc.NoResultFound:
        #default does not exist yet, so add it...
        sess.add(desired_default)
    else:
        #default already exists.  Make sure the values are what we want...
        assert isinstance(existing_default, Template)
        existing_default.name = desired_default.name
        existing_default.template = desired_default.template
        existing_default.description = desired_default.description
    sess.flush()

Est-il mieux ou moins détaillé façon de faire cela? Quelque chose comme ce serait génial:

sess.upsert_this(desired_default, unique_key = "name")

bien que le unique_key kwarg est évidemment inutile (l'ORM devrait être en mesure de facilement comprendre cela) je l'ai ajouté juste parce que SQLAlchemy a tendance à travailler uniquement avec la clé primaire. par exemple: j'ai été regarder si Session.fusion serait applicable, mais cela ne fonctionne que sur la clé primaire, qui dans ce cas est un autoincrementing id qui n'est pas très utile à cette fin.

Un exemple de cas d'utilisation de ce est tout simplement lors du démarrage d'une application serveur qui peut avoir mis à niveau par défaut des données attendues. c'est à dire: pas de simultanéité des préoccupations pour ce upsert.

Pourquoi ne pouvez-vous pas faire la name champ de clé primaire si elle est unique (et de fusion serait de travailler dans ce cas). Pourquoi avez-vous besoin d'une clé primaire distincte?
Je ne veux pas entrer dans un champ id débat, mais... la réponse courte est "clés étrangères". De plus, c'est que même si le nom est en effet la seule clé unique, il y a deux problèmes. 1) lorsqu'un modèle d'enregistrement est référencé par 50 millions d'enregistrements d'une autre table ayant que FK comme un champ de type chaîne est fou. Indexé entier est mieux, donc apparemment inutile colonne id. et 2) s'étendant sur que, si la chaîne a été utilisé comme la FK, il y a maintenant deux emplacements pour mettre à jour le nom de si/quand il change, ce qui est ennuyeux, et en proie à mort les problèmes de relations. L'id ne change jamais.
vous pouvez essayer une nouvelle (beta) upsert bibliothèque pour python... il est compatible avec les psycopg2, sqlite3, MySQLdb
voir aussi ce fil: Ne SQLAlchemy ont un équivalent de Django obtenir ou de créer?

InformationsquelleAutor Russ | 2011-08-23

43

SQLAlchemy a un "enregistrer ou mettre à jour" comportement, qui, dans les versions les plus récentes a été construit dans les session.add, mais auparavant, il a été séparé session.saveorupdate appel. Ce n'est pas une "upsert", mais il peut être assez bon pour répondre à vos besoins.

Il est bon que vous le posez sur une classe avec plusieurs clés uniques; je crois que c'est précisément la raison il n'y a pas qu'une seule bonne façon de le faire. La clé primaire est également une clé unique. Si il n'y avait pas de contraintes unique, seule la clé primaire, il serait assez simple problème: si de rien avec l'ID existe, ou si l'ID n'est, de créer un nouveau record; d'autre mise à jour de tous les autres champs de l'enregistrement existant avec cette clé primaire.

Cependant, quand il y a d'autres contraintes unique, il est logique de problèmes avec cette approche simple. Si vous voulez "upsert" un objet, et la clé primaire de l'objet correspond à un enregistrement existant, mais d'une autre colonne unique correspond à une différents record, alors que faites-vous? De même, si la clé primaire ne correspond à aucun enregistrement existant, mais d'une autre colonne unique ne correspondre à un enregistrement existant, alors quoi? Il y a peut être une réponse correcte pour votre situation particulière, mais en général, je dirais qu'il n'y a pas qu'une seule bonne réponse.

Qui serait la raison, il n'est pas construit dans le "upsert de l'opération". L'application doit définir ce que cela signifie dans chaque cas particulier.

InformationsquelleAutor wberry

SQLAlchemy prend en charge ON CONFLICT maintenant avec deux méthodes on_conflict_do_update() et on_conflict_do_nothing():

La copie de la documentation:

from sqlalchemy.dialects.postgresql import insert

stmt = insert(my_table).values(user_email='[email protected]', data='inserted data')
stmt = stmt.on_conflict_do_update(
    index_elements=[my_table.c.user_email],
    index_where=my_table.c.user_email.like('%@gmail.com'),
    set_=dict(data=stmt.excluded.data)
    )
conn.execute(stmt)

http://docs.sqlalchemy.org/en/latest/dialects/postgresql.html?highlight=conflict#insert-on-conflict-upsert

MySQL est également pris en charge avec on_duplicate_key_update

InformationsquelleAutor P.R.

8

- Je utiliser une "avant de faire le saut" approche:
```
# first get the object from the database if it exists
# we're guaranteed to only get one or zero results
# because we're filtering by primary key
switch_command = session.query(Switch_Command).\
    filter(Switch_Command.switch_id == switch.id).\
    filter(Switch_Command.command_id == command.id).first()

# If we didn't get anything, make one
if not switch_command:
    switch_command = Switch_Command(switch_id=switch.id, command_id=command.id)

# update the stuff we care about
switch_command.output = 'Hooray!'
switch_command.lastseen = datetime.datetime.utcnow()

session.add(switch_command)
# This will generate either an INSERT or UPDATE
# depending on whether we have a new object or not
session.commit()
```
L'avantage est que c'est db-neutre et je pense qu'il est clair à lire. L'inconvénient est qu'il y a un potentiel condition de course dans un scénario comme suit:
- nous requête de la base de données pour un switch_command et de ne pas en trouver un
- nous créer un switch_command
- un autre processus ou thread crée un switch_command avec la même clé primaire comme la nôtre
- nous essayons d'engager notre switch_command
- Cette question poignées de la condition de la course avec un try/catch
- Le but de upsert est d'éviter la condition de la course décrite ici.
- Je sais - c'est pourquoi il est vraiment triste que SQLALchemy fait, il est difficile de le faire proprement et de façon portable... j'ai mis en évidence la condition de la course dans ma réponse
InformationsquelleAutor Ben

De nos jours, SQLAlchemy fournit deux fonctions utiles on_conflict_do_nothing et on_conflict_do_update. Ces fonctions sont utiles, mais vous obliger à swich de l'ORM interface de plus bas niveau d'un SQLAlchemy De Base.

Bien que ces deux fonctions permettent de upserting à l'aide de SQLAlchemy de la syntaxe pas que difficile, ces fonctions sont loin d'offrir une gamme complète de out-of-the-box solution à upserting.

Mon utilisation est à upsert un gros morceau de lignes dans une seule requête SQL/session de l'exécution. J'ai l'habitude de rencontrer deux problèmes avec upserting:

Par exemple, une hausse du niveau de la moraine d'oak ridges fonctionnalités, nous avons pris l'habitude de sont manquants. Vous ne pouvez pas utiliser l'ORM objets, mais plutôt de fournir ForeignKeys au moment de l'insertion.

Je suis en utilisant cette fonction suivante que j'ai écrit pour traiter de ces questions:

def upsert(session, model, rows):
    table = model.__table__
    stmt = postgresql.insert(table)
    primary_keys = [key.name for key in inspect(table).primary_key]
    update_dict = {c.name: c for c in stmt.excluded if not c.primary_key}

    if not update_dict:
        raise ValueError("insert_or_update resulted in an empty update_dict")

    stmt = stmt.on_conflict_do_update(index_elements=primary_keys,
                                      set_=update_dict)

    seen = set()
    foreign_keys = {col.name: list(col.foreign_keys)[0].column for col in table.columns if col.foreign_keys}
    unique_constraints = [c for c in table.constraints if isinstance(c, UniqueConstraint)]
    def handle_foreignkeys_constraints(row):
        for c_name, c_value in foreign_keys.items():
            foreign_obj = row.pop(c_value.table.name, None)
            row[c_name] = getattr(foreign_obj, c_value.name) if foreign_obj else None

        for const in unique_constraints:
            unique = tuple([const,] + [row[col.name] for col in const.columns])
            if unique in seen:
                return None
            seen.add(unique)

        return row

    rows = list(filter(None, (handle_foreignkeys_constraints(row) for row in rows)))
    session.execute(stmt, rows)

on_conflict est uniquement disponible pour les serveurs qui prennent en charge native SUR les CONFLITS clauases. Par conséquent, seulement postgresql
Maintenant SQLAlchemy prend également en charge LE DOUBLE de la CLÉ de mise à JOUR pour MySQL.

InformationsquelleAutor NirIzr

Cela fonctionne pour moi avec sqlite3 et postgres. Mais il peut échouer avec le combiné des contraintes de clé primaire et sera très probablement échouer avec d'autres contraintes uniques.

    try:
        t = self._meta.tables[data['table']]
    except KeyError:
        self._log.error('table "%s" unknown', data['table'])
        return

    try:
        q = insert(t, values=data['values'])
        self._log.debug(q)
        self._db.execute(q)
    except IntegrityError:
        self._log.warning('integrity error')
        where_clause = [c.__eq__(data['values'][c.name]) for c in t.c if c.primary_key]
        update_dict = {c.name: data['values'][c.name] for c in t.c if not c.primary_key}
        q = update(t, values=update_dict).where(*where_clause)
        self._log.debug(q)
        self._db.execute(q)
    except Exception as e:
        self._log.error('%s: %s', t.name, e)

InformationsquelleAutor ThePsyjo

Ci-dessous fonctionne très bien pour moi avec le décalage vers le rouge de la base de données et aussi le travail pour combiné contrainte de clé primaire.

SOURCE : cette

Juste quelques modifications requises pour la création de SQLAlchemy moteur dans la fonction
def start_engine()

from sqlalchemy import Column, Integer, Date ,Metadata
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.dialects import postgresql
Base = declarative_base()
def start_engine():
engine = create_engine(os.getenv('SQLALCHEMY_URI', 
'postgresql://localhost:5432/upsert'))
connect = engine.connect()
meta = MetaData(bind=engine)
meta.reflect(bind=engine)
return engine
class DigitalSpend(Base):
__tablename__ = 'digital_spend'
report_date = Column(Date, nullable=False)
day = Column(Date, nullable=False, primary_key=True)
impressions = Column(Integer)
conversions = Column(Integer)
def __repr__(self):
return str([getattr(self, c.name, None) for c in self.__table__.c])
def compile_query(query):
compiler = query.compile if not hasattr(query, 'statement') else 
query.statement.compile
return compiler(dialect=postgresql.dialect())
def upsert(session, model, rows, as_of_date_col='report_date', no_update_cols=[]):
table = model.__table__
stmt = insert(table).values(rows)
update_cols = [c.name for c in table.c
if c not in list(table.primary_key.columns)
and c.name not in no_update_cols]
on_conflict_stmt = stmt.on_conflict_do_update(
index_elements=table.primary_key.columns,
set_={k: getattr(stmt.excluded, k) for k in update_cols},
index_where=(getattr(model, as_of_date_col) < getattr(stmt.excluded, as_of_date_col))
)
print(compile_query(on_conflict_stmt))
session.execute(on_conflict_stmt)
session = start_engine()
upsert(session, DigitalSpend, initial_rows, no_update_cols=['conversions'])

InformationsquelleAutor Aditi Srivastava

Ce qui permet l'accès à la sous-tendent les modèles basés sur des noms de chaîne

def get_class_by_tablename(tablename):
"""Return class reference mapped to table.
https://stackoverflow.com/questions/11668355/sqlalchemy-get-model-from-table-name-this-may-imply-appending-some-function-to
:param tablename: String with name of table.
:return: Class reference or None.
"""
for c in Base._decl_class_registry.values():
if hasattr(c, '__tablename__') and c.__tablename__ == tablename:
return c
sqla_tbl = get_class_by_tablename(table_name)
def handle_upsert(record_dict, table):
"""
handles updates when there are primary key conflicts
"""
try:
self.active_session().add(table(**record_dict))
except:
# Here we'll assume the error is caused by an integrity error
# We do this because the error classes are passed from the
# underlying package (pyodbc /sqllite) SQLAlchemy doesn't mask
# them with it's own code - this should be updated to have
# explicit error handling for each new db engine
# <update>add explicit error handling for each db engine</update> 
active_session.rollback()
# Query for conflic class, use update method to change values based on dict
c_tbl_primary_keys = [i.name for i in table.__table__.primary_key] # List of primary key col names
c_tbl_cols = dict(sqla_tbl.__table__.columns) # String:Col Object crosswalk
c_query_dict = {k:record_dict[k] for k in c_tbl_primary_keys if k in record_dict} # sub-dict from data of primary key:values
c_oo_query_dict = {c_tbl_cols[k]:v for (k,v) in c_query_dict.items()} # col-object:query value for primary key cols
c_target_record = session.query(sqla_tbl).filter(*[k==v for (k,v) in oo_query_dict.items()]).first()
# apply new data values to the existing record
for k, v in record_dict.items()
setattr(c_target_record, k, v)

InformationsquelleAutor Schalton

Vous devez vous connecter pour publier un commentaire.