从 python 中的 sqlalchemy 中删除对象时出现问题

Question

我有以下测试代码，让我很满意。我有一个开放的会话（self.session），其中可能有我不想打扰的未提交的实体。
我想删除一个我知道与其他任何事物都不相关的专有实体（可以安全删除）。我创建一个新会话并将其删除。我可以确认它已从数据库中删除。

现在，我想确保它已从原始 self.session 中删除，这就是我遇到问题的地方。

from sqlalchemy.orm import Session, sessionmaker

from database.property import Property


def get_session() -> Session:
    engine = create_engine('a legitimate string that connects to my MySql database')
    session = sessionmaker(bind=engine)()
    return session


class TestingOnly:
    def __init__(self):
        self.session = get_session()

    @staticmethod
    def in_all(session: Session,
               prop_id: int) -> bool:
        all_properties = session.query(Property).all()

        for p in all_properties:
            if p.id == prop_id:
                return True
        return False

    @staticmethod
    def delete_prop(prop):
        session = get_session()
        deleting_prop = session.get(Property, prop.id)
        session.delete(deleting_prop)
        session.commit()
        session.close()

    def start(self):
        prop = self.session.query(Property).first()
        prop_id = prop.id
        assert TestingOnly.in_all(self.session, prop_id)
        TestingOnly.delete_prop(prop)
        self.session.flush()
        self.session.expire(prop)
        self.session.expunge(prop)
        self.session.expire_all()
        #################################
        # This throws an assertion error:
        assert not TestingOnly.in_all(self.session, prop_id)


def main():
    testing_only = TestingOnly()
    testing_only.start()


if __name__ == '__main__':
    main()

我仍然对以下代码遇到同样的问题（显然我遗漏了一些东西）：

from sqlalchemy.orm import Session, sessionmaker
from database.property import Property


class TestingOnly:
    engine = create_engine('a legitimate string that connects to my MySql database')

    def __init__(self):
        self.session = self.get_session()

    def get_session(self) -> Session:
        session = sessionmaker(bind=self.engine)()
        return session

    @staticmethod
    def in_all(session: Session,
               prop_id: int) -> bool:
        all_properties = session.query(Property).all()

        for p in all_properties:
            if p.id == prop_id:
                return True
        return False

    def delete_prop(self, prop):
        session = self.get_session()
        deleting_prop = session.get(Property, prop.id)
        session.delete(deleting_prop)
        session.commit()
        session.close()

    def start(self):
        prop = self.session.query(Property).first()
        prop_id = prop.id
        assert TestingOnly.in_all(self.session, prop_id)
        self.delete_prop(prop)
        # self.session.flush()
        # self.session.expire(prop)
        # self.session.expunge(prop)
        # self.session.expire_all()
        #################################
        # This throws an assertion error:
        assert not TestingOnly.in_all(self.session, prop_id)


def main():
    testing_only = TestingOnly()
    testing_only.start()


if __name__ == '__main__':
    main()

更多更改（仍然不起作用）：

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from database.property import Property


class TestingOnly:
    engine = create_engine('a legitimate string that connects to my MySql database')

    def __init__(self):
        self.session = self.get_session()

    def get_session(self) -> Session:
        session = sessionmaker(bind=self.engine)()
        return session

    @staticmethod
    def in_all(session: Session,
               prop_id: int) -> bool:
        all_properties = session.query(Property).all()

        for p in all_properties:
            if p.id == prop_id:
                return True
        return False

    def delete_prop(self, prop):
        session = self.get_session()
        deleting_prop = session.get(Property, prop.id)
        session.delete(deleting_prop)
        session.commit()
        session.close()

    def add_property(self) -> int:
        session = self.get_session()
        prop = Property()
        session.add(prop)
        session.flush()
        prop_id = prop.id
        session.commit()
        session.close()
        return prop_id

    def verify_property_exists(self, prop_id: int) -> bool:
        session = self.get_session()
        prop = session.get(Property, prop_id)
        return prop is not None

    def start(self):
        prop_id = self.add_property()
        self.verify_property_exists(prop_id)
        prop = self.session.get(Property, prop_id)
        assert self.in_all(self.session, prop_id)
        self.delete_prop(prop)
        #self.session.flush()
        #self.session.expire(prop)
        #self.session.expunge(prop)
        #self.session.expire_all()
        #################################
        # This throws an assertion error:
        assert not TestingOnly.in_all(self.session, prop_id)


def main():
    testing_only = TestingOnly()
    testing_only.start()


if __name__ == '__main__':
    main()

Answer 1

@python_user'解决方案对我有用。 引擎 = create_engine(connection_string,isolation_level='READ COMMITTED')