diff --git a/django_celery_beat/schedulers.py b/django_celery_beat/schedulers.py
index 846b97a9..c041f5b7 100644
--- a/django_celery_beat/schedulers.py
+++ b/django_celery_beat/schedulers.py
@@ -119,6 +119,15 @@ def is_due(self):
                 )
                 return schedules.schedstate(False, delay)
 
+        # EXPIRED TASK: Disable task when expired
+        if self.model.expires is not None:
+            now = self._default_now()
+            if getattr(settings, 'DJANGO_CELERY_BEAT_TZ_AWARE', True):
+                now = maybe_make_aware(self._default_now())
+
+            if now >= self.model.expires:
+                self._disable(self.model)
+
         # ONE OFF TASK: Disable one off tasks after they've ran once
         if self.model.one_off and self.model.enabled \
                 and self.model.total_run_count > 0:
diff --git a/t/unit/test_schedulers.py b/t/unit/test_schedulers.py
index d070bb45..4ce36548 100644
--- a/t/unit/test_schedulers.py
+++ b/t/unit/test_schedulers.py
@@ -291,6 +291,35 @@ def test_one_off_task(self):
         assert not isdue
         assert delay == NEVER_CHECK_TIMEOUT
 
+    def test_task_with_expires(self):
+        interval = 10
+        right_now = self.app.now()
+        one_second_later = right_now + timedelta(seconds=interval)
+        m = self.create_model_interval(schedule(timedelta(minutes=interval)),
+                                       one_off=True,
+                                       expires=one_second_later)
+        e = self.Entry(m, app=self.app)
+        isdue, delay = e.is_due()
+        assert isdue
+        assert delay == interval
+
+        m2 = self.create_model_interval(schedule(timedelta(minutes=interval)),
+                                        one_off=True,
+                                        expires=right_now)
+        e2 = self.Entry(m2, app=self.app)
+        isdue, delay = e2.is_due()
+        assert not isdue
+        assert delay == NEVER_CHECK_TIMEOUT
+
+        one_second_ago = right_now - timedelta(seconds=1)
+        m2 = self.create_model_interval(schedule(timedelta(minutes=interval)),
+                                        one_off=True,
+                                        expires=one_second_ago)
+        e2 = self.Entry(m2, app=self.app)
+        isdue, delay = e2.is_due()
+        assert not isdue
+        assert delay == NEVER_CHECK_TIMEOUT
+
 
 @pytest.mark.django_db
 class test_DatabaseSchedulerFromAppConf(SchedulerCase):