fix(deps): upgrade to sqlalchemy version 2

Bump sqlalchemy to version 2.0.17

There are quite some changes that happened with v2.
We are currently relying on the "auto-commit" feature which was removed.
For the moment we can use a wrapper class to mimic the auto-commit
behavior allowing us to move to v2.

Ultimately, the (db) session management needs some overhaul though.

Signed-off-by: moson-mo <mo-son@mailbox.org>
This commit is contained in:
moson-mo 2023-06-30 11:05:18 +02:00 committed by moson
parent 9eda6a42c6
commit ec090d7b30
No known key found for this signature in database
GPG key ID: 4A4760AB4EE15296
12 changed files with 112 additions and 170 deletions

View file

@ -44,7 +44,7 @@ from multiprocessing import Lock
import py
import pytest
from posix_ipc import O_CREAT, Semaphore
from sqlalchemy import create_engine
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
from sqlalchemy.engine.base import Engine
from sqlalchemy.exc import ProgrammingError
@ -106,15 +106,16 @@ def _create_database(engine: Engine, dbname: str) -> None:
:param dbname: Database name to create
"""
conn = engine.connect()
try:
conn.execute(f"CREATE DATABASE {dbname}")
except ProgrammingError: # pragma: no cover
# The database most likely already existed if we hit
# a ProgrammingError. Just drop the database and try
# again. If at that point things still fail, any
# exception will be propogated up to the caller.
conn.execute(f"DROP DATABASE {dbname}")
conn.execute(f"CREATE DATABASE {dbname}")
with conn.begin():
try:
conn.execute(text(f"CREATE DATABASE {dbname}"))
except ProgrammingError: # pragma: no cover
# The database most likely already existed if we hit
# a ProgrammingError. Just drop the database and try
# again. If at that point things still fail, any
# exception will be propogated up to the caller.
conn.execute(text(f"DROP DATABASE {dbname}"))
conn.execute(text(f"CREATE DATABASE {dbname}"))
conn.close()
initdb.run(AlembicArgs)
@ -126,9 +127,10 @@ def _drop_database(engine: Engine, dbname: str) -> None:
:param engine: Engine returned by test_engine()
:param dbname: Database name to drop
"""
aurweb.schema.metadata.drop_all(bind=engine)
# aurweb.schema.metadata.drop_all(bind=engine)
conn = engine.connect()
conn.execute(f"DROP DATABASE {dbname}")
with conn.begin():
conn.execute(text(f"DROP DATABASE {dbname}"))
conn.close()