Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __call__(self, uri):
uri = uri.replace('postgres://', 'http://', 1)
parsed_uri = urlparse.urlsplit(uri)
kw = dict(cgi.parse_qsl(parsed_uri.query))
adapter_factory, kw = self.adapter_helper(parsed_uri, kw)
kw, unused = self.interpret_kwargs(kw)
demostorage = kw.pop('demostorage', False)
options = Options(**kw)
def factory():
adapter = adapter_factory(options)
storage = RelStorage(adapter=adapter, options=options)
if demostorage:
storage = DemoStorage(base=storage)
return storage
return factory, unused
def RelStorageConfigurationFactory(key, dbconfig):
if not RELSTORAGE:
raise Exception("You must install the relstorage package before you can use "
"it as a dabase adapter.")
config = dbconfig.get('configuration', {})
options = Options(**dbconfig['options'])
if dbconfig['type'] == 'postgres':
from relstorage.adapters.postgresql import PostgreSQLAdapter
dsn = "dbname={dbname} user={user} host={host} password={password} port={port}".format(**dbconfig['dsn']) # noqa
adapter = PostgreSQLAdapter(dsn=dsn, options=options)
rs = RelStorage(adapter=adapter, options=options)
db = DB(rs)
try:
conn = db.open()
rootobj = conn.root()
if not IDatabase.providedBy(rootobj):
alsoProvides(rootobj, IDatabase)
transaction.commit()
except: # noqa
pass
finally:
rootobj = None