<h3 dir="auto">Describe the bug</h3>
<p dir="auto">Hello everyone,</p>
<p dir="auto">I am using SQLAlchemy 1.4 with Postgresql. I am trying to populate some tables with a large number of records, with some not conforming to expectations. I have been using the following syntax: <code class="notranslate">db.execute(insert(Table).values(records).on_conflict_do_nothing())</code> to resolve issues with duplicates.<br>
However, as expected some of the records have issues of another nature and I want to skip them and retrieve them. I decided to use SAVEPOINTs (<code class="notranslate">begin_nested()</code>) and <code class="notranslate">rollback()</code>. This led to an error I am not able to resolve mentioned in the title: <code class="notranslate">"Can't reconnect until invalid savepoint transaction is rolled back"</code>.</p>
<p dir="auto">In the code you will notice I am using a context manager for the save points, I would like to point out that I get the same error when using explicit rollbacks without a context manager.</p>
<p dir="auto">Thank you for your time!</p>
<h3 dir="auto">To Reproduce</h3>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="engine = _sql.create_engine(DATABASE_URL)
SessionLocal = _orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = _declarative.declarative_base()
def get_session():
    session = SessionLocal()
    try:
        yield session
    finally:
        db.close()


def _bulk_insert(records: list, table, session):
    failed_records = []
    try:
        with session.begin_nested() as transaction:
            transaction.session.execute(insert(table).values(records).on_conflict_do_nothing())
        print(‘Bulk insert successful without any errors’)
        return failed_records
    except Exception:
        print(‘Failed to do general bulk insert, starting chunks...’)
    for chunk in np.array_split(records, 10):
        try:
            with session.begin_nested() as transaction:
                transaction.session.execute(insert(table).values(chunk.tolist()).on_conflict_do_nothing())
        except Exception:
            failed_items.extend(chunk)
    if len(items) != len(failed_records) and len(failed_records)&gt;0:
        return _bulk_insert(failed_records, table, db)
    return failed_items"><pre class="notranslate"><span class="pl-s1">engine</span> <span class="pl-c1">=</span> <span class="pl-s1">_sql</span>.<span class="pl-en">create_engine</span>(<span class="pl-v">DATABASE_URL</span>)
<span class="pl-v">SessionLocal</span> <span class="pl-c1">=</span> <span class="pl-s1">_orm</span>.<span class="pl-en">sessionmaker</span>(<span class="pl-s1">autocommit</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-s1">autoflush</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-s1">bind</span><span class="pl-c1">=</span><span class="pl-s1">engine</span>)
<span class="pl-v">Base</span> <span class="pl-c1">=</span> <span class="pl-s1">_declarative</span>.<span class="pl-en">declarative_base</span>()
<span class="pl-k">def</span> <span class="pl-en">get_session</span>():
    <span class="pl-s1">session</span> <span class="pl-c1">=</span> <span class="pl-v">SessionLocal</span>()
    <span class="pl-k">try</span>:
        <span class="pl-k">yield</span> <span class="pl-s1">session</span>
    <span class="pl-k">finally</span>:
        <span class="pl-s1">db</span>.<span class="pl-en">close</span>()


<span class="pl-k">def</span> <span class="pl-en">_bulk_insert</span>(<span class="pl-s1">records</span>: <span class="pl-s1">list</span>, <span class="pl-s1">table</span>, <span class="pl-s1">session</span>):
    <span class="pl-s1">failed_records</span> <span class="pl-c1">=</span> []
    <span class="pl-k">try</span>:
        <span class="pl-k">with</span> <span class="pl-s1">session</span>.<span class="pl-en">begin_nested</span>() <span class="pl-k">as</span> <span class="pl-s1">transaction</span>:
            <span class="pl-s1">transaction</span>.<span class="pl-s1">session</span>.<span class="pl-en">execute</span>(<span class="pl-en">insert</span>(<span class="pl-s1">table</span>).<span class="pl-en">values</span>(<span class="pl-s1">records</span>).<span class="pl-en">on_conflict_do_nothing</span>())
        <span class="pl-en">print</span>(‘<span class="pl-v">Bulk</span> <span class="pl-s1">insert</span> <span class="pl-s1">successful</span> <span class="pl-s1">without</span> <span class="pl-s1">any</span> <span class="pl-s1">errors</span>’)
        <span class="pl-k">return</span> <span class="pl-s1">failed_records</span>
    <span class="pl-k">except</span> <span class="pl-v">Exception</span>:
        <span class="pl-en">print</span>(‘<span class="pl-v">Failed</span> <span class="pl-s1">to</span> <span class="pl-s1">do</span> <span class="pl-s1">general</span> <span class="pl-s1">bulk</span> <span class="pl-s1">insert</span>, <span class="pl-s1">starting</span> <span class="pl-s1">chunks</span>...’)
    <span class="pl-k">for</span> <span class="pl-s1">chunk</span> <span class="pl-c1">in</span> <span class="pl-s1">np</span>.<span class="pl-en">array_split</span>(<span class="pl-s1">records</span>, <span class="pl-c1">10</span>):
        <span class="pl-k">try</span>:
            <span class="pl-k">with</span> <span class="pl-s1">session</span>.<span class="pl-en">begin_nested</span>() <span class="pl-k">as</span> <span class="pl-s1">transaction</span>:
                <span class="pl-s1">transaction</span>.<span class="pl-s1">session</span>.<span class="pl-en">execute</span>(<span class="pl-en">insert</span>(<span class="pl-s1">table</span>).<span class="pl-en">values</span>(<span class="pl-s1">chunk</span>.<span class="pl-en">tolist</span>()).<span class="pl-en">on_conflict_do_nothing</span>())
        <span class="pl-k">except</span> <span class="pl-v">Exception</span>:
            <span class="pl-s1">failed_items</span>.<span class="pl-en">extend</span>(<span class="pl-s1">chunk</span>)
    <span class="pl-k">if</span> <span class="pl-en">len</span>(<span class="pl-s1">items</span>) <span class="pl-c1">!=</span> <span class="pl-en">len</span>(<span class="pl-s1">failed_records</span>) <span class="pl-c1">and</span> <span class="pl-en">len</span>(<span class="pl-s1">failed_records</span>)<span class="pl-c1">&gt;</span><span class="pl-c1">0</span>:
        <span class="pl-k">return</span> <span class="pl-en">_bulk_insert</span>(<span class="pl-s1">failed_records</span>, <span class="pl-s1">table</span>, <span class="pl-s1">db</span>)
    <span class="pl-k">return</span> <span class="pl-s1">failed_items</span></pre></div>
<h3 dir="auto">Error</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Can't reconnect until invalid savepoint transaction is rolled back. (Background on this error at: http://sqlalche.me/e/14/8s2b)"><pre class="notranslate"><code class="notranslate">Can't reconnect until invalid savepoint transaction is rolled back. (Background on this error at: http://sqlalche.me/e/14/8s2b)
</code></pre></div>
<h3 dir="auto">Versions</h3>
<ul dir="auto">
<li>OS: MacOS 12.4</li>
<li>Python: 3.9.1</li>
<li>SQLAlchemy: 1.4</li>
<li>Database: PostgreSQL</li>
<li>DBAPI (eg: psycopg, cx_oracle, mysqlclient): psycopg</li>
</ul>
<h3 dir="auto">Additional context</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Describe the use case</h3>
<p dir="auto">Allow the UPDATE-FROM syntax with sqlite instead of raising <code class="notranslate">NotImplementedError</code>.</p>
<h3 dir="auto">Databases / Backends / Drivers targeted</h3>
<p dir="auto">SQLite version &gt;= 3.33.0<br>
This is available in the standard library sqlite3 in Python 3.8.7, 3.9.1, and 3.10, or in the pysqlite3 package.</p>
<h3 dir="auto">Example Use</h3>
<div class="highlight highlight-source-sql notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="UPDATE inventory
   SET quantity = quantity - daily.amt
  FROM (SELECT sum(quantity) AS amt, itemId FROM sales GROUP BY 2) AS daily
 WHERE inventory.itemId = daily.itemId;"><pre class="notranslate"><span class="pl-k">UPDATE</span> inventory
   <span class="pl-k">SET</span> quantity <span class="pl-k">=</span> quantity <span class="pl-k">-</span> <span class="pl-c1">daily</span>.<span class="pl-c1">amt</span>
  <span class="pl-k">FROM</span> (<span class="pl-k">SELECT</span> <span class="pl-c1">sum</span>(quantity) <span class="pl-k">AS</span> amt, itemId <span class="pl-k">FROM</span> sales <span class="pl-k">GROUP BY</span> <span class="pl-c1">2</span>) <span class="pl-k">AS</span> daily
 <span class="pl-k">WHERE</span> <span class="pl-c1">inventory</span>.<span class="pl-c1">itemId</span> <span class="pl-k">=</span> <span class="pl-c1">daily</span>.<span class="pl-c1">itemId</span>;</pre></div>
<h3 dir="auto">Additional context</h3>
<p dir="auto">The SQLite UPDATE-FROM syntax is supposed to be compatible with PostgreSQL.<br>
<a href="https://sqlite.org/lang_update.html" rel="nofollow">https://sqlite.org/lang_update.html</a><br>
I found that copying <code class="notranslate">update_from_clause</code> from <code class="notranslate">sqlalchemy.dialects.postgresql.base</code> to <code class="notranslate">sqlalchemy.dialects.sqlite.base</code> makes this work.</p>