def insert_from_query(self, model, query, **replace):
# The SQLAlchemy core definition of the table
table = inspect(model).local_table
# and the underlying core select statement to source new rows from
print(type(query))
select = query.statement
# validate asssumptions: make sure the query produces rows from the above table
assert table in select.froms, f"{query!r} must produce rows from {model!r}"
assert all(c.name in select.columns for c in table.columns), f"{query!r} must include all {model!r} columns"
# updated select, replacing the indicated columns
as_clause = lambda v: literal(v) if not isinstance(v, ClauseElement) else v
replacements = {name: as_clause(value).label(name) for name, value in replace.items()}
from_select = select.with_only_columns([replacements.get(c.name, c) for c in table.columns])
insert_statement = table.insert().from_select(from_select.columns, from_select)
insert = self.engine.execute(insert_statement)
return insert
I am trying to do mass insert whilst updating one field called 'activity_id' within product table.
The statement generated does that, however I ran into IntegrityError
Error Trace:
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "product_pkey"
DETAIL: Key (id)=(210776) already exists.
I am basically making duplicates of existing entry, and just updating their activity_id, but when creating these cloned entries with one updated field activity_id, I think it should also increment the field id, I think it tries to clone that field right now. In my db it's marked as primary key/ index.
