From: Mike Bayer Date: Sat, 30 Jun 2007 00:32:11 +0000 (+0000) Subject: merged pickleable schema items from trunk r2817 X-Git-Tag: rel_0_4_6~154 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=b0631ad3bde92145acbff833c2463b53b2ee7af8;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git merged pickleable schema items from trunk r2817 --- diff --git a/CHANGES b/CHANGES index 2bb9414053..b40adbe94b 100644 --- a/CHANGES +++ b/CHANGES @@ -114,6 +114,9 @@ to polymorphic mappers that are using a straight "outerjoin" clause - sql + - MetaData and all SchemaItems are safe to use with pickle. slow + table reflections can be dumped into a pickled file to be reused later. + Just reconnect the engine to the metadata after unpickling. [ticket:619] - fixed grouping of compound selects to give correct results. will break on sqlite in some cases, but those cases were producing incorrect results anyway, sqlite doesn't support grouped compound selects diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py index 713064d9cc..f67278d7c6 100644 --- a/lib/sqlalchemy/schema.py +++ b/lib/sqlalchemy/schema.py @@ -1066,6 +1066,14 @@ class MetaData(SchemaItem): if engine or url: self.connect(engine or url, **kwargs) + def __getstate__(self): + return {'tables':self.tables, 'casesensitive':self._case_sensitive_setting} + + def __setstate__(self, state): + self.tables = state['tables'] + self._case_sensitive_setting = state['casesensitive'] + self._engine = None + def is_bound(self): """return True if this MetaData is bound to an Engine.""" return self._engine is not None diff --git a/lib/sqlalchemy/types.py b/lib/sqlalchemy/types.py index f13a4114dc..b7f9e6e992 100644 --- a/lib/sqlalchemy/types.py +++ b/lib/sqlalchemy/types.py @@ -52,7 +52,12 @@ class TypeEngine(AbstractType): return self._impl_dict.setdefault(dialect, dialect.type_descriptor(self)) except KeyError: return self._impl_dict.setdefault(dialect, dialect.type_descriptor(self)) - + + def __getstate__(self): + d = self.__dict__.copy() + d['_impl_dict'] = {} + return d + def get_col_spec(self): raise NotImplementedError() diff --git a/test/engine/reflection.py b/test/engine/reflection.py index 66d5a5f04c..1b6c73e28b 100644 --- a/test/engine/reflection.py +++ b/test/engine/reflection.py @@ -1,6 +1,6 @@ from testbase import PersistTest import testbase - +import pickle import sqlalchemy.ansisql as ansisql from sqlalchemy import * @@ -109,7 +109,7 @@ class ReflectionTest(PersistTest): finally: addresses.drop() users.drop() - + def testoverridecolumns(self): """test that you can override columns which contain foreign keys to other reflected tables""" meta = BoundMetaData(testbase.db) @@ -336,50 +336,80 @@ class ReflectionTest(PersistTest): finally: meta.drop_all() - - def testtometadata(self): + def test_to_metadata(self): meta = MetaData() - meta2 = MetaData() table = Table('mytable', meta, Column('myid', Integer, primary_key=True), - Column('name', String, nullable=False), + Column('name', String(40), nullable=False), Column('description', String(30), CheckConstraint("description='hi'")), - UniqueConstraint('name') + UniqueConstraint('name'), + mysql_engine='InnoDB' ) table2 = Table('othertable', meta, Column('id', Integer, primary_key=True), - Column('myid', Integer, ForeignKey('mytable.myid')) + Column('myid', Integer, ForeignKey('mytable.myid')), + mysql_engine='InnoDB' ) - - - table_c = table.tometadata(meta2) - table2_c = table2.tometadata(meta2) - - assert table is not table_c - assert table_c.c.myid.primary_key - assert not table_c.c.name.nullable - assert table_c.c.description.nullable - assert table.primary_key is not table_c.primary_key - assert [x.name for x in table.primary_key] == [x.name for x in table_c.primary_key] - assert list(table2_c.c.myid.foreign_keys)[0].column is table_c.c.myid - assert list(table2_c.c.myid.foreign_keys)[0].column is not table.c.myid - for c in table_c.c.description.constraints: - if isinstance(c, CheckConstraint): - break - else: - assert False - assert c.sqltext=="description='hi'" - for c in table_c.constraints: - if isinstance(c, UniqueConstraint): - break - else: - assert False - assert c.columns.contains_column(table_c.c.name) - assert not c.columns.contains_column(table.c.name) + def test_to_metadata(): + meta2 = MetaData() + table_c = table.tometadata(meta2) + table2_c = table2.tometadata(meta2) + return (table_c, table2_c) + + def test_pickle(): + meta.connect(testbase.db) + meta2 = pickle.loads(pickle.dumps(meta)) + assert meta2.engine is None + return (meta2.tables['mytable'], meta2.tables['othertable']) + + def test_pickle_via_reflect(): + # this is the most common use case, pickling the results of a + # database reflection + meta2 = MetaData(engine=testbase.db) + t1 = Table('mytable', meta2, autoload=True) + t2 = Table('othertable', meta2, autoload=True) + meta3 = pickle.loads(pickle.dumps(meta2)) + assert meta3.engine is None + assert meta3.tables['mytable'] is not t1 + return (meta3.tables['mytable'], meta3.tables['othertable']) + + meta.create_all(testbase.db) + try: + for test, has_constraints in ((test_to_metadata, True), (test_pickle, True), (test_pickle_via_reflect, False)): + table_c, table2_c = test() + assert table is not table_c + assert table_c.c.myid.primary_key + assert isinstance(table_c.c.myid.type, Integer) + assert isinstance(table_c.c.name.type, String) + assert not table_c.c.name.nullable + assert table_c.c.description.nullable + assert table.primary_key is not table_c.primary_key + assert [x.name for x in table.primary_key] == [x.name for x in table_c.primary_key] + assert list(table2_c.c.myid.foreign_keys)[0].column is table_c.c.myid + assert list(table2_c.c.myid.foreign_keys)[0].column is not table.c.myid + + # constraints dont get reflected for any dialect right now + if has_constraints: + for c in table_c.c.description.constraints: + if isinstance(c, CheckConstraint): + break + else: + assert False + assert c.sqltext=="description='hi'" + for c in table_c.constraints: + if isinstance(c, UniqueConstraint): + break + else: + assert False + assert c.columns.contains_column(table_c.c.name) + assert not c.columns.contains_column(table.c.name) + finally: + meta.drop_all(testbase.db) + # mysql throws its own exception for no such table, resulting in # a sqlalchemy.SQLError instead of sqlalchemy.NoSuchTableError. # this could probably be fixed at some point.