import sqlalchemy.engine as engine
# sqlite in memory
- sqlite_engine = engine.create_engine('sqlite', ':memory:', {}, **opts)
+ sqlite_engine = engine.create_engine('sqlite', {'filename':':memory:'}, **opts)
# sqlite using a file
- sqlite_engine = engine.create_engine('sqlite', 'querytest.db', {}, **opts)
+ sqlite_engine = engine.create_engine('sqlite', {'filename':'querytest.db'}, **opts)
# postgres
postgres_engine = engine.create_engine('postgres',
<li>use_ansi=True : used only by Oracle; when False, the Oracle driver attempts to support a particular "quirk" of some Oracle databases, that the LEFT OUTER JOIN SQL syntax is not supported, and the "Oracle join" syntax of using <% "<column1>(+)=<column2>" |h%> must be used in order to achieve a LEFT OUTER JOIN. Its advised that the Oracle database be configured to have full ANSI support instead of using this feature.</li>
</ul>
</&>
-</&>
\ No newline at end of file
+</&>
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+__ALL__ = ['oracle', 'postgres', 'sqlite']
\ No newline at end of file
def engine(*args, **params):
return OracleSQLEngine(*args, **params)
+
+def descriptor():
+ return {'name':'oracle',
+ 'description':'Oracle',
+ 'arguments':[
+ ('dsn', 'Data Source Name', None),
+ ('user', 'Username', None),
+ ('password', 'Password', None)
+ ]}
class OracleSQLEngine(ansisql.ANSISQLEngine):
def __init__(self, opts, use_ansi = True, module = None, **params):
def engine(opts, **params):
return PGSQLEngine(opts, **params)
+def descriptor():
+ return {'name':'postgres',
+ 'description':'PostGres',
+ 'arguments':[
+ ('user',"Database Username",None),
+ ('password',"Database Password",None),
+ ('database',"Database Name",None),
+ ('host',"Hostname", None),
+ ]}
+
class PGSQLEngine(ansisql.ANSISQLEngine):
def __init__(self, opts, module = None, **params):
if module is None:
'BLOB' : SLBinary,
}
-def engine(filename, opts, **params):
- return SQLiteSQLEngine(filename, opts, **params)
-
+def engine(opts, **params):
+ return SQLiteSQLEngine(opts, **params)
+
+def descriptor():
+ return {'name':'sqlite',
+ 'description':'SQLite',
+ 'arguments':[
+ ('filename', "Database Filename",None)
+ ]}
+
class SQLiteSQLEngine(ansisql.ANSISQLEngine):
- def __init__(self, filename, opts, **params):
- self.filename = filename
+ def __init__(self, opts, **params):
+ self.filename = opts.pop('filename')
self.opts = opts or {}
ansisql.ANSISQLEngine.__init__(self, **params)
import sqlalchemy.sql as sql
import StringIO, sys
import sqlalchemy.types as types
+import sqlalchemy.databases
def create_engine(name, *args ,**kwargs):
"""creates a new SQLEngine instance.
module = getattr(__import__('sqlalchemy.databases.%s' % name).databases, name)
return module.engine(*args, **kwargs)
+def engine_descriptors():
+ result = []
+ for module in sqlalchemy.databases.__ALL__:
+ module = getattr(__import__('sqlalchemy.databases.%s' % module).databases, module)
+ result.append(module.descriptor())
+ return result
+
class SchemaIterator(schema.SchemaVisitor):
"""a visitor that can gather text into a buffer and execute the contents of the buffer."""
def __init__(self, sqlproxy, **params):
class EngineTest(PersistTest):
def testsqlite(self):
- db = sqllite.engine(':memory:', {}, echo = testbase.echo)
+ db = sqllite.engine({'filename':':memory'}, echo = testbase.echo)
self.do_tableops(db)
def testpostgres(self):
#DBTYPE = 'sqlite_file'
if DBTYPE == 'sqlite_memory':
- db = sqlalchemy.engine.create_engine('sqlite', ':memory:', {}, echo = testbase.echo)
+ db = sqlalchemy.engine.create_engine('sqlite', {'filename':':memory:'}, echo = testbase.echo)
elif DBTYPE == 'sqlite_file':
import sqlalchemy.databases.sqlite as sqllite
- db = sqlalchemy.engine.create_engine('sqlite', 'querytest.db', {}, echo = testbase.echo)
+ db = sqlalchemy.engine.create_engine('sqlite', {'filename':'querytest.db'}, echo = testbase.echo)
elif DBTYPE == 'postgres':
db = sqlalchemy.engine.create_engine('postgres', {'database':'test', 'host':'127.0.0.1', 'user':'scott', 'password':'tiger'}, echo=testbase.echo)