We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Purging deleted packages does not delete associated filestore/datastore resources
**CKAN version**
2.9

**Describe the bug**
After deleting and purging dataset packages, the associated datastore/filestore resources are orphaned and not deleted.

**Steps to reproduce**

1. Delete datasets
2. Go to sysadmin/trash
3. Select "Purge All"
4. Check datastore and filestore.  Associated resources are still there.

**Expected behavior**
Dataset resources in the filestore and the datastore are deleted when the associated package resources are purged.

**Additional details**
This is related to #4705 and a lot of work has been done on PRs #4867 and #4905, but neither have been merged.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ckanext/datastore/cli.py`
Content:
```
1 # encoding: utf-8
2 
3 import logging
4 import os
5 
6 import click
7 
8 from ckan.model import parse_db_config
9 from ckan.common import config
10 
11 import ckanext.datastore as datastore_module
12 from ckanext.datastore.backend.postgres import identifier
13 from ckanext.datastore.blueprint import DUMP_FORMATS, dump_to
14 
15 log = logging.getLogger(__name__)
16 
17 
18 @click.group()
19 def datastore():
20     u'''Perform commands to set up the datastore.
21     '''
22 
23 
24 @datastore.command(
25     u'set-permissions',
26     short_help=u'Generate SQL for permission configuration.'
27 )
28 def set_permissions():
29     u'''Emit an SQL script that will set the permissions for the datastore
30     users as configured in your configuration file.'''
31 
32     write_url = _parse_db_config(u'ckan.datastore.write_url')
33     read_url = _parse_db_config(u'ckan.datastore.read_url')
34     db_url = _parse_db_config(u'sqlalchemy.url')
35 
36     # Basic validation that read and write URLs reference the same database.
37     # This obviously doesn't check they're the same database (the hosts/ports
38     # could be different), but it's better than nothing, I guess.
39 
40     if write_url[u'db_name'] != read_url[u'db_name']:
41         click.secho(
42             u'The datastore write_url and read_url must refer to the same '
43             u'database!',
44             fg=u'red',
45             bold=True
46         )
47         raise click.Abort()
48 
49     sql = permissions_sql(
50         maindb=db_url[u'db_name'],
51         datastoredb=write_url[u'db_name'],
52         mainuser=db_url[u'db_user'],
53         writeuser=write_url[u'db_user'],
54         readuser=read_url[u'db_user']
55     )
56 
57     click.echo(sql)
58 
59 
60 def permissions_sql(maindb, datastoredb, mainuser, writeuser, readuser):
61     template_filename = os.path.join(
62         os.path.dirname(datastore_module.__file__), u'set_permissions.sql'
63     )
64     with open(template_filename) as fp:
65         template = fp.read()
66     return template.format(
67         maindb=identifier(maindb),
68         datastoredb=identifier(datastoredb),
69         mainuser=identifier(mainuser),
70         writeuser=identifier(writeuser),
71         readuser=identifier(readuser)
72     )
73 
74 
75 @datastore.command()
76 @click.argument(u'resource-id', nargs=1)
77 @click.argument(
78     u'output-file',
79     type=click.File(u'wb'),
80     default=click.get_binary_stream(u'stdout')
81 )
82 @click.option(u'--format', default=u'csv', type=click.Choice(DUMP_FORMATS))
83 @click.option(u'--offset', type=click.IntRange(0, None), default=0)
84 @click.option(u'--limit', type=click.IntRange(0))
85 @click.option(u'--bom', is_flag=True)  # FIXME: options based on format
86 @click.pass_context
87 def dump(ctx, resource_id, output_file, format, offset, limit, bom):
88     u'''Dump a datastore resource.
89     '''
90     flask_app = ctx.meta['flask_app']
91     with flask_app.test_request_context():
92         dump_to(
93             resource_id,
94             output_file,
95             fmt=format,
96             offset=offset,
97             limit=limit,
98             options={u'bom': bom},
99             sort=u'_id',
100             search_params={}
101         )
102 
103 
104 def _parse_db_config(config_key=u'sqlalchemy.url'):
105     db_config = parse_db_config(config_key)
106     if not db_config:
107         click.secho(
108             u'Could not extract db details from url: %r' % config[config_key],
109             fg=u'red',
110             bold=True
111         )
112         raise click.Abort()
113     return db_config
114
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ckanext/datastore/cli.py b/ckanext/datastore/cli.py
--- a/ckanext/datastore/cli.py
+++ b/ckanext/datastore/cli.py
@@ -7,6 +7,7 @@
 
 from ckan.model import parse_db_config
 from ckan.common import config
+import ckan.logic as logic
 
 import ckanext.datastore as datastore_module
 from ckanext.datastore.backend.postgres import identifier
@@ -111,3 +112,69 @@
         )
         raise click.Abort()
     return db_config
+
+
+@datastore.command(
+    u'purge',
+    short_help=u'purge orphaned resources from the datastore.'
+)
+def purge():
+    u'''Purge orphaned resources from the datastore using the datastore_delete
+    action, which drops tables when called without filters.'''
+
+    site_user = logic.get_action(u'get_site_user')({u'ignore_auth': True}, {})
+    context = {u'user': site_user[u'name']}
+
+    result = logic.get_action(u'datastore_search')(
+        context,
+        {u'resource_id': u'_table_metadata'}
+    )
+
+    resource_id_list = []
+    for record in result[u'records']:
+        try:
+            # ignore 'alias' records (views) as they are automatically
+            # deleted when the parent resource table is dropped
+            if record[u'alias_of']:
+                continue
+
+            # we need to do this to trigger resource_show auth function
+            site_user = logic.get_action(u'get_site_user')(
+                {u'ignore_auth': True}, {})
+            context = {u'user': site_user[u'name']}
+
+            logic.get_action(u'resource_show')(
+                context,
+                {u'id': record[u'name']}
+            )
+        except logic.NotFound:
+            resource_id_list.append(record[u'name'])
+            click.echo(u"Resource '%s' orphaned - queued for drop" %
+                       record[u'name'])
+        except KeyError:
+            continue
+
+    orphaned_table_count = len(resource_id_list)
+    click.echo(u'%d orphaned tables found.' % orphaned_table_count)
+
+    if not orphaned_table_count:
+        return
+
+    click.confirm(u'Proceed with purge?', abort=True)
+
+    # Drop the orphaned datastore tables. When datastore_delete is called
+    # without filters, it does a drop table cascade
+    drop_count = 0
+    for resource_id in resource_id_list:
+        logic.get_action(u'datastore_delete')(
+            context,
+            {u'resource_id': resource_id, u'force': True}
+        )
+        click.echo(u"Table '%s' dropped)" % resource_id)
+        drop_count += 1
+
+    click.echo(u'Dropped %s tables' % drop_count)
+
+
+def get_commands():
+    return (set_permissions, dump, purge)

{"golden_diff": "diff --git a/ckanext/datastore/cli.py b/ckanext/datastore/cli.py\n--- a/ckanext/datastore/cli.py\n+++ b/ckanext/datastore/cli.py\n@@ -7,6 +7,7 @@\n \n from ckan.model import parse_db_config\n from ckan.common import config\n+import ckan.logic as logic\n \n import ckanext.datastore as datastore_module\n from ckanext.datastore.backend.postgres import identifier\n@@ -111,3 +112,69 @@\n         )\n         raise click.Abort()\n     return db_config\n+\n+\n+@datastore.command(\n+    u'purge',\n+    short_help=u'purge orphaned resources from the datastore.'\n+)\n+def purge():\n+    u'''Purge orphaned resources from the datastore using the datastore_delete\n+    action, which drops tables when called without filters.'''\n+\n+    site_user = logic.get_action(u'get_site_user')({u'ignore_auth': True}, {})\n+    context = {u'user': site_user[u'name']}\n+\n+    result = logic.get_action(u'datastore_search')(\n+        context,\n+        {u'resource_id': u'_table_metadata'}\n+    )\n+\n+    resource_id_list = []\n+    for record in result[u'records']:\n+        try:\n+            # ignore 'alias' records (views) as they are automatically\n+            # deleted when the parent resource table is dropped\n+            if record[u'alias_of']:\n+                continue\n+\n+            # we need to do this to trigger resource_show auth function\n+            site_user = logic.get_action(u'get_site_user')(\n+                {u'ignore_auth': True}, {})\n+            context = {u'user': site_user[u'name']}\n+\n+            logic.get_action(u'resource_show')(\n+                context,\n+                {u'id': record[u'name']}\n+            )\n+        except logic.NotFound:\n+            resource_id_list.append(record[u'name'])\n+            click.echo(u\"Resource '%s' orphaned - queued for drop\" %\n+                       record[u'name'])\n+        except KeyError:\n+            continue\n+\n+    orphaned_table_count = len(resource_id_list)\n+    click.echo(u'%d orphaned tables found.' % orphaned_table_count)\n+\n+    if not orphaned_table_count:\n+        return\n+\n+    click.confirm(u'Proceed with purge?', abort=True)\n+\n+    # Drop the orphaned datastore tables. When datastore_delete is called\n+    # without filters, it does a drop table cascade\n+    drop_count = 0\n+    for resource_id in resource_id_list:\n+        logic.get_action(u'datastore_delete')(\n+            context,\n+            {u'resource_id': resource_id, u'force': True}\n+        )\n+        click.echo(u\"Table '%s' dropped)\" % resource_id)\n+        drop_count += 1\n+\n+    click.echo(u'Dropped %s tables' % drop_count)\n+\n+\n+def get_commands():\n+    return (set_permissions, dump, purge)\n", "issue": "Purging deleted packages does not delete associated filestore/datastore resources\n**CKAN version**\r\n2.9\r\n\r\n**Describe the bug**\r\nAfter deleting and purging dataset packages, the associated datastore/filestore resources are orphaned and not deleted.\r\n\r\n**Steps to reproduce**\r\n\r\n1. Delete datasets\r\n2. Go to sysadmin/trash\r\n3. Select \"Purge All\"\r\n4. Check datastore and filestore.  Associated resources are still there.\r\n\r\n**Expected behavior**\r\nDataset resources in the filestore and the datastore are deleted when the associated package resources are purged.\r\n\r\n**Additional details**\r\nThis is related to #4705 and a lot of work has been done on PRs #4867 and #4905, but neither have been merged.\r\n\n", "before_files": [{"content": "# encoding: utf-8\n\nimport logging\nimport os\n\nimport click\n\nfrom ckan.model import parse_db_config\nfrom ckan.common import config\n\nimport ckanext.datastore as datastore_module\nfrom ckanext.datastore.backend.postgres import identifier\nfrom ckanext.datastore.blueprint import DUMP_FORMATS, dump_to\n\nlog = logging.getLogger(__name__)\n\n\n@click.group()\ndef datastore():\n    u'''Perform commands to set up the datastore.\n    '''\n\n\n@datastore.command(\n    u'set-permissions',\n    short_help=u'Generate SQL for permission configuration.'\n)\ndef set_permissions():\n    u'''Emit an SQL script that will set the permissions for the datastore\n    users as configured in your configuration file.'''\n\n    write_url = _parse_db_config(u'ckan.datastore.write_url')\n    read_url = _parse_db_config(u'ckan.datastore.read_url')\n    db_url = _parse_db_config(u'sqlalchemy.url')\n\n    # Basic validation that read and write URLs reference the same database.\n    # This obviously doesn't check they're the same database (the hosts/ports\n    # could be different), but it's better than nothing, I guess.\n\n    if write_url[u'db_name'] != read_url[u'db_name']:\n        click.secho(\n            u'The datastore write_url and read_url must refer to the same '\n            u'database!',\n            fg=u'red',\n            bold=True\n        )\n        raise click.Abort()\n\n    sql = permissions_sql(\n        maindb=db_url[u'db_name'],\n        datastoredb=write_url[u'db_name'],\n        mainuser=db_url[u'db_user'],\n        writeuser=write_url[u'db_user'],\n        readuser=read_url[u'db_user']\n    )\n\n    click.echo(sql)\n\n\ndef permissions_sql(maindb, datastoredb, mainuser, writeuser, readuser):\n    template_filename = os.path.join(\n        os.path.dirname(datastore_module.__file__), u'set_permissions.sql'\n    )\n    with open(template_filename) as fp:\n        template = fp.read()\n    return template.format(\n        maindb=identifier(maindb),\n        datastoredb=identifier(datastoredb),\n        mainuser=identifier(mainuser),\n        writeuser=identifier(writeuser),\n        readuser=identifier(readuser)\n    )\n\n\n@datastore.command()\n@click.argument(u'resource-id', nargs=1)\n@click.argument(\n    u'output-file',\n    type=click.File(u'wb'),\n    default=click.get_binary_stream(u'stdout')\n)\n@click.option(u'--format', default=u'csv', type=click.Choice(DUMP_FORMATS))\n@click.option(u'--offset', type=click.IntRange(0, None), default=0)\n@click.option(u'--limit', type=click.IntRange(0))\n@click.option(u'--bom', is_flag=True)  # FIXME: options based on format\n@click.pass_context\ndef dump(ctx, resource_id, output_file, format, offset, limit, bom):\n    u'''Dump a datastore resource.\n    '''\n    flask_app = ctx.meta['flask_app']\n    with flask_app.test_request_context():\n        dump_to(\n            resource_id,\n            output_file,\n            fmt=format,\n            offset=offset,\n            limit=limit,\n            options={u'bom': bom},\n            sort=u'_id',\n            search_params={}\n        )\n\n\ndef _parse_db_config(config_key=u'sqlalchemy.url'):\n    db_config = parse_db_config(config_key)\n    if not db_config:\n        click.secho(\n            u'Could not extract db details from url: %r' % config[config_key],\n            fg=u'red',\n            bold=True\n        )\n        raise click.Abort()\n    return db_config\n", "path": "ckanext/datastore/cli.py"}], "after_files": [{"content": "# encoding: utf-8\n\nimport logging\nimport os\n\nimport click\n\nfrom ckan.model import parse_db_config\nfrom ckan.common import config\nimport ckan.logic as logic\n\nimport ckanext.datastore as datastore_module\nfrom ckanext.datastore.backend.postgres import identifier\nfrom ckanext.datastore.blueprint import DUMP_FORMATS, dump_to\n\nlog = logging.getLogger(__name__)\n\n\n@click.group()\ndef datastore():\n    u'''Perform commands to set up the datastore.\n    '''\n\n\n@datastore.command(\n    u'set-permissions',\n    short_help=u'Generate SQL for permission configuration.'\n)\ndef set_permissions():\n    u'''Emit an SQL script that will set the permissions for the datastore\n    users as configured in your configuration file.'''\n\n    write_url = _parse_db_config(u'ckan.datastore.write_url')\n    read_url = _parse_db_config(u'ckan.datastore.read_url')\n    db_url = _parse_db_config(u'sqlalchemy.url')\n\n    # Basic validation that read and write URLs reference the same database.\n    # This obviously doesn't check they're the same database (the hosts/ports\n    # could be different), but it's better than nothing, I guess.\n\n    if write_url[u'db_name'] != read_url[u'db_name']:\n        click.secho(\n            u'The datastore write_url and read_url must refer to the same '\n            u'database!',\n            fg=u'red',\n            bold=True\n        )\n        raise click.Abort()\n\n    sql = permissions_sql(\n        maindb=db_url[u'db_name'],\n        datastoredb=write_url[u'db_name'],\n        mainuser=db_url[u'db_user'],\n        writeuser=write_url[u'db_user'],\n        readuser=read_url[u'db_user']\n    )\n\n    click.echo(sql)\n\n\ndef permissions_sql(maindb, datastoredb, mainuser, writeuser, readuser):\n    template_filename = os.path.join(\n        os.path.dirname(datastore_module.__file__), u'set_permissions.sql'\n    )\n    with open(template_filename) as fp:\n        template = fp.read()\n    return template.format(\n        maindb=identifier(maindb),\n        datastoredb=identifier(datastoredb),\n        mainuser=identifier(mainuser),\n        writeuser=identifier(writeuser),\n        readuser=identifier(readuser)\n    )\n\n\n@datastore.command()\n@click.argument(u'resource-id', nargs=1)\n@click.argument(\n    u'output-file',\n    type=click.File(u'wb'),\n    default=click.get_binary_stream(u'stdout')\n)\n@click.option(u'--format', default=u'csv', type=click.Choice(DUMP_FORMATS))\n@click.option(u'--offset', type=click.IntRange(0, None), default=0)\n@click.option(u'--limit', type=click.IntRange(0))\n@click.option(u'--bom', is_flag=True)  # FIXME: options based on format\n@click.pass_context\ndef dump(ctx, resource_id, output_file, format, offset, limit, bom):\n    u'''Dump a datastore resource.\n    '''\n    flask_app = ctx.meta['flask_app']\n    with flask_app.test_request_context():\n        dump_to(\n            resource_id,\n            output_file,\n            fmt=format,\n            offset=offset,\n            limit=limit,\n            options={u'bom': bom},\n            sort=u'_id',\n            search_params={}\n        )\n\n\ndef _parse_db_config(config_key=u'sqlalchemy.url'):\n    db_config = parse_db_config(config_key)\n    if not db_config:\n        click.secho(\n            u'Could not extract db details from url: %r' % config[config_key],\n            fg=u'red',\n            bold=True\n        )\n        raise click.Abort()\n    return db_config\n\n\n@datastore.command(\n    u'purge',\n    short_help=u'purge orphaned resources from the datastore.'\n)\ndef purge():\n    u'''Purge orphaned resources from the datastore using the datastore_delete\n    action, which drops tables when called without filters.'''\n\n    site_user = logic.get_action(u'get_site_user')({u'ignore_auth': True}, {})\n    context = {u'user': site_user[u'name']}\n\n    result = logic.get_action(u'datastore_search')(\n        context,\n        {u'resource_id': u'_table_metadata'}\n    )\n\n    resource_id_list = []\n    for record in result[u'records']:\n        try:\n            # ignore 'alias' records (views) as they are automatically\n            # deleted when the parent resource table is dropped\n            if record[u'alias_of']:\n                continue\n\n            # we need to do this to trigger resource_show auth function\n            site_user = logic.get_action(u'get_site_user')(\n                {u'ignore_auth': True}, {})\n            context = {u'user': site_user[u'name']}\n\n            logic.get_action(u'resource_show')(\n                context,\n                {u'id': record[u'name']}\n            )\n        except logic.NotFound:\n            resource_id_list.append(record[u'name'])\n            click.echo(u\"Resource '%s' orphaned - queued for drop\" %\n                       record[u'name'])\n        except KeyError:\n            continue\n\n    orphaned_table_count = len(resource_id_list)\n    click.echo(u'%d orphaned tables found.' % orphaned_table_count)\n\n    if not orphaned_table_count:\n        return\n\n    click.confirm(u'Proceed with purge?', abort=True)\n\n    # Drop the orphaned datastore tables. When datastore_delete is called\n    # without filters, it does a drop table cascade\n    drop_count = 0\n    for resource_id in resource_id_list:\n        logic.get_action(u'datastore_delete')(\n            context,\n            {u'resource_id': resource_id, u'force': True}\n        )\n        click.echo(u\"Table '%s' dropped)\" % resource_id)\n        drop_count += 1\n\n    click.echo(u'Dropped %s tables' % drop_count)\n\n\ndef get_commands():\n    return (set_permissions, dump, purge)\n", "path": "ckanext/datastore/cli.py"}]}