diff --git a/changes/218.canada.changes b/changes/218.canada.changes
new file mode 100644
index 00000000000..b0c4fe1baeb
--- /dev/null
+++ b/changes/218.canada.changes
@@ -0,0 +1 @@
+Added `--list` and `--yes` options to the `datastore purge` command. The command now handled soft-deleted Resources and Datasets.
\ No newline at end of file
diff --git a/ckanext/datastore/cli.py b/ckanext/datastore/cli.py
index d0f0499be99..2748ff79680 100644
--- a/ckanext/datastore/cli.py
+++ b/ckanext/datastore/cli.py
@@ -129,10 +129,17 @@ def _parse_db_config(config_key: str = u'sqlalchemy.url'):
 
 @datastore.command(
     'purge',
-    short_help='purge orphaned resources from the datastore.'
+    short_help='purge orphaned or deleted resources from the datastore.'
 )
-def purge():
-    '''Purge orphaned resources from the datastore using the datastore_delete
+# (canada fork only): more options and state=deleted handling
+# TODO: upstream contrib!!
+@click.option('-l', '--list', is_flag=True,
+              type=click.BOOL,
+              help='Only output the list of oprhaned or deleted Resource IDs.')
+@click.option('-y', '--yes', is_flag=True,
+              type=click.BOOL, help='Purge without asking for confirmation.')
+def purge(list: bool = False, yes: bool = False):
+    '''Purge orphaned or deleted resources from the datastore using the datastore_delete
     action, which drops tables when called without filters.'''
 
     site_user = logic.get_action('get_site_user')({'ignore_auth': True}, {})
@@ -150,16 +157,42 @@ def purge():
             if record['alias_of']:
                 continue
 
-            logic.get_action('resource_show')(
+            # (canada fork only): more options and state=deleted handling
+            # TODO: upstream contrib!!
+            res = logic.get_action('resource_show')(
                 {'user': site_user['name']},
                 {'id': record['name']}
             )
+            pkg = logic.get_action('package_show')(
+                {'user': site_user['name']},
+                {'id': res['package_id']}
+            )
         except logic.NotFound:
             resource_id_list.append(record['name'])
-            click.echo("Resource '%s' orphaned - queued for drop" %
-                       record[u'name'])
+            if not list:
+                click.echo("Resource '%s' orphaned - queued for drop" %
+                           record['name'])
+            continue
         except KeyError:
             continue
+        # (canada fork only): more options and state=deleted handling
+        # TODO: upstream contrib!!
+        if res['state'] == 'deleted':
+            resource_id_list.append(record['name'])
+            if not list:
+                click.echo("Resource '%s' deleted - queued for drop" %
+                           record['name'])
+        if pkg['state'] == 'deleted':
+            resource_id_list.append(record['name'])
+            if not list:
+                click.echo("Package '%s' deleted - queued for drop" %
+                           pkg['id'])
+
+    # (canada fork only): more options and state=deleted handling
+    # TODO: upstream contrib!!
+    if list:
+        click.echo('\n'.join(resource_id_list))
+        return
 
     orphaned_table_count = len(resource_id_list)
     click.echo('%d orphaned tables found.' % orphaned_table_count)
@@ -167,7 +200,10 @@ def purge():
     if not orphaned_table_count:
         return
 
-    click.confirm('Proceed with purge?', abort=True)
+    # (canada fork only): more options and state=deleted handling
+    # TODO: upstream contrib!!
+    if not yes:
+        click.confirm('Proceed with purge?', abort=True)
 
     # Drop the orphaned datastore tables. When datastore_delete is called
     # without filters, it does a drop table cascade