We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Concurrency issues(?) (with tasks workers?)
<!--
    NOTE: IF YOUR ISSUE DOES NOT FOLLOW THIS TEMPLATE, IT WILL BE CLOSED.

    This form is only for reporting reproducible bugs. If you need assistance
    with Nautobot installation, or if you have a general question, please start a
    discussion instead: https://github.com/nautobot/nautobot/discussions

    Please describe the environment in which you are running Nautobot. Be sure
    that you are running an unmodified instance of the latest stable release
    before submitting a bug report, and that any plugins have been disabled.
-->
### Environment
* Python version: 3.9
* Nautobot version: 1.1.0
* Redis: redis:6.2.5-alpine
* PSQL: postgres:13.3-alpine

Docker-compose extract:
```
services:
  nautobot: &nautobot
    image: networktocode/nautobot:1.1.0-py3.9
    depends_on:
      - postgres
      - redis
    volumes:
      - ./volumes/media:/opt/nautobot/media:z,rw
      - ./volumes/git:/opt/nautobot/git:z,rw
      - ./volumes/jobs:/opt/nautobot/jobs:z,rw
    environment:
      NAUTOBOT_DB_HOST: postgres
      NAUTOBOT_DB_USER: nautobot
      NAUTOBOT_DB_PASSWORD: nautobot
      NAUTOBOT_DB_NAME: nautobot
      NAUTOBOT_REDIS_HOST: redis
      NAUTOBOT_SECRET_KEY: "*****"
      NAUTOBOT_MAX_PAGE_SIZE: "50000"
      NAUTOBOT_CHANGELOG_RETENTION: "366"
      NAUTOBOT_METRICS_ENABLED: "true"
      #NAUTOBOT_CACHEOPS_ENABLED: "false"

  celery_worker:
    <<: *nautobot
    entrypoint: "nautobot-server celery worker -B -l INFO"
    networks:
      - default
    labels: []
    depends_on:
      - nautobot
    healthcheck:
      interval: 5s
      timeout: 5s
      start_period: 5s
      retries: 3
      test: ["CMD", "nautobot-server", "health_check"]

  rq_worker:
    <<: *nautobot
    entrypoint: "nautobot-server rqworker"
    networks:
      - default
    labels: []
    depends_on:
      - nautobot
    healthcheck:
      interval: 5s
      timeout: 5s
      start_period: 5s
      retries: 3
      test: ["CMD", "nautobot-server", "health_check"]

  # postgres - https://hub.docker.com/_/postgres
  postgres:
    image: postgres:13.3-alpine
    volumes:
      - ./volumes/pgsql:/var/lib/postgresql/data
    environment:
      POSTGRES_USER: nautobot
      POSTGRES_PASSWORD: nautobot
      POSTGRES_DB: nautobot

  # redis - https://hub.docker.com/_/redis
  redis:
    image: redis:6.2.5-alpine
```

<!--
    Describe in detail the exact steps that someone else can take to reproduce
    this bug using the current stable release of Nautobot. Begin with the
    creation of any necessary database objects and call out every operation
    being performed explicitly. If reporting a bug in the REST API, be sure to
    reconstruct the raw HTTP request(s) being made: Don't rely on a client
    library such as pynautobot.
-->
### Steps to Reproduce
1. Stop the Celery worker
2. Create jobs (doing many queries?):
```py
from nautobot.extras.jobs import Job
from nautobot.dcim.models import Device, Interface
import time


class TestA(Job):

  devices = Device.objects

  class Meta:
    read_only = True

  def test_a(self):
    j = 0
    for i, device in enumerate(self.devices.all()):
      for j, interface in enumerate(device.interfaces.all(), start=j):
        if j > 50:
          break
        self.log_info(obj=interface, message=f'Iteration {i}/{j}, name={device.name}/{interface.name}, {interface.connected_endpoint}')
        time.sleep(0.1)


class TestB(Job):

  devices = Device.objects

  class Meta:
    read_only = True

  def test_b(self):
    j = 0
    for i, device in enumerate(self.devices.all()):
      for j, interface in enumerate(device.interfaces.all(), start=j):
        if j > 50:
          break
        self.log_info(obj=interface, message=f'Iteration {i}/{j}, name={device.name}/{interface.name}, {interface.connected_endpoint}')
        time.sleep(0.1)
```

2. Start multiple instances of each (using the API if not allowed in the UI)
3. Start the worker

<!-- What did you expect to happen? -->
### Expected Behavior
Jobs are fully ran in the start order

<!-- What happened instead? -->
### Observed Behavior
Some jobs are stuck in either the pending or the running state, with errors in the worker logs

### Additional informations

This is the best repro I could find in many hours of really weird and random errors. I noticed setting `NAUTOBOT_CACHEOPS_ENABLED: 'false'` could help getting errors more often.

And in that case (no cacheops), right after an instance is started, listing the job results (which seems to imply `page_size` times git refresh ?) + loading another page (like job list or git list/detail) is also a good way to crash the web container with weird errors too, like:
* `<class 'AttributeError'>, 'datetime.date' object has no attribute 'encode'`
* `<class 'RuntimeError'>, generator raised StopIteration`
* `<class 'AttributeError'>, 'NoneType' object has no attribute 'DoesNotExist'`

And in the logs about the git refresh:
```
nautobot_1       | 18:42:23.628 INFO    nautobot.jobs :
nautobot_1       |   Repository successfully refreshed
nautobot_1       | 18:42:23.655 INFO    nautobot.jobs :
nautobot_1       |   Repository successfully refreshed
nautobot_1       | 18:42:23.677 INFO    nautobot.jobs :
nautobot_1       |   Repository successfully refreshed
[..]
```

So I'm not sure the issue is restricted to the jobs part.

Some more random errors observed:
* `<class 'django.contrib.contenttypes.models.ContentType.DoesNotExist'> ContentType matching query does not exist.`
* `<class 'django.db.utils.DatabaseError'> error with status PGRES_TUPLES_OK and no message from the libpq`
* `<class 'ValueError'> Field 'id' expected a number but got 'ryws2lq3ihs******md9auxf1ua3'.`

For the job part, a possible workaround seems to be to set the pool implementation to solo (https://docs.celeryproject.org/en/stable/reference/cli.html#cmdoption-celery-worker-P): 
```yaml
    entrypoint: "nautobot-server celery worker -B -l INFO-P solo"
```

This really feels like I'm doing something wrong (especially since nobody else seems complaining?), I really hope not, but if that's the case I can't point out what it is.

Best, Alexandre

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nautobot/core/celery.py`
Content:
```
1 import json
2 import logging
3 
4 import nautobot
5 
6 from celery import Celery, shared_task
7 from django.core.serializers.json import DjangoJSONEncoder
8 from django.utils.module_loading import import_string
9 from kombu.serialization import register
10 
11 logger = logging.getLogger(__name__)
12 
13 # The Celery documentation tells us to call setup on the app to initialize
14 # settings, but we will NOT be doing that because of a chicken-and-egg problem
15 # when bootstrapping the Django settings with `nautobot-server`.
16 #
17 # Note this would normally set the `DJANGO_SETTINGS_MODULE` environment variable
18 # which Celery and its workers need under the hood.The Celery docs and examples
19 # normally have you set it here, but because of our custom settings bootstrapping
20 # it is handled in the `nautobot.setup() call, and we have implemented a
21 # `nautobot-server celery` command to provide the correct context so this does
22 # NOT need to be called here.
23 # nautobot.setup()
24 
25 app = Celery("nautobot")
26 
27 # Using a string here means the worker doesn't have to serialize
28 # the configuration object to child processes. Again, this is possible
29 # only after calling `nautobot.setup()` which sets `DJANGO_SETTINGS_MODULE`.
30 # - namespace='CELERY' means all celery-related configuration keys
31 #   should have a `CELERY_` prefix.
32 app.config_from_object("django.conf:settings", namespace="CELERY")
33 
34 # Load task modules from all registered Django apps.
35 app.autodiscover_tasks()
36 
37 
38 class NautobotKombuJSONEncoder(DjangoJSONEncoder):
39     """
40     Custom json encoder based on DjangoJSONEncoder that serializes objects that implement
41     the `nautobot_serialize()` method via the `__nautobot_type__` interface. This is useful
42     in passing special objects to and from Celery tasks.
43 
44     This pattern should generally be avoided by passing pointers to persisted objects to the
45     Celery tasks and retrieving them from within the task execution. While this is always possible
46     for model instances (which covers 99% of use cases), for rare instances where it does not,
47     and the actual object must be passed, this pattern allows for encoding and decoding
48     of such objects.
49 
50     It requires a conforming class to implement the instance method `nautobot_serialize()` which
51     returns a json serializable dictionary of the object representation. The class must also implement
52     the `nautobot_deserialize()` class method which takes the dictionary representation and returns
53     an actual instance of the class.
54     """
55 
56     def default(self, obj):
57         if hasattr(obj, "nautobot_serialize"):
58             cls = obj.__class__
59             module = cls.__module__
60             qual_name = ".".join([module, cls.__qualname__])  # fully qualified dotted import path
61             logger.debug("Performing nautobot serialization on %s for type %s", obj, qual_name)
62             data = {"__nautobot_type__": qual_name}
63             data.update(obj.nautobot_serialize())
64             return data
65 
66         elif isinstance(obj, set):
67             # Convert a set to a list for passing to and from a task
68             return list(obj)
69 
70         else:
71             return DjangoJSONEncoder.default(self, obj)
72 
73 
74 def nautobot_kombu_json_loads_hook(data):
75     """
76     In concert with the NautobotKombuJSONEncoder json encoder, this object hook method decodes
77     objects that implement the `__nautobot_type__` interface via the `nautobot_deserialize()` class method.
78     """
79     if "__nautobot_type__" in data:
80         qual_name = data.pop("__nautobot_type__")
81         logger.debug("Performing nautobot deserialization for type %s", qual_name)
82         cls = import_string(qual_name)  # fully qualified dotted import path
83         if cls:
84             return cls.nautobot_deserialize(data)
85         else:
86             raise TypeError(f"Unable to import {qual_name} during nautobot deserialization")
87     else:
88         return data
89 
90 
91 # Encoder function
92 def _dumps(obj):
93     return json.dumps(obj, cls=NautobotKombuJSONEncoder)
94 
95 
96 # Decoder function
97 def _loads(obj):
98     return json.loads(obj, object_hook=nautobot_kombu_json_loads_hook)
99 
100 
101 # Register the custom serialization type
102 register("nautobot_json", _dumps, _loads, content_type="application/x-nautobot-json", content_encoding="utf-8")
103 
104 
105 #
106 # nautobot_task
107 #
108 # By exposing `shared_task` within our own namespace, we leave the door open to
109 # extending and expanding the usage and meaning of shared_task without having
110 # to undergo further refactoring of task's decorators. We could also transparently
111 # swap out shared_task to a custom base task.
112 #
113 
114 nautobot_task = shared_task
115
```
--- 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/nautobot/core/celery.py b/nautobot/core/celery.py
--- a/nautobot/core/celery.py
+++ b/nautobot/core/celery.py
@@ -4,6 +4,7 @@
 import nautobot
 
 from celery import Celery, shared_task
+from celery.fixups.django import DjangoFixup
 from django.core.serializers.json import DjangoJSONEncoder
 from django.utils.module_loading import import_string
 from kombu.serialization import register
@@ -31,6 +32,11 @@
 #   should have a `CELERY_` prefix.
 app.config_from_object("django.conf:settings", namespace="CELERY")
 
+# Because of the chicken-and-egg Django settings bootstrapping issue,
+# Celery doesn't automatically install its Django-specific patches.
+# So we need to explicitly do so ourselves:
+DjangoFixup(app).install()
+
 # Load task modules from all registered Django apps.
 app.autodiscover_tasks()

{"golden_diff": "diff --git a/nautobot/core/celery.py b/nautobot/core/celery.py\n--- a/nautobot/core/celery.py\n+++ b/nautobot/core/celery.py\n@@ -4,6 +4,7 @@\n import nautobot\n \n from celery import Celery, shared_task\n+from celery.fixups.django import DjangoFixup\n from django.core.serializers.json import DjangoJSONEncoder\n from django.utils.module_loading import import_string\n from kombu.serialization import register\n@@ -31,6 +32,11 @@\n #   should have a `CELERY_` prefix.\n app.config_from_object(\"django.conf:settings\", namespace=\"CELERY\")\n \n+# Because of the chicken-and-egg Django settings bootstrapping issue,\n+# Celery doesn't automatically install its Django-specific patches.\n+# So we need to explicitly do so ourselves:\n+DjangoFixup(app).install()\n+\n # Load task modules from all registered Django apps.\n app.autodiscover_tasks()\n", "issue": "Concurrency issues(?) (with tasks workers?)\n<!--\r\n    NOTE: IF YOUR ISSUE DOES NOT FOLLOW THIS TEMPLATE, IT WILL BE CLOSED.\r\n\r\n    This form is only for reporting reproducible bugs. If you need assistance\r\n    with Nautobot installation, or if you have a general question, please start a\r\n    discussion instead: https://github.com/nautobot/nautobot/discussions\r\n\r\n    Please describe the environment in which you are running Nautobot. Be sure\r\n    that you are running an unmodified instance of the latest stable release\r\n    before submitting a bug report, and that any plugins have been disabled.\r\n-->\r\n### Environment\r\n* Python version: 3.9\r\n* Nautobot version: 1.1.0\r\n* Redis: redis:6.2.5-alpine\r\n* PSQL: postgres:13.3-alpine\r\n\r\nDocker-compose extract:\r\n```\r\nservices:\r\n  nautobot: &nautobot\r\n    image: networktocode/nautobot:1.1.0-py3.9\r\n    depends_on:\r\n      - postgres\r\n      - redis\r\n    volumes:\r\n      - ./volumes/media:/opt/nautobot/media:z,rw\r\n      - ./volumes/git:/opt/nautobot/git:z,rw\r\n      - ./volumes/jobs:/opt/nautobot/jobs:z,rw\r\n    environment:\r\n      NAUTOBOT_DB_HOST: postgres\r\n      NAUTOBOT_DB_USER: nautobot\r\n      NAUTOBOT_DB_PASSWORD: nautobot\r\n      NAUTOBOT_DB_NAME: nautobot\r\n      NAUTOBOT_REDIS_HOST: redis\r\n      NAUTOBOT_SECRET_KEY: \"*****\"\r\n      NAUTOBOT_MAX_PAGE_SIZE: \"50000\"\r\n      NAUTOBOT_CHANGELOG_RETENTION: \"366\"\r\n      NAUTOBOT_METRICS_ENABLED: \"true\"\r\n      #NAUTOBOT_CACHEOPS_ENABLED: \"false\"\r\n\r\n  celery_worker:\r\n    <<: *nautobot\r\n    entrypoint: \"nautobot-server celery worker -B -l INFO\"\r\n    networks:\r\n      - default\r\n    labels: []\r\n    depends_on:\r\n      - nautobot\r\n    healthcheck:\r\n      interval: 5s\r\n      timeout: 5s\r\n      start_period: 5s\r\n      retries: 3\r\n      test: [\"CMD\", \"nautobot-server\", \"health_check\"]\r\n\r\n  rq_worker:\r\n    <<: *nautobot\r\n    entrypoint: \"nautobot-server rqworker\"\r\n    networks:\r\n      - default\r\n    labels: []\r\n    depends_on:\r\n      - nautobot\r\n    healthcheck:\r\n      interval: 5s\r\n      timeout: 5s\r\n      start_period: 5s\r\n      retries: 3\r\n      test: [\"CMD\", \"nautobot-server\", \"health_check\"]\r\n\r\n  # postgres - https://hub.docker.com/_/postgres\r\n  postgres:\r\n    image: postgres:13.3-alpine\r\n    volumes:\r\n      - ./volumes/pgsql:/var/lib/postgresql/data\r\n    environment:\r\n      POSTGRES_USER: nautobot\r\n      POSTGRES_PASSWORD: nautobot\r\n      POSTGRES_DB: nautobot\r\n\r\n  # redis - https://hub.docker.com/_/redis\r\n  redis:\r\n    image: redis:6.2.5-alpine\r\n```\r\n\r\n<!--\r\n    Describe in detail the exact steps that someone else can take to reproduce\r\n    this bug using the current stable release of Nautobot. Begin with the\r\n    creation of any necessary database objects and call out every operation\r\n    being performed explicitly. If reporting a bug in the REST API, be sure to\r\n    reconstruct the raw HTTP request(s) being made: Don't rely on a client\r\n    library such as pynautobot.\r\n-->\r\n### Steps to Reproduce\r\n1. Stop the Celery worker\r\n2. Create jobs (doing many queries?):\r\n```py\r\nfrom nautobot.extras.jobs import Job\r\nfrom nautobot.dcim.models import Device, Interface\r\nimport time\r\n\r\n\r\nclass TestA(Job):\r\n\r\n  devices = Device.objects\r\n\r\n  class Meta:\r\n    read_only = True\r\n\r\n  def test_a(self):\r\n    j = 0\r\n    for i, device in enumerate(self.devices.all()):\r\n      for j, interface in enumerate(device.interfaces.all(), start=j):\r\n        if j > 50:\r\n          break\r\n        self.log_info(obj=interface, message=f'Iteration {i}/{j}, name={device.name}/{interface.name}, {interface.connected_endpoint}')\r\n        time.sleep(0.1)\r\n\r\n\r\nclass TestB(Job):\r\n\r\n  devices = Device.objects\r\n\r\n  class Meta:\r\n    read_only = True\r\n\r\n  def test_b(self):\r\n    j = 0\r\n    for i, device in enumerate(self.devices.all()):\r\n      for j, interface in enumerate(device.interfaces.all(), start=j):\r\n        if j > 50:\r\n          break\r\n        self.log_info(obj=interface, message=f'Iteration {i}/{j}, name={device.name}/{interface.name}, {interface.connected_endpoint}')\r\n        time.sleep(0.1)\r\n```\r\n\r\n2. Start multiple instances of each (using the API if not allowed in the UI)\r\n3. Start the worker\r\n\r\n<!-- What did you expect to happen? -->\r\n### Expected Behavior\r\nJobs are fully ran in the start order\r\n\r\n<!-- What happened instead? -->\r\n### Observed Behavior\r\nSome jobs are stuck in either the pending or the running state, with errors in the worker logs\r\n\r\n### Additional informations\r\n\r\nThis is the best repro I could find in many hours of really weird and random errors. I noticed setting `NAUTOBOT_CACHEOPS_ENABLED: 'false'` could help getting errors more often.\r\n\r\nAnd in that case (no cacheops), right after an instance is started, listing the job results (which seems to imply `page_size` times git refresh ?) + loading another page (like job list or git list/detail) is also a good way to crash the web container with weird errors too, like:\r\n* `<class 'AttributeError'>, 'datetime.date' object has no attribute 'encode'`\r\n* `<class 'RuntimeError'>, generator raised StopIteration`\r\n* `<class 'AttributeError'>, 'NoneType' object has no attribute 'DoesNotExist'`\r\n\r\nAnd in the logs about the git refresh:\r\n```\r\nnautobot_1       | 18:42:23.628 INFO    nautobot.jobs :\r\nnautobot_1       |   Repository successfully refreshed\r\nnautobot_1       | 18:42:23.655 INFO    nautobot.jobs :\r\nnautobot_1       |   Repository successfully refreshed\r\nnautobot_1       | 18:42:23.677 INFO    nautobot.jobs :\r\nnautobot_1       |   Repository successfully refreshed\r\n[..]\r\n```\r\n\r\nSo I'm not sure the issue is restricted to the jobs part.\r\n\r\nSome more random errors observed:\r\n* `<class 'django.contrib.contenttypes.models.ContentType.DoesNotExist'> ContentType matching query does not exist.`\r\n* `<class 'django.db.utils.DatabaseError'> error with status PGRES_TUPLES_OK and no message from the libpq`\r\n* `<class 'ValueError'> Field 'id' expected a number but got 'ryws2lq3ihs******md9auxf1ua3'.`\r\n\r\nFor the job part, a possible workaround seems to be to set the pool implementation to solo (https://docs.celeryproject.org/en/stable/reference/cli.html#cmdoption-celery-worker-P): \r\n```yaml\r\n    entrypoint: \"nautobot-server celery worker -B -l INFO-P solo\"\r\n```\r\n\r\nThis really feels like I'm doing something wrong (especially since nobody else seems complaining?), I really hope not, but if that's the case I can't point out what it is.\r\n\r\nBest, Alexandre\n", "before_files": [{"content": "import json\nimport logging\n\nimport nautobot\n\nfrom celery import Celery, shared_task\nfrom django.core.serializers.json import DjangoJSONEncoder\nfrom django.utils.module_loading import import_string\nfrom kombu.serialization import register\n\nlogger = logging.getLogger(__name__)\n\n# The Celery documentation tells us to call setup on the app to initialize\n# settings, but we will NOT be doing that because of a chicken-and-egg problem\n# when bootstrapping the Django settings with `nautobot-server`.\n#\n# Note this would normally set the `DJANGO_SETTINGS_MODULE` environment variable\n# which Celery and its workers need under the hood.The Celery docs and examples\n# normally have you set it here, but because of our custom settings bootstrapping\n# it is handled in the `nautobot.setup() call, and we have implemented a\n# `nautobot-server celery` command to provide the correct context so this does\n# NOT need to be called here.\n# nautobot.setup()\n\napp = Celery(\"nautobot\")\n\n# Using a string here means the worker doesn't have to serialize\n# the configuration object to child processes. Again, this is possible\n# only after calling `nautobot.setup()` which sets `DJANGO_SETTINGS_MODULE`.\n# - namespace='CELERY' means all celery-related configuration keys\n#   should have a `CELERY_` prefix.\napp.config_from_object(\"django.conf:settings\", namespace=\"CELERY\")\n\n# Load task modules from all registered Django apps.\napp.autodiscover_tasks()\n\n\nclass NautobotKombuJSONEncoder(DjangoJSONEncoder):\n    \"\"\"\n    Custom json encoder based on DjangoJSONEncoder that serializes objects that implement\n    the `nautobot_serialize()` method via the `__nautobot_type__` interface. This is useful\n    in passing special objects to and from Celery tasks.\n\n    This pattern should generally be avoided by passing pointers to persisted objects to the\n    Celery tasks and retrieving them from within the task execution. While this is always possible\n    for model instances (which covers 99% of use cases), for rare instances where it does not,\n    and the actual object must be passed, this pattern allows for encoding and decoding\n    of such objects.\n\n    It requires a conforming class to implement the instance method `nautobot_serialize()` which\n    returns a json serializable dictionary of the object representation. The class must also implement\n    the `nautobot_deserialize()` class method which takes the dictionary representation and returns\n    an actual instance of the class.\n    \"\"\"\n\n    def default(self, obj):\n        if hasattr(obj, \"nautobot_serialize\"):\n            cls = obj.__class__\n            module = cls.__module__\n            qual_name = \".\".join([module, cls.__qualname__])  # fully qualified dotted import path\n            logger.debug(\"Performing nautobot serialization on %s for type %s\", obj, qual_name)\n            data = {\"__nautobot_type__\": qual_name}\n            data.update(obj.nautobot_serialize())\n            return data\n\n        elif isinstance(obj, set):\n            # Convert a set to a list for passing to and from a task\n            return list(obj)\n\n        else:\n            return DjangoJSONEncoder.default(self, obj)\n\n\ndef nautobot_kombu_json_loads_hook(data):\n    \"\"\"\n    In concert with the NautobotKombuJSONEncoder json encoder, this object hook method decodes\n    objects that implement the `__nautobot_type__` interface via the `nautobot_deserialize()` class method.\n    \"\"\"\n    if \"__nautobot_type__\" in data:\n        qual_name = data.pop(\"__nautobot_type__\")\n        logger.debug(\"Performing nautobot deserialization for type %s\", qual_name)\n        cls = import_string(qual_name)  # fully qualified dotted import path\n        if cls:\n            return cls.nautobot_deserialize(data)\n        else:\n            raise TypeError(f\"Unable to import {qual_name} during nautobot deserialization\")\n    else:\n        return data\n\n\n# Encoder function\ndef _dumps(obj):\n    return json.dumps(obj, cls=NautobotKombuJSONEncoder)\n\n\n# Decoder function\ndef _loads(obj):\n    return json.loads(obj, object_hook=nautobot_kombu_json_loads_hook)\n\n\n# Register the custom serialization type\nregister(\"nautobot_json\", _dumps, _loads, content_type=\"application/x-nautobot-json\", content_encoding=\"utf-8\")\n\n\n#\n# nautobot_task\n#\n# By exposing `shared_task` within our own namespace, we leave the door open to\n# extending and expanding the usage and meaning of shared_task without having\n# to undergo further refactoring of task's decorators. We could also transparently\n# swap out shared_task to a custom base task.\n#\n\nnautobot_task = shared_task\n", "path": "nautobot/core/celery.py"}], "after_files": [{"content": "import json\nimport logging\n\nimport nautobot\n\nfrom celery import Celery, shared_task\nfrom celery.fixups.django import DjangoFixup\nfrom django.core.serializers.json import DjangoJSONEncoder\nfrom django.utils.module_loading import import_string\nfrom kombu.serialization import register\n\nlogger = logging.getLogger(__name__)\n\n# The Celery documentation tells us to call setup on the app to initialize\n# settings, but we will NOT be doing that because of a chicken-and-egg problem\n# when bootstrapping the Django settings with `nautobot-server`.\n#\n# Note this would normally set the `DJANGO_SETTINGS_MODULE` environment variable\n# which Celery and its workers need under the hood.The Celery docs and examples\n# normally have you set it here, but because of our custom settings bootstrapping\n# it is handled in the `nautobot.setup() call, and we have implemented a\n# `nautobot-server celery` command to provide the correct context so this does\n# NOT need to be called here.\n# nautobot.setup()\n\napp = Celery(\"nautobot\")\n\n# Using a string here means the worker doesn't have to serialize\n# the configuration object to child processes. Again, this is possible\n# only after calling `nautobot.setup()` which sets `DJANGO_SETTINGS_MODULE`.\n# - namespace='CELERY' means all celery-related configuration keys\n#   should have a `CELERY_` prefix.\napp.config_from_object(\"django.conf:settings\", namespace=\"CELERY\")\n\n# Because of the chicken-and-egg Django settings bootstrapping issue,\n# Celery doesn't automatically install its Django-specific patches.\n# So we need to explicitly do so ourselves:\nDjangoFixup(app).install()\n\n# Load task modules from all registered Django apps.\napp.autodiscover_tasks()\n\n\nclass NautobotKombuJSONEncoder(DjangoJSONEncoder):\n    \"\"\"\n    Custom json encoder based on DjangoJSONEncoder that serializes objects that implement\n    the `nautobot_serialize()` method via the `__nautobot_type__` interface. This is useful\n    in passing special objects to and from Celery tasks.\n\n    This pattern should generally be avoided by passing pointers to persisted objects to the\n    Celery tasks and retrieving them from within the task execution. While this is always possible\n    for model instances (which covers 99% of use cases), for rare instances where it does not,\n    and the actual object must be passed, this pattern allows for encoding and decoding\n    of such objects.\n\n    It requires a conforming class to implement the instance method `nautobot_serialize()` which\n    returns a json serializable dictionary of the object representation. The class must also implement\n    the `nautobot_deserialize()` class method which takes the dictionary representation and returns\n    an actual instance of the class.\n    \"\"\"\n\n    def default(self, obj):\n        if hasattr(obj, \"nautobot_serialize\"):\n            cls = obj.__class__\n            module = cls.__module__\n            qual_name = \".\".join([module, cls.__qualname__])  # fully qualified dotted import path\n            logger.debug(\"Performing nautobot serialization on %s for type %s\", obj, qual_name)\n            data = {\"__nautobot_type__\": qual_name}\n            data.update(obj.nautobot_serialize())\n            return data\n\n        elif isinstance(obj, set):\n            # Convert a set to a list for passing to and from a task\n            return list(obj)\n\n        else:\n            return DjangoJSONEncoder.default(self, obj)\n\n\ndef nautobot_kombu_json_loads_hook(data):\n    \"\"\"\n    In concert with the NautobotKombuJSONEncoder json encoder, this object hook method decodes\n    objects that implement the `__nautobot_type__` interface via the `nautobot_deserialize()` class method.\n    \"\"\"\n    if \"__nautobot_type__\" in data:\n        qual_name = data.pop(\"__nautobot_type__\")\n        logger.debug(\"Performing nautobot deserialization for type %s\", qual_name)\n        cls = import_string(qual_name)  # fully qualified dotted import path\n        if cls:\n            return cls.nautobot_deserialize(data)\n        else:\n            raise TypeError(f\"Unable to import {qual_name} during nautobot deserialization\")\n    else:\n        return data\n\n\n# Encoder function\ndef _dumps(obj):\n    return json.dumps(obj, cls=NautobotKombuJSONEncoder)\n\n\n# Decoder function\ndef _loads(obj):\n    return json.loads(obj, object_hook=nautobot_kombu_json_loads_hook)\n\n\n# Register the custom serialization type\nregister(\"nautobot_json\", _dumps, _loads, content_type=\"application/x-nautobot-json\", content_encoding=\"utf-8\")\n\n\n#\n# nautobot_task\n#\n# By exposing `shared_task` within our own namespace, we leave the door open to\n# extending and expanding the usage and meaning of shared_task without having\n# to undergo further refactoring of task's decorators. We could also transparently\n# swap out shared_task to a custom base task.\n#\n\nnautobot_task = shared_task\n", "path": "nautobot/core/celery.py"}]}