Невозможно получить вставленные данные в базу данных sqlite в памяти с помощью sqlalchemy. ⇐ Python

[Anonymous]

У меня есть класс для интеграционного тестирования, который создает базу данных sqlite в памяти с использованием sqlalchemy, а также предоставляет сеансы для управления операциями с базой данных.
sqlite_connection. py

Код: Выделить всё

class SqliteConnection:
def __init__(self, models: list = None):
self.__engine: Engine = self.__create_engine()
self.__session_factory = sessionmaker(self.__engine)
self.__models = models
self.__session = None
self.__migrate()
self.__seed()

def session(self, has_transaction=False) -> Union[AbstractContextManager[Session], Session]:
if has_transaction:
self.__session = self.__session_factory.begin()
else:
self.__session = self.__session_factory()
return self.__session

def __seed(self) -> None:
models = self.__models if isinstance(self.__models, list) else [self.__models]
if self.__models:
with self.ms_sql_server_session(has_transaction=False) as session:
session.add_all(models)
session.flush()
session.expunge_all()

def __create_engine(self) -> Engine:
return create_engine(f'sqlite:///file:?mode=memory&cache=shared&uri=true')

def __migrate(self) -> None:
self.__base_model: DeclarativeMeta = Base
self.__base_model.metadata.reflect(self.__engine, schema=None)
self.__base_model.metadata.create_all(self.__engine)

Я хочу использовать этот класс для обработки операций с базой данных и впоследствии не терять состояние базы данных. Я хочу иметь возможность запускать запросы, связанные с интеграционным тестом, чтобы убедиться, что операции с базой данных работают должным образом.
sample_integration_test.py

Код: Выделить всё

Base = declarative_base()

class UserModel(Base):
__tablename__ = "user_account"

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(30))
fullname: Mapped[Optional[str]]

addresses: Mapped[List["AddressModel"]] = relationship(
back_populates="user", cascade="all, delete-orphan"
)

def __repr__(self) -> str:
return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class AddressModel(Base):
__tablename__ = "address"

id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
email_address: Mapped[str]

user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey(UserModel.id))

user: Mapped["UserModel"] = relationship(back_populates="addresses")

def __repr__(self) -> str:
return f"Address(id={self.id!r}, email_address={self.email_address!r})"

john = UserModel(
name="john",
fullname="John Doe",
addresses=[AddressModel(email_address="john@example.com")],
)
jane = UserModel(
name="jane",
fullname="Jane Doe",
addresses=[
AddressModel(email_address="jane@example.com"),
AddressModel(email_address="jane@example.org"),
],
)
jack = UserModel(name="jack", fullname="Jack Doe")

sample_models = [john, jane, jack]

sqlite_connection = SqliteConnection()

with sqlite_connection.session() as session:
session.add_all(sample_models)
session.flush()

with sqlite_connection.session() as session:
query = select(UserModel)
result = session.execute(query).scalars().all()

print(result)

Проблема в том, что после первого блока состояние базы данных, кажется, обновляется, и я не могу найти вставленные модели в базе данных. Есть ли способ перенести вставленные модели в базу данных?


Подробнее здесь: https://stackoverflow.com/questions/789 ... sqlalchemy