-
Notifications
You must be signed in to change notification settings - Fork 2
Open
Description
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
Labels
No labels