chrisplim · May 13, 2025 07:06 · chrisplim · May 13, 2025
diff --git a/test_autocommit_behavior.py b/test_autocommit_behavior.py
 from contextlib import contextmanager
 from typing import Iterator

 from sqlalchemy import BigInteger, Engine, create_engine
 from sqlalchemy.engine.url import URL
 from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

 Column = mapped_column


 def create_autocommit_engine() -> Engine:
    url = URL.create(
        drivername="mysql+pymysql",
        username="root",
        password="strong_password",
        host="127.0.0.1",
        port=3306,
        database="mydb",
    )

    options = {
        "pool_size": 1,
        "max_overflow": 0,
        "pool_pre_ping": True,
        "pool_use_lifo": True,
        "connect_args": {"connect_timeout": 1},
    }

    return create_engine(
        url,
        **options,
        pool_reset_on_return=None,
        isolation_level="AUTOCOMMIT",
        future=True,
        execution_options={
            "autocommit": True,
            "isolation_level": "AUTOCOMMIT",
        },
    )


 @contextmanager
 def with_readonly_session(engine: Engine) -> Iterator[Session]:
    connection = engine.connect()
    connection.execution_options(isolation_level="AUTOCOMMIT", reset_on_return=False, expire_on_commit=False)
    session = Session(connection)
    try:
        yield session
    finally:
        session.close()


 class Base(DeclarativeBase):
    pass


 class TestTable(Base):
    __tablename__ = "test_table"

    id: Mapped[int] = Column(BigInteger(), primary_key=True, autoincrement=True)

    def __init__(self, *, id: int | None = None):
        self.id = id


 def main() -> None:
    engine = create_autocommit_engine()
    TestTable.metadata.create_all(engine)

    with with_readonly_session(engine) as session:
        session.query(TestTable).all()


 if __name__ == "__main__":
    main()
	from contextlib import contextmanager
	from typing import Iterator

	from sqlalchemy import BigInteger, Engine, create_engine
	from sqlalchemy.engine.url import URL
	from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

	Column = mapped_column


	def create_autocommit_engine() -> Engine:
	url = URL.create(
	drivername="mysql+pymysql",
	username="root",
	password="strong_password",
	host="127.0.0.1",
	port=3306,
	database="mydb",
	)

	options = {
	"pool_size": 1,
	"max_overflow": 0,
	"pool_pre_ping": True,
	"pool_use_lifo": True,
	"connect_args": {"connect_timeout": 1},
	}

	return create_engine(
	url,
	**options,
	pool_reset_on_return=None,
	isolation_level="AUTOCOMMIT",
	future=True,
	execution_options={
	"autocommit": True,
	"isolation_level": "AUTOCOMMIT",
	},
	)


	@contextmanager
	def with_readonly_session(engine: Engine) -> Iterator[Session]:
	connection = engine.connect()
	connection.execution_options(isolation_level="AUTOCOMMIT", reset_on_return=False, expire_on_commit=False)
	session = Session(connection)
	try:
	yield session
	finally:
	session.close()


	class Base(DeclarativeBase):
	pass


	class TestTable(Base):
	__tablename__ = "test_table"

	id: Mapped[int] = Column(BigInteger(), primary_key=True, autoincrement=True)

	def __init__(self, *, id: int \| None = None):
	self.id = id


	def main() -> None:
	engine = create_autocommit_engine()
	TestTable.metadata.create_all(engine)

	with with_readonly_session(engine) as session:
	session.query(TestTable).all()


	if __name__ == "__main__":
	main()
No results found