Merge pull request #62 from dr-rodriguez/master

Foreign Key Updates

astrodbkit/__init__.py

@@ -1,2 +1,2 @@
 from pkg_resources import get_distribution
-__version__ = '0.3.0.post7'
+__version__ = '0.4.0'

astrodbkit/astrodb.py

@@ -1136,13 +1136,20 @@ def table(self, table, columns, types, constraints='', pk='', new_table=False):
                 create_txt = "CREATE TABLE {0} ({1}".format(table, ', '.join(
                         ['{} {} {}'.format(c, t, r) for c, t, r in zip(columns, types, constraints)]))
                 create_txt += ', PRIMARY KEY({}))'.format(', '.join([elem for elem in pk]))
-                print(create_txt.replace(',', ',\n'))
+                # print(create_txt.replace(',', ',\n'))
                 self.list(create_txt)
 
                 # Populate the new table and drop the old one
                 old_columns = [c for c in self.query("PRAGMA table_info(TempOldTable)", unpack=True)[1] if c in columns]
                 self.list("INSERT INTO {0} ({1}) SELECT {1} FROM TempOldTable".format(table, ','.join(old_columns)))
-                self.list("DROP TABLE TempOldTable")
+
+                # Check for and add any foreign key constraints
+                t = self.query('PRAGMA foreign_key_list(TempOldTable)', fmt='table')
+                if not isinstance(t, type(None)):
+                    self.list("DROP TABLE TempOldTable")
+                    self.add_foreign_key(table, t['table'].tolist(), t['from'].tolist(), t['to'].tolist())
+                else:
+                    self.list("DROP TABLE TempOldTable")
 
             # If the table does not exist and new_table is True, create it
             elif table not in tables and new_table:
@@ -1204,26 +1211,26 @@ def add_foreign_key(self, table, parent, key_child, key_parent, verbose=True):
 
         try:
             # Rename the old table and create a new one
-            self.list("DROP TABLE IF EXISTS TempOldTable")
-            self.list("ALTER TABLE {0} RENAME TO TempOldTable".format(table))
+            self.list("DROP TABLE IF EXISTS TempOldTable_foreign")
+            self.list("ALTER TABLE {0} RENAME TO TempOldTable_foreign".format(table))
 
             # Re-create the table specifying the FOREIGN KEY
             sqltxt = "CREATE TABLE {0} ({1}".format(table, ', '.join(['{} {} {}'.format(c, t, r)
                                                                       for c, t, r in zip(columns, types, constraints)]))
             sqltxt += ', PRIMARY KEY({})'.format(', '.join([elem for elem in pk_names]))
             if isinstance(key_child, type(list())):
                 for kc, p, kp in zip(key_child, parent, key_parent):
-                    sqltxt += ', FOREIGN KEY ({0}) REFERENCES {1} ({2})'.format(kc, p, kp)
+                    sqltxt += ', FOREIGN KEY ({0}) REFERENCES {1} ({2}) ON UPDATE CASCADE'.format(kc, p, kp)
             else:
-                sqltxt += ', FOREIGN KEY ({0}) REFERENCES {1} ({2})'.format(key_child, parent, key_parent)
+                sqltxt += ', FOREIGN KEY ({0}) REFERENCES {1} ({2}) ON UPDATE CASCADE'.format(key_child, parent, key_parent)
             sqltxt += ' )'
 
             self.list(sqltxt)
 
             # Populate the new table and drop the old one
-            old_columns = [c for c in self.query("PRAGMA table_info(TempOldTable)", unpack=True)[1] if c in columns]
-            self.modify("INSERT INTO {0} ({1}) SELECT {1} FROM TempOldTable".format(table, ','.join(old_columns)))
-            self.list("DROP TABLE TempOldTable")
+            old_columns = [c for c in self.query("PRAGMA table_info(TempOldTable_foreign)", unpack=True)[1] if c in columns]
+            self.list("INSERT INTO {0} ({1}) SELECT {1} FROM TempOldTable_foreign".format(table, ','.join(old_columns)))
+            self.list("DROP TABLE TempOldTable_foreign")
 
             if verbose:
                 print('Successfully added foreign key.')
@@ -1233,7 +1240,7 @@ def add_foreign_key(self, table, parent, key_child, key_parent, verbose=True):
         except:
             print('Error attempting to add foreign key.')
             self.list("DROP TABLE IF EXISTS {0}".format(table))
-            self.list("ALTER TABLE TempOldTable RENAME TO {0}".format(table))
+            self.list("ALTER TABLE TempOldTable_foreign RENAME TO {0}".format(table))
 
         # Reactivate foreign keys
         self.list('PRAGMA foreign_keys=ON')

astrodbkit/tests/test_astrodb.py

@@ -3,36 +3,78 @@
 import os
 from astropy.utils.data import download_file
 from .. import astrodb
+from sqlite3 import IntegrityError
 
 
 def setup_module(module):
-    db_path = download_file("http://github.com/BDNYC/BDNYCdb/raw/master/BDNYCv1.0.db")
+    try:
+        db_path = download_file("http://github.com/BDNYC/BDNYCdb/raw/master/bdnyc_database.db")
+    except:
+        db_path = download_file("http://github.com/BDNYC/BDNYCdb/raw/master/BDNYCv1.0.db")
     module.bdnyc_db = astrodb.Database(db_path)
     filename = os.path.join(tempfile.mkdtemp(), 'empty_db.db')
-    module.empty_db = astrodb.create_database(filename)
+    astrodb.create_database(filename)
+    module.empty_db = astrodb.Database(filename)
 
-def test_loaddb():
+
+def test_load_bdnyc():
     print(bdnyc_db)
+    assert not isinstance(bdnyc_db, type(None))
+
+
+def test_load_empty():
     print(empty_db)
+    assert not isinstance(empty_db, type(None))
+
 
 def test_search(): 
     bdnyc_db.search('1234','sources')
 
+
 def test_inventory():
     bdnyc_db.inventory(825)
 
+
 def test_sqlquery():
     bdnyc_db.query("SELECT s.id, s.ra, s.dec, s.shortname, p.source_id, p.band, p.magnitude "
         "FROM sources as s JOIN photometry as p ON s.id=p.source_id "
         "WHERE s.dec<=-10 AND p.band=='W1'")
 
+
 def test_schema():
     bdnyc_db.schema('sources')
-
-@pytest.mark.xfail
+
+
+def test_table():
+    columns = ['id', 'ra', 'dec', 'shortname', 'source_id']
+    types = ['INTEGER', 'REAL', 'REAL', 'TEXT', 'INTEGER']
+    constraints = ['NOT NULL UNIQUE', '', '', '', '']
+    empty_db.table('new_sources', columns, types, constraints, new_table=True)
+
+
 def test_add_data():
-    assert False
-
+    data = list()
+    data.append(['ra', 'dec', 'shortname'])
+    data.append([12, -12, 'fakesource'])
+    empty_db.add_data(data, 'sources')
+
+
+def test_add_foreign_key():
+    empty_db.add_foreign_key('new_sources', ['sources'], ['source_id'], ['id'])
+
+
+def test_foreign_key_support():
+    data = list()
+    data.append(['ra', 'dec', 'shortname', 'source_id'])
+    data.append([12, -12, 'fakesource', 9999])
+    with pytest.raises(IntegrityError):
+        empty_db.add_data(data, 'new_sources')  # Foreign key error
+
+
+def test_lookup():
+    bdnyc_db.lookup([1, '2MASS'], 'sources')
+
+
 @pytest.mark.xfail
 def test_clean_up():
     assert False
@@ -48,7 +90,3 @@ def test_output_spectrum():
 @pytest.mark.xfail
 def test_plot_spectrum():
     assert False
-
-@pytest.mark.xfail
-def test_table():
-    assert False