sqlite transactions cat be "deferred", "immediate" or "exclusive". The default is "deferred" which means, not starting the transaction until absolutely necessary. This can cause transaction aborts if concurrent transactions start with reading and the proceed to writing. Such aborts can be avoided by using immediate transactions (at the cost of degrading performance).
sqlalchemy abstracts sql dialects including sqlite. It also has a model for writing transactions:
with engine.begin() as connection:
do_something_with_connection
How does one tell sqlalchemy that such a transaction should be immediate. Alternatively, how does one tell sqlalchemy that all sqlite transactions should be immediate?
Core events https://docs.sqlalchemy.org/en/latest/core/events.html can be used to intercept connection events and rewrite the BEGIN statement issued at the start of the transaction to achieve what you want.
See the section of the sqlalchemy documentation on the sqlite dialect for more details https://docs.sqlalchemy.org/en/latest/dialects/sqlite.html.
The example code below is copied directly from the documentation other than for changing BEGIN to BEGIN IMMEDIATE.
from sqlalchemy import create_engine, event
engine = create_engine("sqlite:///myfile.db")
@event.listens_for(engine, "connect")
def do_connect(dbapi_connection, connection_record):
# disable pysqlite's emitting of the BEGIN statement entirely.
# also stops it from emitting COMMIT before any DDL.
dbapi_connection.isolation_level = None
@event.listens_for(engine, "begin")
def do_begin(conn):
# emit our own BEGIN
conn.execute("BEGIN IMMEDIATE")