We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Remove the ability to call non-pulp-related functions as task targets


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `pulpcore/tasking/tasks.py`
Content:
```
1 import asyncio
2 import contextlib
3 import contextvars
4 import importlib
5 import logging
6 import sys
7 import traceback
8 from gettext import gettext as _
9 
10 from django.db import connection, transaction
11 from django.db.models import Model
12 from django_guid import get_guid
13 from pulpcore.app.apps import MODULE_PLUGIN_VERSIONS
14 from pulpcore.app.loggers import deprecation_logger
15 from pulpcore.app.models import Task
16 from pulpcore.app.util import current_task, get_domain, get_url
17 from pulpcore.constants import TASK_FINAL_STATES, TASK_INCOMPLETE_STATES, TASK_STATES
18 
19 _logger = logging.getLogger(__name__)
20 
21 
22 def _validate_and_get_resources(resources):
23     resource_set = set()
24     for r in resources:
25         if isinstance(r, str):
26             resource_set.add(r)
27         elif isinstance(r, Model):
28             resource_set.add(get_url(r))
29         elif r is None:
30             # Silently drop None values
31             pass
32         else:
33             raise ValueError(_("Must be (str|Model)"))
34     return list(resource_set)
35 
36 
37 def wakeup_worker():
38     # Notify workers
39     with connection.connection.cursor() as cursor:
40         cursor.execute("NOTIFY pulp_worker_wakeup")
41 
42 
43 def execute_task(task):
44     # This extra stack is needed to isolate the current_task ContextVar
45     contextvars.copy_context().run(_execute_task, task)
46 
47 
48 def _execute_task(task):
49     # Store the task id in the context for `Task.current()`.
50     current_task.set(task)
51     task.set_running()
52     try:
53         _logger.info(_("Starting task %s"), task.pk)
54 
55         # Execute task
56         module_name, function_name = task.name.rsplit(".", 1)
57         module = importlib.import_module(module_name)
58         func = getattr(module, function_name)
59         args = task.enc_args or ()
60         kwargs = task.enc_kwargs or {}
61         result = func(*args, **kwargs)
62         if asyncio.iscoroutine(result):
63             _logger.debug(_("Task is coroutine %s"), task.pk)
64             loop = asyncio.get_event_loop()
65             loop.run_until_complete(result)
66 
67     except Exception:
68         exc_type, exc, tb = sys.exc_info()
69         task.set_failed(exc, tb)
70         _logger.info(_("Task %s failed (%s)"), task.pk, exc)
71         _logger.info("\n".join(traceback.format_list(traceback.extract_tb(tb))))
72     else:
73         task.set_completed()
74         _logger.info(_("Task completed %s"), task.pk)
75 
76 
77 def dispatch(
78     func,
79     args=None,
80     kwargs=None,
81     task_group=None,
82     exclusive_resources=None,
83     shared_resources=None,
84     immediate=False,
85     deferred=True,
86     versions=None,
87 ):
88     """
89     Enqueue a message to Pulp workers with a reservation.
90 
91     This method provides normal enqueue functionality, while also requesting necessary locks for
92     serialized urls. No two tasks that claim the same resource can execute concurrently. It
93     accepts resources which it transforms into a list of urls (one for each resource).
94 
95     This method creates a :class:`pulpcore.app.models.Task` object and returns it.
96 
97     The values in `args` and `kwargs` must be JSON serializable, but may contain instances of
98     ``uuid.UUID``.
99 
100     Args:
101         func (callable | str): The function to be run when the necessary locks are acquired.
102         args (tuple): The positional arguments to pass on to the task.
103         kwargs (dict): The keyword arguments to pass on to the task.
104         task_group (pulpcore.app.models.TaskGroup): A TaskGroup to add the created Task to.
105         exclusive_resources (list): A list of resources this task needs exclusive access to while
106             running. Each resource can be either a `str` or a `django.models.Model` instance.
107         shared_resources (list): A list of resources this task needs non-exclusive access to while
108             running. Each resource can be either a `str` or a `django.models.Model` instance.
109         immediate (bool): Whether to allow running this task immediately. It must be guaranteed to
110             execute fast without blocking. If not all resource constraints are met, the task will
111             either be returned in a canceled state or, if `deferred` is `True` be left in the queue
112             to be picked up by a worker eventually. Defaults to `False`.
113         deferred (bool): Whether to allow defer running the task to a pulpcore_worker. Defaults to
114             `True`. `immediate` and `deferred` cannot both be `False`.
115         versions (Optional[Dict[str, str]]): Minimum versions of components by app_label the worker
116             must provide to handle the task.
117 
118     Returns (pulpcore.app.models.Task): The Pulp Task that was created.
119 
120     Raises:
121         ValueError: When `resources` is an unsupported type.
122     """
123 
124     assert deferred or immediate, "A task must be at least `deferred` or `immediate`."
125 
126     if callable(func):
127         function_name = f"{func.__module__}.{func.__name__}"
128     else:
129         function_name = func
130 
131     if versions is None:
132         try:
133             versions = MODULE_PLUGIN_VERSIONS[function_name.split(".", maxsplit=1)[0]]
134         except KeyError:
135             deprecation_logger.warn(
136                 _(
137                     "Using functions outside of pulp components as tasks is not supported and will "
138                     "result in runtime errors with pulpcore>=3.40."
139                 )
140             )
141             # The best we can do now...
142             versions = MODULE_PLUGIN_VERSIONS["pulpcore"]
143 
144     if exclusive_resources is None:
145         exclusive_resources = []
146     else:
147         exclusive_resources = _validate_and_get_resources(exclusive_resources)
148     if shared_resources is None:
149         shared_resources = []
150     else:
151         shared_resources = _validate_and_get_resources(shared_resources)
152 
153     # A task that is exclusive on a domain will block all tasks within that domain
154     domain_url = get_url(get_domain())
155     if domain_url not in exclusive_resources:
156         shared_resources.append(domain_url)
157     resources = exclusive_resources + [f"shared:{resource}" for resource in shared_resources]
158 
159     notify_workers = False
160     with contextlib.ExitStack() as stack:
161         with transaction.atomic():
162             task = Task.objects.create(
163                 state=TASK_STATES.WAITING,
164                 logging_cid=(get_guid()),
165                 task_group=task_group,
166                 name=function_name,
167                 enc_args=args,
168                 enc_kwargs=kwargs,
169                 parent_task=Task.current(),
170                 reserved_resources_record=resources,
171                 versions=versions,
172             )
173             if immediate:
174                 # Grab the advisory lock before the task hits the db.
175                 stack.enter_context(task)
176             else:
177                 notify_workers = True
178         if immediate:
179             prior_tasks = Task.objects.filter(
180                 state__in=TASK_INCOMPLETE_STATES, pulp_created__lt=task.pulp_created
181             )
182             # Compile a list of resources that must not be taken by other tasks.
183             colliding_resources = (
184                 shared_resources
185                 + exclusive_resources
186                 + [f"shared:{resource}" for resource in exclusive_resources]
187             )
188             # Can we execute this task immediately?
189             if (
190                 not colliding_resources
191                 or not prior_tasks.filter(
192                     reserved_resources_record__overlap=colliding_resources
193                 ).exists()
194             ):
195                 execute_task(task)
196                 if resources:
197                     notify_workers = True
198             elif deferred:
199                 notify_workers = True
200             else:
201                 task.set_canceling()
202                 task.set_canceled(TASK_STATES.CANCELED, "Resources temporarily unavailable.")
203     if notify_workers:
204         wakeup_worker()
205     return task
206 
207 
208 def cancel_task(task_id):
209     """
210     Cancel the task that is represented by the given task_id.
211 
212     This method cancels only the task with given task_id, not the spawned tasks. This also updates
213     task's state to 'canceling'.
214 
215     Args:
216         task_id (str): The ID of the task you wish to cancel
217 
218     Raises:
219         rest_framework.exceptions.NotFound: If a task with given task_id does not exist
220     """
221     task = Task.objects.get(pk=task_id)
222 
223     if task.state in TASK_FINAL_STATES:
224         # If the task is already done, just stop
225         _logger.debug(
226             "Task [{task_id}] already in a final state: {state}".format(
227                 task_id=task_id, state=task.state
228             )
229         )
230         return task
231 
232     _logger.info(_("Canceling task: {id}").format(id=task_id))
233 
234     # This is the only valid transition without holding the task lock
235     task.set_canceling()
236     # Notify the worker that might be running that task and other workers to clean up
237     with connection.cursor() as cursor:
238         cursor.execute("SELECT pg_notify('pulp_worker_cancel', %s)", (str(task.pk),))
239         cursor.execute("NOTIFY pulp_worker_wakeup")
240     return task
241
```
--- 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/pulpcore/tasking/tasks.py b/pulpcore/tasking/tasks.py
--- a/pulpcore/tasking/tasks.py
+++ b/pulpcore/tasking/tasks.py
@@ -11,7 +11,6 @@
 from django.db.models import Model
 from django_guid import get_guid
 from pulpcore.app.apps import MODULE_PLUGIN_VERSIONS
-from pulpcore.app.loggers import deprecation_logger
 from pulpcore.app.models import Task
 from pulpcore.app.util import current_task, get_domain, get_url
 from pulpcore.constants import TASK_FINAL_STATES, TASK_INCOMPLETE_STATES, TASK_STATES
@@ -129,17 +128,7 @@
         function_name = func
 
     if versions is None:
-        try:
-            versions = MODULE_PLUGIN_VERSIONS[function_name.split(".", maxsplit=1)[0]]
-        except KeyError:
-            deprecation_logger.warn(
-                _(
-                    "Using functions outside of pulp components as tasks is not supported and will "
-                    "result in runtime errors with pulpcore>=3.40."
-                )
-            )
-            # The best we can do now...
-            versions = MODULE_PLUGIN_VERSIONS["pulpcore"]
+        versions = MODULE_PLUGIN_VERSIONS[function_name.split(".", maxsplit=1)[0]]
 
     if exclusive_resources is None:
         exclusive_resources = []

{"golden_diff": "diff --git a/pulpcore/tasking/tasks.py b/pulpcore/tasking/tasks.py\n--- a/pulpcore/tasking/tasks.py\n+++ b/pulpcore/tasking/tasks.py\n@@ -11,7 +11,6 @@\n from django.db.models import Model\n from django_guid import get_guid\n from pulpcore.app.apps import MODULE_PLUGIN_VERSIONS\n-from pulpcore.app.loggers import deprecation_logger\n from pulpcore.app.models import Task\n from pulpcore.app.util import current_task, get_domain, get_url\n from pulpcore.constants import TASK_FINAL_STATES, TASK_INCOMPLETE_STATES, TASK_STATES\n@@ -129,17 +128,7 @@\n         function_name = func\n \n     if versions is None:\n-        try:\n-            versions = MODULE_PLUGIN_VERSIONS[function_name.split(\".\", maxsplit=1)[0]]\n-        except KeyError:\n-            deprecation_logger.warn(\n-                _(\n-                    \"Using functions outside of pulp components as tasks is not supported and will \"\n-                    \"result in runtime errors with pulpcore>=3.40.\"\n-                )\n-            )\n-            # The best we can do now...\n-            versions = MODULE_PLUGIN_VERSIONS[\"pulpcore\"]\n+        versions = MODULE_PLUGIN_VERSIONS[function_name.split(\".\", maxsplit=1)[0]]\n \n     if exclusive_resources is None:\n         exclusive_resources = []\n", "issue": "Remove the ability to call non-pulp-related functions as task targets\n\n", "before_files": [{"content": "import asyncio\nimport contextlib\nimport contextvars\nimport importlib\nimport logging\nimport sys\nimport traceback\nfrom gettext import gettext as _\n\nfrom django.db import connection, transaction\nfrom django.db.models import Model\nfrom django_guid import get_guid\nfrom pulpcore.app.apps import MODULE_PLUGIN_VERSIONS\nfrom pulpcore.app.loggers import deprecation_logger\nfrom pulpcore.app.models import Task\nfrom pulpcore.app.util import current_task, get_domain, get_url\nfrom pulpcore.constants import TASK_FINAL_STATES, TASK_INCOMPLETE_STATES, TASK_STATES\n\n_logger = logging.getLogger(__name__)\n\n\ndef _validate_and_get_resources(resources):\n    resource_set = set()\n    for r in resources:\n        if isinstance(r, str):\n            resource_set.add(r)\n        elif isinstance(r, Model):\n            resource_set.add(get_url(r))\n        elif r is None:\n            # Silently drop None values\n            pass\n        else:\n            raise ValueError(_(\"Must be (str|Model)\"))\n    return list(resource_set)\n\n\ndef wakeup_worker():\n    # Notify workers\n    with connection.connection.cursor() as cursor:\n        cursor.execute(\"NOTIFY pulp_worker_wakeup\")\n\n\ndef execute_task(task):\n    # This extra stack is needed to isolate the current_task ContextVar\n    contextvars.copy_context().run(_execute_task, task)\n\n\ndef _execute_task(task):\n    # Store the task id in the context for `Task.current()`.\n    current_task.set(task)\n    task.set_running()\n    try:\n        _logger.info(_(\"Starting task %s\"), task.pk)\n\n        # Execute task\n        module_name, function_name = task.name.rsplit(\".\", 1)\n        module = importlib.import_module(module_name)\n        func = getattr(module, function_name)\n        args = task.enc_args or ()\n        kwargs = task.enc_kwargs or {}\n        result = func(*args, **kwargs)\n        if asyncio.iscoroutine(result):\n            _logger.debug(_(\"Task is coroutine %s\"), task.pk)\n            loop = asyncio.get_event_loop()\n            loop.run_until_complete(result)\n\n    except Exception:\n        exc_type, exc, tb = sys.exc_info()\n        task.set_failed(exc, tb)\n        _logger.info(_(\"Task %s failed (%s)\"), task.pk, exc)\n        _logger.info(\"\\n\".join(traceback.format_list(traceback.extract_tb(tb))))\n    else:\n        task.set_completed()\n        _logger.info(_(\"Task completed %s\"), task.pk)\n\n\ndef dispatch(\n    func,\n    args=None,\n    kwargs=None,\n    task_group=None,\n    exclusive_resources=None,\n    shared_resources=None,\n    immediate=False,\n    deferred=True,\n    versions=None,\n):\n    \"\"\"\n    Enqueue a message to Pulp workers with a reservation.\n\n    This method provides normal enqueue functionality, while also requesting necessary locks for\n    serialized urls. No two tasks that claim the same resource can execute concurrently. It\n    accepts resources which it transforms into a list of urls (one for each resource).\n\n    This method creates a :class:`pulpcore.app.models.Task` object and returns it.\n\n    The values in `args` and `kwargs` must be JSON serializable, but may contain instances of\n    ``uuid.UUID``.\n\n    Args:\n        func (callable | str): The function to be run when the necessary locks are acquired.\n        args (tuple): The positional arguments to pass on to the task.\n        kwargs (dict): The keyword arguments to pass on to the task.\n        task_group (pulpcore.app.models.TaskGroup): A TaskGroup to add the created Task to.\n        exclusive_resources (list): A list of resources this task needs exclusive access to while\n            running. Each resource can be either a `str` or a `django.models.Model` instance.\n        shared_resources (list): A list of resources this task needs non-exclusive access to while\n            running. Each resource can be either a `str` or a `django.models.Model` instance.\n        immediate (bool): Whether to allow running this task immediately. It must be guaranteed to\n            execute fast without blocking. If not all resource constraints are met, the task will\n            either be returned in a canceled state or, if `deferred` is `True` be left in the queue\n            to be picked up by a worker eventually. Defaults to `False`.\n        deferred (bool): Whether to allow defer running the task to a pulpcore_worker. Defaults to\n            `True`. `immediate` and `deferred` cannot both be `False`.\n        versions (Optional[Dict[str, str]]): Minimum versions of components by app_label the worker\n            must provide to handle the task.\n\n    Returns (pulpcore.app.models.Task): The Pulp Task that was created.\n\n    Raises:\n        ValueError: When `resources` is an unsupported type.\n    \"\"\"\n\n    assert deferred or immediate, \"A task must be at least `deferred` or `immediate`.\"\n\n    if callable(func):\n        function_name = f\"{func.__module__}.{func.__name__}\"\n    else:\n        function_name = func\n\n    if versions is None:\n        try:\n            versions = MODULE_PLUGIN_VERSIONS[function_name.split(\".\", maxsplit=1)[0]]\n        except KeyError:\n            deprecation_logger.warn(\n                _(\n                    \"Using functions outside of pulp components as tasks is not supported and will \"\n                    \"result in runtime errors with pulpcore>=3.40.\"\n                )\n            )\n            # The best we can do now...\n            versions = MODULE_PLUGIN_VERSIONS[\"pulpcore\"]\n\n    if exclusive_resources is None:\n        exclusive_resources = []\n    else:\n        exclusive_resources = _validate_and_get_resources(exclusive_resources)\n    if shared_resources is None:\n        shared_resources = []\n    else:\n        shared_resources = _validate_and_get_resources(shared_resources)\n\n    # A task that is exclusive on a domain will block all tasks within that domain\n    domain_url = get_url(get_domain())\n    if domain_url not in exclusive_resources:\n        shared_resources.append(domain_url)\n    resources = exclusive_resources + [f\"shared:{resource}\" for resource in shared_resources]\n\n    notify_workers = False\n    with contextlib.ExitStack() as stack:\n        with transaction.atomic():\n            task = Task.objects.create(\n                state=TASK_STATES.WAITING,\n                logging_cid=(get_guid()),\n                task_group=task_group,\n                name=function_name,\n                enc_args=args,\n                enc_kwargs=kwargs,\n                parent_task=Task.current(),\n                reserved_resources_record=resources,\n                versions=versions,\n            )\n            if immediate:\n                # Grab the advisory lock before the task hits the db.\n                stack.enter_context(task)\n            else:\n                notify_workers = True\n        if immediate:\n            prior_tasks = Task.objects.filter(\n                state__in=TASK_INCOMPLETE_STATES, pulp_created__lt=task.pulp_created\n            )\n            # Compile a list of resources that must not be taken by other tasks.\n            colliding_resources = (\n                shared_resources\n                + exclusive_resources\n                + [f\"shared:{resource}\" for resource in exclusive_resources]\n            )\n            # Can we execute this task immediately?\n            if (\n                not colliding_resources\n                or not prior_tasks.filter(\n                    reserved_resources_record__overlap=colliding_resources\n                ).exists()\n            ):\n                execute_task(task)\n                if resources:\n                    notify_workers = True\n            elif deferred:\n                notify_workers = True\n            else:\n                task.set_canceling()\n                task.set_canceled(TASK_STATES.CANCELED, \"Resources temporarily unavailable.\")\n    if notify_workers:\n        wakeup_worker()\n    return task\n\n\ndef cancel_task(task_id):\n    \"\"\"\n    Cancel the task that is represented by the given task_id.\n\n    This method cancels only the task with given task_id, not the spawned tasks. This also updates\n    task's state to 'canceling'.\n\n    Args:\n        task_id (str): The ID of the task you wish to cancel\n\n    Raises:\n        rest_framework.exceptions.NotFound: If a task with given task_id does not exist\n    \"\"\"\n    task = Task.objects.get(pk=task_id)\n\n    if task.state in TASK_FINAL_STATES:\n        # If the task is already done, just stop\n        _logger.debug(\n            \"Task [{task_id}] already in a final state: {state}\".format(\n                task_id=task_id, state=task.state\n            )\n        )\n        return task\n\n    _logger.info(_(\"Canceling task: {id}\").format(id=task_id))\n\n    # This is the only valid transition without holding the task lock\n    task.set_canceling()\n    # Notify the worker that might be running that task and other workers to clean up\n    with connection.cursor() as cursor:\n        cursor.execute(\"SELECT pg_notify('pulp_worker_cancel', %s)\", (str(task.pk),))\n        cursor.execute(\"NOTIFY pulp_worker_wakeup\")\n    return task\n", "path": "pulpcore/tasking/tasks.py"}], "after_files": [{"content": "import asyncio\nimport contextlib\nimport contextvars\nimport importlib\nimport logging\nimport sys\nimport traceback\nfrom gettext import gettext as _\n\nfrom django.db import connection, transaction\nfrom django.db.models import Model\nfrom django_guid import get_guid\nfrom pulpcore.app.apps import MODULE_PLUGIN_VERSIONS\nfrom pulpcore.app.models import Task\nfrom pulpcore.app.util import current_task, get_domain, get_url\nfrom pulpcore.constants import TASK_FINAL_STATES, TASK_INCOMPLETE_STATES, TASK_STATES\n\n_logger = logging.getLogger(__name__)\n\n\ndef _validate_and_get_resources(resources):\n    resource_set = set()\n    for r in resources:\n        if isinstance(r, str):\n            resource_set.add(r)\n        elif isinstance(r, Model):\n            resource_set.add(get_url(r))\n        elif r is None:\n            # Silently drop None values\n            pass\n        else:\n            raise ValueError(_(\"Must be (str|Model)\"))\n    return list(resource_set)\n\n\ndef wakeup_worker():\n    # Notify workers\n    with connection.connection.cursor() as cursor:\n        cursor.execute(\"NOTIFY pulp_worker_wakeup\")\n\n\ndef execute_task(task):\n    # This extra stack is needed to isolate the current_task ContextVar\n    contextvars.copy_context().run(_execute_task, task)\n\n\ndef _execute_task(task):\n    # Store the task id in the context for `Task.current()`.\n    current_task.set(task)\n    task.set_running()\n    try:\n        _logger.info(_(\"Starting task %s\"), task.pk)\n\n        # Execute task\n        module_name, function_name = task.name.rsplit(\".\", 1)\n        module = importlib.import_module(module_name)\n        func = getattr(module, function_name)\n        args = task.enc_args or ()\n        kwargs = task.enc_kwargs or {}\n        result = func(*args, **kwargs)\n        if asyncio.iscoroutine(result):\n            _logger.debug(_(\"Task is coroutine %s\"), task.pk)\n            loop = asyncio.get_event_loop()\n            loop.run_until_complete(result)\n\n    except Exception:\n        exc_type, exc, tb = sys.exc_info()\n        task.set_failed(exc, tb)\n        _logger.info(_(\"Task %s failed (%s)\"), task.pk, exc)\n        _logger.info(\"\\n\".join(traceback.format_list(traceback.extract_tb(tb))))\n    else:\n        task.set_completed()\n        _logger.info(_(\"Task completed %s\"), task.pk)\n\n\ndef dispatch(\n    func,\n    args=None,\n    kwargs=None,\n    task_group=None,\n    exclusive_resources=None,\n    shared_resources=None,\n    immediate=False,\n    deferred=True,\n    versions=None,\n):\n    \"\"\"\n    Enqueue a message to Pulp workers with a reservation.\n\n    This method provides normal enqueue functionality, while also requesting necessary locks for\n    serialized urls. No two tasks that claim the same resource can execute concurrently. It\n    accepts resources which it transforms into a list of urls (one for each resource).\n\n    This method creates a :class:`pulpcore.app.models.Task` object and returns it.\n\n    The values in `args` and `kwargs` must be JSON serializable, but may contain instances of\n    ``uuid.UUID``.\n\n    Args:\n        func (callable | str): The function to be run when the necessary locks are acquired.\n        args (tuple): The positional arguments to pass on to the task.\n        kwargs (dict): The keyword arguments to pass on to the task.\n        task_group (pulpcore.app.models.TaskGroup): A TaskGroup to add the created Task to.\n        exclusive_resources (list): A list of resources this task needs exclusive access to while\n            running. Each resource can be either a `str` or a `django.models.Model` instance.\n        shared_resources (list): A list of resources this task needs non-exclusive access to while\n            running. Each resource can be either a `str` or a `django.models.Model` instance.\n        immediate (bool): Whether to allow running this task immediately. It must be guaranteed to\n            execute fast without blocking. If not all resource constraints are met, the task will\n            either be returned in a canceled state or, if `deferred` is `True` be left in the queue\n            to be picked up by a worker eventually. Defaults to `False`.\n        deferred (bool): Whether to allow defer running the task to a pulpcore_worker. Defaults to\n            `True`. `immediate` and `deferred` cannot both be `False`.\n        versions (Optional[Dict[str, str]]): Minimum versions of components by app_label the worker\n            must provide to handle the task.\n\n    Returns (pulpcore.app.models.Task): The Pulp Task that was created.\n\n    Raises:\n        ValueError: When `resources` is an unsupported type.\n    \"\"\"\n\n    assert deferred or immediate, \"A task must be at least `deferred` or `immediate`.\"\n\n    if callable(func):\n        function_name = f\"{func.__module__}.{func.__name__}\"\n    else:\n        function_name = func\n\n    if versions is None:\n        versions = MODULE_PLUGIN_VERSIONS[function_name.split(\".\", maxsplit=1)[0]]\n\n    if exclusive_resources is None:\n        exclusive_resources = []\n    else:\n        exclusive_resources = _validate_and_get_resources(exclusive_resources)\n    if shared_resources is None:\n        shared_resources = []\n    else:\n        shared_resources = _validate_and_get_resources(shared_resources)\n\n    # A task that is exclusive on a domain will block all tasks within that domain\n    domain_url = get_url(get_domain())\n    if domain_url not in exclusive_resources:\n        shared_resources.append(domain_url)\n    resources = exclusive_resources + [f\"shared:{resource}\" for resource in shared_resources]\n\n    notify_workers = False\n    with contextlib.ExitStack() as stack:\n        with transaction.atomic():\n            task = Task.objects.create(\n                state=TASK_STATES.WAITING,\n                logging_cid=(get_guid()),\n                task_group=task_group,\n                name=function_name,\n                enc_args=args,\n                enc_kwargs=kwargs,\n                parent_task=Task.current(),\n                reserved_resources_record=resources,\n                versions=versions,\n            )\n            if immediate:\n                # Grab the advisory lock before the task hits the db.\n                stack.enter_context(task)\n            else:\n                notify_workers = True\n        if immediate:\n            prior_tasks = Task.objects.filter(\n                state__in=TASK_INCOMPLETE_STATES, pulp_created__lt=task.pulp_created\n            )\n            # Compile a list of resources that must not be taken by other tasks.\n            colliding_resources = (\n                shared_resources\n                + exclusive_resources\n                + [f\"shared:{resource}\" for resource in exclusive_resources]\n            )\n            # Can we execute this task immediately?\n            if (\n                not colliding_resources\n                or not prior_tasks.filter(\n                    reserved_resources_record__overlap=colliding_resources\n                ).exists()\n            ):\n                execute_task(task)\n                if resources:\n                    notify_workers = True\n            elif deferred:\n                notify_workers = True\n            else:\n                task.set_canceling()\n                task.set_canceled(TASK_STATES.CANCELED, \"Resources temporarily unavailable.\")\n    if notify_workers:\n        wakeup_worker()\n    return task\n\n\ndef cancel_task(task_id):\n    \"\"\"\n    Cancel the task that is represented by the given task_id.\n\n    This method cancels only the task with given task_id, not the spawned tasks. This also updates\n    task's state to 'canceling'.\n\n    Args:\n        task_id (str): The ID of the task you wish to cancel\n\n    Raises:\n        rest_framework.exceptions.NotFound: If a task with given task_id does not exist\n    \"\"\"\n    task = Task.objects.get(pk=task_id)\n\n    if task.state in TASK_FINAL_STATES:\n        # If the task is already done, just stop\n        _logger.debug(\n            \"Task [{task_id}] already in a final state: {state}\".format(\n                task_id=task_id, state=task.state\n            )\n        )\n        return task\n\n    _logger.info(_(\"Canceling task: {id}\").format(id=task_id))\n\n    # This is the only valid transition without holding the task lock\n    task.set_canceling()\n    # Notify the worker that might be running that task and other workers to clean up\n    with connection.cursor() as cursor:\n        cursor.execute(\"SELECT pg_notify('pulp_worker_cancel', %s)\", (str(task.pk),))\n        cursor.execute(\"NOTIFY pulp_worker_wakeup\")\n    return task\n", "path": "pulpcore/tasking/tasks.py"}]}