PŘI AKTUALIZACI DUPLIKÁTNÍHO KLÍČE post verze 1.2 pro MySQL
Tato funkce je nyní zabudována pouze do SQLAlchemy pro MySQL. Odpověď somada141 níže má nejlepší řešení:https://stackoverflow.com/a/48373874/319066
PŘI AKTUALIZACI DUPLIKÁTNÍHO KLÍČE v příkazu SQL
Pokud chcete, aby vygenerovaný SQL skutečně obsahoval ON DUPLICATE KEY UPDATE , nejjednodušší způsob zahrnuje použití @compiles dekoratér.
Kód (propojený z dobrého vlákna na toto téma na redditu ) příklad lze nalézt na github :
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Insert
@compiles(Insert)
def append_string(insert, compiler, **kw):
s = compiler.visit_insert(insert, **kw)
if 'append_string' in insert.kwargs:
return s + " " + insert.kwargs['append_string']
return s
my_connection.execute(my_table.insert(append_string = 'ON DUPLICATE KEY UPDATE foo=foo'), my_values)
Všimněte si však, že v tomto přístupu musíte ručně vytvořit řetězec append_string. Pravděpodobně byste mohli změnit funkci append_string tak, aby automaticky změnila vložený řetězec na insert s řetězcem 'ON DUPLICATE KEY UPDATE', ale to zde kvůli lenosti neudělám.
PŘI AKTUALIZACI DUPLIKÁTNÍHO KLÍČE funkčnost v rámci ORM
SQLAlchemy neposkytuje rozhraní pro ON DUPLICATE KEY UPDATE nebo SLOUČIT nebo jakákoli jiná podobná funkce ve vrstvě ORM. Přesto má session.merge()
funkce, která dokáže replikovat funkci pouze pokud je dotyčný klíč primárním klíčem .
session.merge(ModelObject) nejprve zkontroluje, zda existuje řádek se stejnou hodnotou primárního klíče odesláním SELECT dotazem (nebo vyhledáním lokálně). Pokud ano, nastaví někde příznak označující, že ModelObject je již v databázi a že SQLAlchemy by měla používat UPDATE dotaz. Všimněte si, že sloučení je o něco složitější než toto, ale dobře replikuje funkčnost s primárními klíči.
Ale co když chcete NA AKTUALIZACI DUPLIKÁTNÍHO KLÍČE funkčnost s neprimárním klíčem (například jiným jedinečným klíčem)? SQLAlchemy bohužel žádnou takovou funkci nemá. Místo toho musíte vytvořit něco, co se podobá Djangově get_or_create() . Pokrývá to další odpověď StackOverflow
, a já sem pro usnadnění vložím jeho upravenou funkční verzi.
def get_or_create(session, model, defaults=None, **kwargs):
instance = session.query(model).filter_by(**kwargs).first()
if instance:
return instance
else:
params = dict((k, v) for k, v in kwargs.iteritems() if not isinstance(v, ClauseElement))
if defaults:
params.update(defaults)
instance = model(**params)
return instance