diff --git a/django_celery_beat/models.py b/django_celery_beat/models.py
index 08cd904f..583d8b63 100644
--- a/django_celery_beat/models.py
+++ b/django_celery_beat/models.py
@@ -117,13 +117,13 @@ def from_schedule(cls, schedule):
         spec = {'event': schedule.event,
                 'latitude': schedule.lat,
                 'longitude': schedule.lon}
+
+        # we do not check for MultipleObjectsReturned exception here because
+        # the unique_together constraint safely prevents from duplicates
         try:
             return cls.objects.get(**spec)
         except cls.DoesNotExist:
             return cls(**spec)
-        except MultipleObjectsReturned:
-            cls.objects.filter(**spec).delete()
-            return cls(**spec)
 
     def __str__(self):
         return '{0} ({1}, {2})'.format(
@@ -183,8 +183,7 @@ def from_schedule(cls, schedule, period=SECONDS):
         except cls.DoesNotExist:
             return cls(every=every, period=period)
         except MultipleObjectsReturned:
-            cls.objects.filter(every=every, period=period).delete()
-            return cls(every=every, period=period)
+            return cls.objects.filter(every=every, period=period).first()
 
     def __str__(self):
         readable_period = None
@@ -236,8 +235,7 @@ def from_schedule(cls, schedule):
         except cls.DoesNotExist:
             return cls(**spec)
         except MultipleObjectsReturned:
-            cls.objects.filter(**spec).delete()
-            return cls(**spec)
+            return cls.objects.filter(**spec).first()
 
 
 class CrontabSchedule(models.Model):
@@ -350,8 +348,7 @@ def from_schedule(cls, schedule):
         except cls.DoesNotExist:
             return cls(**spec)
         except MultipleObjectsReturned:
-            cls.objects.filter(**spec).delete()
-            return cls(**spec)
+            return cls.objects.filter(**spec).first()
 
 
 class PeriodicTasks(models.Model):
diff --git a/t/unit/test_models.py b/t/unit/test_models.py
index 337cc9e6..627e47b5 100644
--- a/t/unit/test_models.py
+++ b/t/unit/test_models.py
@@ -7,6 +7,7 @@
 from django.db.migrations.autodetector import MigrationAutodetector
 from django.db.migrations.loader import MigrationLoader
 from django.db.migrations.questioner import NonInteractiveMigrationQuestioner
+from django.utils import timezone
 
 import timezone_field
 
@@ -14,6 +15,9 @@
 from django_celery_beat.models import (
     crontab_schedule_celery_timezone,
     SolarSchedule,
+    CrontabSchedule,
+    ClockedSchedule,
+    IntervalSchedule,
 )
 
 
@@ -63,7 +67,22 @@ def test_models_match_migrations(self):
             msg='Model changes exist that do not have a migration')
 
 
-class CrontabScheduleTestCase(TestCase):
+class TestDuplicatesMixin:
+    def _test_duplicate_schedules(self, cls, kwargs, schedule=None):
+        sched1 = cls.objects.create(**kwargs)
+        cls.objects.create(**kwargs)
+        self.assertEqual(cls.objects.filter(**kwargs).count(), 2)
+        # try to create a duplicate schedule from a celery schedule
+        if schedule is None:
+            schedule = sched1.schedule
+        sched3 = cls.from_schedule(schedule)
+        # the schedule should be the first of the 2 previous duplicates
+        self.assertEqual(sched3, sched1)
+        # and the duplicates should not be deleted !
+        self.assertEqual(cls.objects.filter(**kwargs).count(), 2)
+
+
+class CrontabScheduleTestCase(TestCase, TestDuplicatesMixin):
     FIRST_VALID_TIMEZONE = timezone_field.\
         TimeZoneField.default_choices[0][0].zone
 
@@ -74,6 +93,19 @@ def test_default_timezone_without_settings_config(self):
     def test_default_timezone_with_settings_config(self):
         assert crontab_schedule_celery_timezone() == self.FIRST_VALID_TIMEZONE
 
+    def test_duplicate_schedules(self):
+        # See: https://github.com/celery/django-celery-beat/issues/322
+        kwargs = {
+            "minute": "*",
+            "hour": "4",
+            "day_of_week": "*",
+            "day_of_month": "*",
+            "month_of_year": "*",
+            "day_of_week": "*",
+        }
+        schedule = schedules.crontab(hour="4")
+        self._test_duplicate_schedules(CrontabSchedule, kwargs, schedule)
+
 
 class SolarScheduleTestCase(TestCase):
     EVENT_CHOICES = SolarSchedule._meta.get_field("event").choices
@@ -99,3 +131,18 @@ def test_celery_solar_schedules_included_as_event_choices(self):
 
         for event_choice in event_choices_values:
             assert event_choice in schedules.solar._all_events
+
+
+class IntervalScheduleTestCase(TestCase, TestDuplicatesMixin):
+
+    def test_duplicate_schedules(self):
+        kwargs = {'every': 1, 'period': IntervalSchedule.SECONDS}
+        schedule = schedules.schedule(run_every=1.0)
+        self._test_duplicate_schedules(IntervalSchedule, kwargs, schedule)
+
+
+class ClockedScheduleTestCase(TestCase, TestDuplicatesMixin):
+
+    def test_duplicate_schedules(self):
+        kwargs = {'clocked_time': timezone.now()}
+        self._test_duplicate_schedules(ClockedSchedule, kwargs)