Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Add support for primary foreign keys

  • Loading branch information...
commit 60698e7e6691ae240e92a95eec8f6f9c63970793 1 parent 7af7076
Tom Moertel authored
Showing with 38 additions and 1 deletion.
  1. +2 −1  peewee.py
  2. +36 −0 tests.py
View
3  peewee.py
@@ -1549,7 +1549,8 @@ def create_index(self, model_class, fields, unique=False):
return self.execute_sql(qc.create_index(model_class, field_objs, unique))
def create_foreign_key(self, model_class, field):
- return self.create_index(model_class, [field], field.unique)
+ if not field.primary_key:
+ return self.create_index(model_class, [field], field.unique)
def create_sequence(self, seq):
if self.sequences:
View
36 tests.py
@@ -1763,3 +1763,39 @@ def test_sequence_shared(self):
elif TEST_VERBOSITY > 0:
print 'Skipping "sequence" tests'
+
+
+class Job(TestModel):
+ """A job that can be queued for later execution."""
+ name = CharField()
+
+class JobExecutionRecord(TestModel):
+ """Record of a job having been executed."""
+ # the foreign key is also the primary key to enforce the
+ # constraint that a job can be executed once and only once
+ job = ForeignKeyField(Job, primary_key=True)
+ status = CharField()
+
+class PrimaryForeignKeyTestCase(unittest.TestCase):
+ def setUp(self):
+ Job.create_table()
+ JobExecutionRecord.create_table()
+
+ def tearDown(self):
+ JobExecutionRecord.drop_table()
+ Job.drop_table()
+
+ def test_primary_foreign_key(self):
+ # we have one job, unexecuted, and therefore no executed jobs
+ job = Job.create(name='Job One')
+ executed_jobs = Job.select().join(JobExecutionRecord)
+ self.assertEqual([], list(executed_jobs))
+
+ # after execution, we must have one executed job
+ exec_record = JobExecutionRecord.create(job=job, status='success')
+ executed_jobs = Job.select().join(JobExecutionRecord)
+ self.assertEqual([job], list(executed_jobs))
+
+ # we must not be able to create another execution record for the job
+ with self.assertRaises(sqlite3.IntegrityError):
+ JobExecutionRecord.create(job=job, status='success')
Please sign in to comment.
Something went wrong with that request. Please try again.