Am Mittwoch, 4. Juni 2008 22:09 schrieb Laurence Rowe:
Hermann Himmelbauer wrote:
In my application, I then use getSASession() to retrieve my session.
However, what I think is not that beautiful is the s.bind = engine
part. Are there any suggestions how to improve this?
You have two options
If you ever need to mix objects from different `sites` into the same
session, you should use an adapter on your root object like:
If you don't need to mix objects from different `sites` then you can
register a local utility for ISessionConfig.
def scope():
return getUtility(ISessionConfig).uid, thread.get_ident()
def factory():
engine = Engine(getUtility(ISessionConfig).url)
return create_session(
transactional=True, autoflush=True, bind=engine
extension=ZopeTransactionExtension(),
))
Session = scoped_session(factory, scopefunc=scope)
Then you can just import Session and use:
session = Session()
Ok, great, thanks for help. The only thing I don't understand is what uid
from the SessionConfig utility is. Below is my full database integration code
which works for me, perhaps this is helpful to someone else.
Btw., I'd suggest to put such code / session use cases in some Zope package,
maybe into zope.sqlalchemy, or e.g. zope.sqlalchemy_utility as it's really
difficult for non-insiders to set this up.
-
import thread
from persistent import Persistent
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, create_session
from zope.interface import Interface, implements, alsoProvides
from zope.schema import TextLine, Bool
from zope.component import getUtility, adapter
from zope.sqlalchemy import ZopeTransactionExtension
from zmyapp.interfaces import INewMySiteEvent
from zope.i18nmessageid import MessageFactory
_ = MessageFactory('xyz')
class ISAEngineUtility(Interface):
SQLAlchemy Engine Utility
db_type = TextLine(title = _(uDatabase Type))
db_username = TextLine(title = _(uDatabase Username))
db_password = TextLine(title = _(uDatabase Password))
db_host = TextLine(title = _(uDatabase Host))
db_name = TextLine(title = _(uDatabase Name))
db_echo = Bool(title = _(uEcho Database Operations))
db_enconding = TextLine(title = _(uDatabase Encoding))
db_convert_unicode = Bool(title = _(uDatabase Unicode Conversion))
class SAEngineUtility(Persistent):
implements(ISAEngineUtility)
# FIXME FIXME!!!
uid = 12345
def reset_engine_on_attrset(dsn_part):
doc = Reset engine when setting variable (stored in %s) % dsn_part
def fget(self):
return getattr(self, dsn_part, None)
def fset(self, value):
oldvalue = getattr(self, dsn_part, None)
if oldvalue != value:
setattr(self, dsn_part, value)
if getattr(self, 'init_done', False):
self._resetEngine()
return {'fget': fget, 'fset': fset, 'doc': doc}
db_type = property(**reset_engine_on_attrset('_db_type'))
db_username = property(**reset_engine_on_attrset('_db_username'))
db_password = property(**reset_engine_on_attrset('_db_password'))
db_host = property(**reset_engine_on_attrset('_db_host'))
db_name = property(**reset_engine_on_attrset('_db_name'))
db_echo = property(**reset_engine_on_attrset('_db_echo'))
db_encoding = property(**reset_engine_on_attrset('_db_encoding'))
db_convert_unicode =property(
**reset_engine_on_attrset('_db_convert_unicode'))
def __init__(self, db_type, db_username, db_password, db_host, db_name,
db_echo = False,
db_encoding = 'utf-8',
db_convert_unicode = False):
self.db_type = db_type
self.db_username = db_username
self.db_password = db_password
self.db_host = db_host
self.db_name = db_name
self.db_echo = db_echo
self.db_encoding = db_encoding
self.db_convert_unicode = db_convert_unicode
self.init_done = True
def mkDSN(self):
Create database DSN out of DSN elements
userpass = ''
if self.db_username:
userpass += self.db_username
if self.db_password:
userpass += ':' + self.db_password
if self.db_host == 'localhost':
db_host = ''
else:
db_host = self.db_host
if db_host and userpass:
db_host = userpass + '@' + self.db_host
elif not db_host and userpass:
db_host = userpass + '@localhost'
return '%s://%s/%s' % (self.db_type, db_host, self.db_name)
def getEngine(self):
engine = getattr(self, '_v_engine', None)
if engine:
return engine
# No engine available, create a new one
self._v_engine = create_engine(self.mkDSN(),
echo = self.db_echo,