Skip to content
This repository
Browse code

Merge pull request #113 from tmoertel/feature/2.0-primary_foreign_keys

Add support for primary foreign keys
  • Loading branch information...
commit 8e342c0b0ac25526f9fdd0573409e122ef8d63d2 2 parents 862945f + 60698e7
Charles Leifer authored October 08, 2012

Showing 2 changed files with 38 additions and 1 deletion. Show diff stats Hide diff stats

  1. 3  peewee.py
  2. 36  tests.py
3  peewee.py
@@ -1549,7 +1549,8 @@ def create_index(self, model_class, fields, unique=False):
1549 1549
         return self.execute_sql(qc.create_index(model_class, field_objs, unique))
1550 1550
 
1551 1551
     def create_foreign_key(self, model_class, field):
1552  
-        return self.create_index(model_class, [field], field.unique)
  1552
+        if not field.primary_key:
  1553
+            return self.create_index(model_class, [field], field.unique)
1553 1554
 
1554 1555
     def create_sequence(self, seq):
1555 1556
         if self.sequences:
36  tests.py
@@ -1763,3 +1763,39 @@ def test_sequence_shared(self):
1763 1763
 
1764 1764
 elif TEST_VERBOSITY > 0:
1765 1765
     print 'Skipping "sequence" tests'
  1766
+
  1767
+
  1768
+class Job(TestModel):
  1769
+    """A job that can be queued for later execution."""
  1770
+    name = CharField()
  1771
+
  1772
+class JobExecutionRecord(TestModel):
  1773
+    """Record of a job having been executed."""
  1774
+    # the foreign key is also the primary key to enforce the
  1775
+    # constraint that a job can be executed once and only once
  1776
+    job = ForeignKeyField(Job, primary_key=True)
  1777
+    status = CharField()
  1778
+
  1779
+class PrimaryForeignKeyTestCase(unittest.TestCase):
  1780
+    def setUp(self):
  1781
+        Job.create_table()
  1782
+        JobExecutionRecord.create_table()
  1783
+
  1784
+    def tearDown(self):
  1785
+        JobExecutionRecord.drop_table()
  1786
+        Job.drop_table()
  1787
+
  1788
+    def test_primary_foreign_key(self):
  1789
+        # we have one job, unexecuted, and therefore no executed jobs
  1790
+        job = Job.create(name='Job One')
  1791
+        executed_jobs = Job.select().join(JobExecutionRecord)
  1792
+        self.assertEqual([], list(executed_jobs))
  1793
+
  1794
+        # after execution, we must have one executed job
  1795
+        exec_record = JobExecutionRecord.create(job=job, status='success')
  1796
+        executed_jobs = Job.select().join(JobExecutionRecord)
  1797
+        self.assertEqual([job], list(executed_jobs))
  1798
+
  1799
+        # we must not be able to create another execution record for the job
  1800
+        with self.assertRaises(sqlite3.IntegrityError):
  1801
+            JobExecutionRecord.create(job=job, status='success')

0 notes on commit 8e342c0

Please sign in to comment.
Something went wrong with that request. Please try again.