Skip to content

Trouble using in pytests #8

@Chitrank-Dixit

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()

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