Skip to content

Trouble using in pytests #8

Description

@Chitrank-Dixit

I am trying to use this in my pytests but I am getting error

AttributeError: type object 'Base' has no attribute 'metadata'

My code is

class Base:

    def get_session(self):
        cherrypy.tools.db = SQLAlchemyTool()

        db_uri = 'postgresql://{0}:{1}@{2}:{3}/{4}'.format(
            settings['database']['user'],
            settings['database']['password'],
            settings['database']['host'],
            settings['database']['port'],
            settings['database']['dbname']
        )
        sqlalchemy_plugin = SQLAlchemyPlugin(
            cherrypy.engine, Base, db_uri,
            echo=True
        )
        sqlalchemy_plugin.subscribe()
        sqlalchemy_plugin.bind_db()

        engine = create_engine(db_uri)
        Session = sessionmaker(bind=engine)
        return Session()

and when I try to use this

class TestAuditModel(Base):

    @property
    def get_db(self):
        return cherrypy.request.db

    def test_insert(self):
        db = self.get_session() # On this line I get issues 
        test = Test(starttime=datetime.now(),
                      endtime=datetime.now(),
                      count=1)

Now I had written this file after looking at the module files

class SQLAlchemyPlugin(plugins.SimplePlugin):
    def __init__(self, bus, orm_base, db_uri, **kwargs):
        """
        The plugin is registered to the CherryPy engine and therefore
        is part of the bus (the engine *is* a bus) registery.

        We use this plugin to create the SA engine. At the same time,
        when the plugin starts we create the tables into the database
        using the mapped class of the global metadata.

        Finally we create a new 'bind' channel that the SA tool
        will use to map a session to the SA engine at request time.
        """
        plugins.SimplePlugin.__init__(self, bus)
        self.db_uri = db_uri
        self.orm_base = orm_base
        self.create_kwargs = kwargs

        self.bus.subscribe('db.start_session', self.start_session)
        self.bus.subscribe('db.bind_db', self.bind_db)

        self.sa_engine = None

    def start(self):
        self.sa_engine = create_engine(self.db_uri)

    def bind_db(self):
        if not self.sa_engine:
            self.start()
        self.orm_base.metadata.bind = self.sa_engine # (here I get the error

    def stop(self):
        if self.sa_engine:
            self.sa_engine.dispose()
            self.sa_engine = None

    def start_session(self, session):
        session.configure(bind=self.sa_engine)

    def execute(self):
        return create_engine(self.db_uri).connect()

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions