"""erp-platform 0.9.13's per-request identity map and hr's one raw write path (infrastructure/batching.execute_many).
On psycopg the batch is sent in a pipeline on the raw driver, which bypasses SQLAlchemy's cursor events — the map
would keep answering the rows it read before the batch. Here the pipeline is played by the raw sqlite3 connection
(same transaction, no SQLAlchemy events), which is exactly the bypass.
"""
from contextlib import contextmanager
from sqlalchemy import insert, update
from sqlalchemy.dialects import sqlite
from erp_platform import repository
from brick_hr.infrastructure import batching as B
from brick_hr.infrastructure import tables as T
class _RawPipeline:
"""psycopg's pipeline()/cursor() shape over the DBAPI connection under `conn`."""
def __init__(self, conn):
self.raw = conn.connection.dbapi_connection
@contextmanager
def pipeline(self):
yield self
@contextmanager
def cursor(self):
cur = self.raw.cursor()
try:
yield cur
finally:
cur.close()
def _named(conn, stmt):
c = stmt.compile(dialect=sqlite.dialect(paramstyle="named"))
return str(c), c.construct_params()
def test_a_pipelined_write_drops_the_rows_the_request_already_read(engine, monkeypatch):
monkeypatch.setattr(B, "_pipeline_conn", lambda conn: _RawPipeline(conn))
monkeypatch.setattr(B, "_compile", _named)
key = {"playbook_id": "starter", "version": 1}
with engine.begin() as conn:
conn.execute(insert(T.playbook).values(**key, name="before", status="active"))
with repository.request_scope({"sub": "hr-admin"}), engine.begin() as conn:
assert T.PLAYBOOK.load(conn, key)["hr_playbook"][0]["name"] == "before"
B.execute_many(conn, [update(T.playbook).where(T.playbook.c.playbook_id == "starter")
.values(name="after")])
assert T.PLAYBOOK.load(conn, key)["hr_playbook"][0]["name"] == "after"