from sqlalchemy.schema import *
from sqlalchemy.engine import create_engine
-from sqlalchemy.schema import default_metadata
-def global_connect(*args, **kwargs):
- default_metadata.connect(*args, **kwargs)
-
\ No newline at end of file
else:
raise exceptions.InvalidRequestError("This SchemaItem is not connected to any Engine")
- def _set_casing_strategy(self, name, kwargs, keyname='case_sensitive'):
+ def _set_casing_strategy(self, kwargs, keyname='case_sensitive'):
"""Set the "case_sensitive" argument sent via keywords to the item's constructor.
For the purposes of Table's 'schema' property, the name of the
"""
setattr(self, '_%s_setting' % keyname, kwargs.pop(keyname, None))
- def _determine_case_sensitive(self, name, keyname='case_sensitive'):
+ def _determine_case_sensitive(self, keyname='case_sensitive'):
"""Determine the `case_sensitive` value for this item.
For the purposes of Table's `schema` property, the name of the
try:
return self.__case_sensitive
except AttributeError:
- self.__case_sensitive = self._determine_case_sensitive(self.name)
+ self.__case_sensitive = self._determine_case_sensitive()
return self.__case_sensitive
case_sensitive = property(_get_case_sensitive)
"""A metaclass used by the ``Table`` object to provide singleton behavior."""
def __call__(self, name, metadata, *args, **kwargs):
- if isinstance(metadata, sql.Executor):
- # backwards compatibility - get a BoundSchema associated with the engine
- engine = metadata
- if not hasattr(engine, '_legacy_metadata'):
- engine._legacy_metadata = BoundMetaData(engine)
- metadata = engine._legacy_metadata
- elif metadata is not None and not isinstance(metadata, MetaData):
- # they left MetaData out, so assume its another SchemaItem, add it to *args
- args = list(args)
- args.insert(0, metadata)
- metadata = None
-
- if metadata is None:
- metadata = default_metadata
-
schema = kwargs.get('schema', None)
autoload = kwargs.pop('autoload', False)
autoload_with = kwargs.pop('autoload_with', False)
self.fullname = self.name
self.owner = kwargs.pop('owner', None)
- self._set_casing_strategy(name, kwargs)
- self._set_casing_strategy(self.schema or '', kwargs, keyname='case_sensitive_schema')
+ self._set_casing_strategy(kwargs)
+ self._set_casing_strategy(kwargs, keyname='case_sensitive_schema')
if len([k for k in kwargs if not re.match(r'^(?:%s)_' % '|'.join(databases.__all__), k)]):
raise TypeError("Invalid argument(s) for Table: %s" % repr(kwargs.keys()))
try:
return getattr(self, '_case_sensitive_schema')
except AttributeError:
- setattr(self, '_case_sensitive_schema', self._determine_case_sensitive(self.schema or '', keyname='case_sensitive_schema'))
+ setattr(self, '_case_sensitive_schema', self._determine_case_sensitive(keyname='case_sensitive_schema'))
return getattr(self, '_case_sensitive_schema')
case_sensitive_schema = property(_get_case_sensitive_schema)
self.index = kwargs.pop('index', None)
self.unique = kwargs.pop('unique', None)
self.quote = kwargs.pop('quote', False)
- self._set_casing_strategy(name, kwargs)
+ self._set_casing_strategy(kwargs)
self.onupdate = kwargs.pop('onupdate', None)
self.autoincrement = kwargs.pop('autoincrement', True)
self.constraints = util.Set()
self.increment = increment
self.optional=optional
self.quote = quote
- self._set_casing_strategy(name, kwargs)
+ self._set_casing_strategy(kwargs)
def __repr__(self):
return "Sequence(%s)" % string.join(
class MetaData(SchemaItem):
"""Represent a collection of Tables and their associated schema constructs."""
- def __init__(self, name=None, url=None, engine=None, **kwargs):
+ def __init__(self, url=None, engine=None, **kwargs):
"""create a new MetaData object.
- name
- optional name for this MetaData instance.
-
url
a string or URL instance which will be passed to create_engine(),
along with \**kwargs - this MetaData will be bound to the resulting
"""
self.tables = {}
- self.name = name
self._engine = None
- self._set_casing_strategy(name, kwargs)
+ self._set_casing_strategy(kwargs)
if engine or url:
self.connect(engine or url, **kwargs)
"""
- def __init__(self, engine_or_url, name=None, **kwargs):
+ def __init__(self, engine_or_url, **kwargs):
from sqlalchemy.engine.url import URL
if isinstance(engine_or_url, basestring) or isinstance(engine_or_url, URL):
- super(BoundMetaData, self).__init__(name=name, url=engine_or_url, **kwargs)
+ super(BoundMetaData, self).__init__(url=engine_or_url, **kwargs)
else:
- super(BoundMetaData, self).__init__(name=name, engine=engine_or_url, **kwargs)
+ super(BoundMetaData, self).__init__(engine=engine_or_url, **kwargs)
class DynamicMetaData(MetaData):
thread-local basis.
"""
- def __init__(self, name=None, threadlocal=True, **kwargs):
+ def __init__(self, threadlocal=True, **kwargs):
if threadlocal:
self.context = util.ThreadLocal()
else:
self.__engines[engine_or_url] = e
self.context._engine = e
else:
+ # TODO: this is squirrely. we shouldnt have to hold onto engines
+ # in a case like this
if not self.__engines.has_key(engine_or_url):
self.__engines[engine_or_url] = engine_or_url
self.context._engine = engine_or_url
"""Visit a ``CheckConstraint`` on a ``Column``."""
pass
-default_metadata = DynamicMetaData('default')
+
from sqlalchemy import *
class MetaDataTest(testbase.PersistTest):
- def test_global_metadata(self):
- t1 = Table('table1', Column('col1', Integer, primary_key=True),
- Column('col2', String(20)))
- t2 = Table('table2', Column('col1', Integer, primary_key=True),
- Column('col2', String(20)))
-
- assert t1.c.col1
- global_connect(testbase.db)
- default_metadata.create_all()
- try:
- assert t1.count().scalar() == 0
- finally:
- default_metadata.drop_all()
- default_metadata.clear()
-
def test_metadata_connect(self):
metadata = MetaData()
t1 = Table('table1', metadata, Column('col1', Integer, primary_key=True),
def testtometadata(self):
- meta = MetaData('md1')
- meta2 = MetaData('md2')
+ meta = MetaData()
+ meta2 = MetaData()
table = Table('mytable', meta,
Column('myid', Integer, primary_key=True),
def test_nonexistent(self):
self.assertRaises(NoSuchTableError, Table,
'fake_table',
- testbase.db, autoload=True)
+ BoundMetaData(testbase.db), autoload=True)
def testoverride(self):
meta = BoundMetaData(testbase.db)
class CaseTest(testbase.PersistTest):
def setUpAll(self):
+ metadata = BoundMetaData(testbase.db)
global info_table
- info_table = Table('infos', testbase.db,
+ info_table = Table('infos', metadata,
Column('pk', Integer, primary_key=True),
Column('info', String(30)))
class DefaultTest(PersistTest):
def setUpAll(self):
- global t, f, f2, ts, currenttime
+ global t, f, f2, ts, currenttime, metadata
+ metadata = BoundMetaData(testbase.db)
x = {'x':50}
def mydefault():
x['x'] += 1
ts = 3
deftype = Integer
- t = Table('default_test1', db,
+ t = Table('default_test1', metadata,
# python function
Column('col1', Integer, primary_key=True, default=mydefault),
\r
from sqlalchemy import *\r
\r
-\r
-table = Table('table1', db, \r
+metadata = BoundMetaData(db)\r
+table = Table('table1', metadata, \r
Column('col1', Integer, primary_key=True),\r
Column('col2', String(20)),\r
Column('col3', Integer),\r
\r
)\r
\r
-table2 = Table('table2', db,\r
+table2 = Table('table2', metadata,\r
Column('col1', Integer, primary_key=True),\r
Column('col2', Integer, ForeignKey('table1.col1')),\r
Column('col3', String(20)),\r
def setUpAll(self):
global users
- users = Table('type_users', db,
+ users = Table('type_users', BoundMetaData(db),
Column('user_id', Integer, primary_key = True),
# totall custom type
Column('goofy', MyType, nullable = False),
expectedResults['float_column'] = 'float_column FLOAT(25)'
print db.engine.__module__
- testTable = Table('testColumns', db,
+ testTable = Table('testColumns', BoundMetaData(db),
Column('int_column', Integer),
Column('smallint_column', Smallinteger),
Column('varchar_column', String(20)),
"""tests the Unicode type. also tests the TypeDecorator with instances in the types package."""
def setUpAll(self):
global unicode_table
- unicode_table = Table('unicode_table', db,
+ metadata = BoundMetaData(db)
+ unicode_table = Table('unicode_table', metadata,
Column('id', Integer, Sequence('uni_id_seq', optional=True), primary_key=True),
Column('unicode_data', Unicode(250)),
Column('plain_data', String(250))
class BinaryTest(AssertMixin):
def setUpAll(self):
global binary_table
- binary_table = Table('binary_table', db,
+ binary_table = Table('binary_table', BoundMetaData(db),
Column('primary_id', Integer, Sequence('binary_id_seq', optional=True), primary_key=True),
Column('data', Binary),
Column('data_slice', Binary(100)),
collist = [Column('user_id', INT, primary_key = True), Column('user_name', VARCHAR(20)), Column('user_datetime', DateTime(timezone=False)),
Column('user_date', Date), Column('user_time', Time)]
- users_with_date = Table('query_users_with_date', db, *collist)
+ users_with_date = Table('query_users_with_date', BoundMetaData(db), *collist)
users_with_date.create()
insert_dicts = [dict(zip(fnames, d)) for d in insert_data]
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
