We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Replication assumes every plugin supports Publications
The ```Replicator``` class, which is being subclassed inside plugins to support the replication feature, assumes that every plugin supports publications as it tries to access ```upstream_distribution["publication"]``` inside one of its methods (for some plugins, the dictionary simply doesn't contain the "publication" key so an exception gets raised). This forces certain subclasses of ```Replicator``` to create workarounds or rewrite the given method.

I propose making the method more general, removing such assumptions.

Relevant code: https://github.com/pulp/pulpcore/blob/c31a39c186fd734e9e0409321735d7d6509a12ee/pulpcore/app/replica.py#L97C9-L97C96


--- 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/app/replica.py`
Content:
```
1 from django.db.models import Model
2 import logging
3 
4 from pulp_glue.common.context import PulpContext
5 from pulpcore.tasking.tasks import dispatch
6 from pulpcore.app.tasks.base import (
7     general_update,
8     general_create,
9     general_multi_delete,
10 )
11 from pulpcore.plugin.util import get_url, get_domain
12 
13 _logger = logging.getLogger(__name__)
14 
15 
16 class ReplicaContext(PulpContext):
17     def __init__(self, **kwargs):
18         super().__init__(**kwargs)
19         self.out_buf = ""
20         self.err_buf = ""
21 
22     def echo(self, message: str, nl: bool = True, err: bool = False) -> None:
23         if err:
24             self.err_buf += message
25             if nl:
26                 _logger.warn("{}", self.err_buf)
27                 self.err_buf = ""
28         else:
29             self.out_buf += message
30             if nl:
31                 _logger.info("{}", self.out_buf)
32                 self.out_buf = ""
33 
34 
35 class Replicator:
36     distribution_ctx_cls = None
37     repository_ctx_cls = None
38     publication_ctx_cls = None
39     remote_model_cls = None
40     repository_model_cls = None
41     distribution_model_cls = None
42     distribution_serializer_name = None
43     repository_serializer_name = None
44     remote_serializer_name = None
45     app_label = None
46     sync_task = None
47 
48     def __init__(self, pulp_ctx, task_group, tls_settings):
49         """
50         :param pulp_ctx: PulpReplicaContext
51         :param task_group: TaskGroup
52         :param ca_cert: str
53         """
54         self.pulp_ctx = pulp_ctx
55         self.task_group = task_group
56         self.tls_settings = tls_settings
57         self.domain = get_domain()
58         self.distros_uris = [f"pdrn:{self.domain.pulp_id}:distributions"]
59 
60     @staticmethod
61     def needs_update(fields_dict, model_instance):
62         """
63         Compares a Model instance's attributes against a dictionary where keys are attribute
64         names and values are expected values.
65         """
66         needs_update = False
67         for field_name, value in fields_dict.items():
68             if isinstance(getattr(model_instance, field_name), Model):
69                 if get_url(getattr(model_instance, field_name)) != value:
70                     needs_update = True
71             elif getattr(model_instance, field_name) != value:
72                 needs_update = True
73         return needs_update
74 
75     def upstream_distributions(self, labels=None):
76         if labels:
77             params = {"pulp_label_select": labels}
78         else:
79             params = {}
80         offset = 0
81         list_size = 100
82         while True:
83             distributions = self.distribution_ctx_cls(self.pulp_ctx).list(list_size, offset, params)
84             for distro in distributions:
85                 yield distro
86             if len(distributions) < list_size:
87                 break
88             offset += list_size
89 
90     def url(self, upstream_distribution):
91         return upstream_distribution["base_url"]
92 
93     def remote_extra_fields(self, upstream_distribution):
94         return {}
95 
96     def create_or_update_remote(self, upstream_distribution):
97         if not upstream_distribution["repository"] and not upstream_distribution["publication"]:
98             return None
99         url = self.url(upstream_distribution)
100         remote_fields_dict = {"url": url}
101         remote_fields_dict.update(self.tls_settings)
102         remote_fields_dict.update(self.remote_extra_fields(upstream_distribution))
103 
104         # Check if there is a remote pointing to this distribution
105         try:
106             remote = self.remote_model_cls.objects.get(
107                 name=upstream_distribution["name"], pulp_domain=self.domain
108             )
109             needs_update = self.needs_update(remote_fields_dict, remote)
110             if needs_update:
111                 dispatch(
112                     general_update,
113                     task_group=self.task_group,
114                     exclusive_resources=[remote],
115                     args=(remote.pk, self.app_label, self.remote_serializer_name),
116                     kwargs={"data": remote_fields_dict, "partial": True},
117                 )
118         except self.remote_model_cls.DoesNotExist:
119             # Create the remote
120             remote = self.remote_model_cls(name=upstream_distribution["name"], **remote_fields_dict)
121             remote.save()
122 
123         return remote
124 
125     def repository_extra_fields(self, remote):
126         return {}
127 
128     def create_or_update_repository(self, remote):
129         try:
130             repository = self.repository_model_cls.objects.get(
131                 name=remote.name, pulp_domain=self.domain
132             )
133             repo_fields_dict = self.repository_extra_fields(remote)
134             needs_update = self.needs_update(repo_fields_dict, repository)
135             if needs_update:
136                 dispatch(
137                     general_update,
138                     task_group=self.task_group,
139                     exclusive_resources=[repository],
140                     args=(repository.pk, self.app_label, self.repository_serializer_name),
141                     kwargs={"data": repo_fields_dict, "partial": True},
142                 )
143         except self.repository_model_cls.DoesNotExist:
144             repository = self.repository_model_cls(
145                 name=remote.name, **self.repository_extra_fields(remote)
146             )
147             repository.save()
148         return repository
149 
150     def distribution_data(self, repository, upstream_distribution):
151         """
152         Return the fields that need to be updated/cleared on distributions for idempotence.
153         """
154         return {
155             "repository": get_url(repository),
156             "publication": None,
157             "base_path": upstream_distribution["base_path"],
158         }
159 
160     def create_or_update_distribution(self, repository, upstream_distribution):
161         distribution_data = self.distribution_data(repository, upstream_distribution)
162         try:
163             distro = self.distribution_model_cls.objects.get(
164                 name=upstream_distribution["name"], pulp_domain=self.domain
165             )
166             # Check that the distribution has the right repository associated
167             needs_update = self.needs_update(distribution_data, distro)
168             if needs_update:
169                 # Update the distribution
170                 dispatch(
171                     general_update,
172                     task_group=self.task_group,
173                     shared_resources=[repository],
174                     exclusive_resources=self.distros_uris,
175                     args=(distro.pk, self.app_label, self.distribution_serializer_name),
176                     kwargs={
177                         "data": distribution_data,
178                         "partial": True,
179                     },
180                 )
181         except self.distribution_model_cls.DoesNotExist:
182             # Dispatch a task to create the distribution
183             distribution_data["name"] = upstream_distribution["name"]
184             dispatch(
185                 general_create,
186                 task_group=self.task_group,
187                 shared_resources=[repository],
188                 exclusive_resources=self.distros_uris,
189                 args=(self.app_label, self.distribution_serializer_name),
190                 kwargs={"data": distribution_data},
191             )
192 
193     def sync_params(self, repository, remote):
194         """This method returns a dict that will be passed as kwargs to the sync task."""
195         raise NotImplementedError("Each replicator must supply its own sync params.")
196 
197     def sync(self, repository, remote):
198         dispatch(
199             self.sync_task,
200             task_group=self.task_group,
201             shared_resources=[remote],
202             exclusive_resources=[repository],
203             kwargs=self.sync_params(repository, remote),
204         )
205 
206     def remove_missing(self, names):
207         # Remove all distributions with names not present in the list of names
208         # Perform this in an extra task, because we hold a big lock here.
209         distribution_ids = [
210             (distribution.pk, self.app_label, self.distribution_serializer_name)
211             for distribution in self.distribution_model_cls.objects.filter(
212                 pulp_domain=self.domain
213             ).exclude(name__in=names)
214         ]
215         if distribution_ids:
216             dispatch(
217                 general_multi_delete,
218                 task_group=self.task_group,
219                 exclusive_resources=self.distros_uris,
220                 args=(distribution_ids,),
221             )
222 
223         # Remove all the repositories and remotes of the missing distributions
224         repositories = list(
225             self.repository_model_cls.objects.filter(
226                 pulp_domain=self.domain, user_hidden=False
227             ).exclude(name__in=names)
228         )
229         repository_ids = [
230             (repo.pk, self.app_label, self.repository_serializer_name) for repo in repositories
231         ]
232 
233         remotes = list(
234             self.remote_model_cls.objects.filter(pulp_domain=self.domain).exclude(name__in=names)
235         )
236         remote_ids = [
237             (remote.pk, self.app_label, self.remote_serializer_name) for remote in remotes
238         ]
239 
240         if repository_ids or remote_ids:
241             dispatch(
242                 general_multi_delete,
243                 task_group=self.task_group,
244                 exclusive_resources=repositories + remotes,
245                 args=(repository_ids + remote_ids,),
246             )
247
```
--- 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/app/replica.py b/pulpcore/app/replica.py
--- a/pulpcore/app/replica.py
+++ b/pulpcore/app/replica.py
@@ -94,7 +94,9 @@
         return {}
 
     def create_or_update_remote(self, upstream_distribution):
-        if not upstream_distribution["repository"] and not upstream_distribution["publication"]:
+        if not upstream_distribution.get("repository") and not upstream_distribution.get(
+            "publication"
+        ):
             return None
         url = self.url(upstream_distribution)
         remote_fields_dict = {"url": url}

{"golden_diff": "diff --git a/pulpcore/app/replica.py b/pulpcore/app/replica.py\n--- a/pulpcore/app/replica.py\n+++ b/pulpcore/app/replica.py\n@@ -94,7 +94,9 @@\n         return {}\n \n     def create_or_update_remote(self, upstream_distribution):\n-        if not upstream_distribution[\"repository\"] and not upstream_distribution[\"publication\"]:\n+        if not upstream_distribution.get(\"repository\") and not upstream_distribution.get(\n+            \"publication\"\n+        ):\n             return None\n         url = self.url(upstream_distribution)\n         remote_fields_dict = {\"url\": url}\n", "issue": "Replication assumes every plugin supports Publications\nThe ```Replicator``` class, which is being subclassed inside plugins to support the replication feature, assumes that every plugin supports publications as it tries to access ```upstream_distribution[\"publication\"]``` inside one of its methods (for some plugins, the dictionary simply doesn't contain the \"publication\" key so an exception gets raised). This forces certain subclasses of ```Replicator``` to create workarounds or rewrite the given method.\r\n\r\nI propose making the method more general, removing such assumptions.\r\n\r\nRelevant code: https://github.com/pulp/pulpcore/blob/c31a39c186fd734e9e0409321735d7d6509a12ee/pulpcore/app/replica.py#L97C9-L97C96\r\n\n", "before_files": [{"content": "from django.db.models import Model\nimport logging\n\nfrom pulp_glue.common.context import PulpContext\nfrom pulpcore.tasking.tasks import dispatch\nfrom pulpcore.app.tasks.base import (\n    general_update,\n    general_create,\n    general_multi_delete,\n)\nfrom pulpcore.plugin.util import get_url, get_domain\n\n_logger = logging.getLogger(__name__)\n\n\nclass ReplicaContext(PulpContext):\n    def __init__(self, **kwargs):\n        super().__init__(**kwargs)\n        self.out_buf = \"\"\n        self.err_buf = \"\"\n\n    def echo(self, message: str, nl: bool = True, err: bool = False) -> None:\n        if err:\n            self.err_buf += message\n            if nl:\n                _logger.warn(\"{}\", self.err_buf)\n                self.err_buf = \"\"\n        else:\n            self.out_buf += message\n            if nl:\n                _logger.info(\"{}\", self.out_buf)\n                self.out_buf = \"\"\n\n\nclass Replicator:\n    distribution_ctx_cls = None\n    repository_ctx_cls = None\n    publication_ctx_cls = None\n    remote_model_cls = None\n    repository_model_cls = None\n    distribution_model_cls = None\n    distribution_serializer_name = None\n    repository_serializer_name = None\n    remote_serializer_name = None\n    app_label = None\n    sync_task = None\n\n    def __init__(self, pulp_ctx, task_group, tls_settings):\n        \"\"\"\n        :param pulp_ctx: PulpReplicaContext\n        :param task_group: TaskGroup\n        :param ca_cert: str\n        \"\"\"\n        self.pulp_ctx = pulp_ctx\n        self.task_group = task_group\n        self.tls_settings = tls_settings\n        self.domain = get_domain()\n        self.distros_uris = [f\"pdrn:{self.domain.pulp_id}:distributions\"]\n\n    @staticmethod\n    def needs_update(fields_dict, model_instance):\n        \"\"\"\n        Compares a Model instance's attributes against a dictionary where keys are attribute\n        names and values are expected values.\n        \"\"\"\n        needs_update = False\n        for field_name, value in fields_dict.items():\n            if isinstance(getattr(model_instance, field_name), Model):\n                if get_url(getattr(model_instance, field_name)) != value:\n                    needs_update = True\n            elif getattr(model_instance, field_name) != value:\n                needs_update = True\n        return needs_update\n\n    def upstream_distributions(self, labels=None):\n        if labels:\n            params = {\"pulp_label_select\": labels}\n        else:\n            params = {}\n        offset = 0\n        list_size = 100\n        while True:\n            distributions = self.distribution_ctx_cls(self.pulp_ctx).list(list_size, offset, params)\n            for distro in distributions:\n                yield distro\n            if len(distributions) < list_size:\n                break\n            offset += list_size\n\n    def url(self, upstream_distribution):\n        return upstream_distribution[\"base_url\"]\n\n    def remote_extra_fields(self, upstream_distribution):\n        return {}\n\n    def create_or_update_remote(self, upstream_distribution):\n        if not upstream_distribution[\"repository\"] and not upstream_distribution[\"publication\"]:\n            return None\n        url = self.url(upstream_distribution)\n        remote_fields_dict = {\"url\": url}\n        remote_fields_dict.update(self.tls_settings)\n        remote_fields_dict.update(self.remote_extra_fields(upstream_distribution))\n\n        # Check if there is a remote pointing to this distribution\n        try:\n            remote = self.remote_model_cls.objects.get(\n                name=upstream_distribution[\"name\"], pulp_domain=self.domain\n            )\n            needs_update = self.needs_update(remote_fields_dict, remote)\n            if needs_update:\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    exclusive_resources=[remote],\n                    args=(remote.pk, self.app_label, self.remote_serializer_name),\n                    kwargs={\"data\": remote_fields_dict, \"partial\": True},\n                )\n        except self.remote_model_cls.DoesNotExist:\n            # Create the remote\n            remote = self.remote_model_cls(name=upstream_distribution[\"name\"], **remote_fields_dict)\n            remote.save()\n\n        return remote\n\n    def repository_extra_fields(self, remote):\n        return {}\n\n    def create_or_update_repository(self, remote):\n        try:\n            repository = self.repository_model_cls.objects.get(\n                name=remote.name, pulp_domain=self.domain\n            )\n            repo_fields_dict = self.repository_extra_fields(remote)\n            needs_update = self.needs_update(repo_fields_dict, repository)\n            if needs_update:\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    exclusive_resources=[repository],\n                    args=(repository.pk, self.app_label, self.repository_serializer_name),\n                    kwargs={\"data\": repo_fields_dict, \"partial\": True},\n                )\n        except self.repository_model_cls.DoesNotExist:\n            repository = self.repository_model_cls(\n                name=remote.name, **self.repository_extra_fields(remote)\n            )\n            repository.save()\n        return repository\n\n    def distribution_data(self, repository, upstream_distribution):\n        \"\"\"\n        Return the fields that need to be updated/cleared on distributions for idempotence.\n        \"\"\"\n        return {\n            \"repository\": get_url(repository),\n            \"publication\": None,\n            \"base_path\": upstream_distribution[\"base_path\"],\n        }\n\n    def create_or_update_distribution(self, repository, upstream_distribution):\n        distribution_data = self.distribution_data(repository, upstream_distribution)\n        try:\n            distro = self.distribution_model_cls.objects.get(\n                name=upstream_distribution[\"name\"], pulp_domain=self.domain\n            )\n            # Check that the distribution has the right repository associated\n            needs_update = self.needs_update(distribution_data, distro)\n            if needs_update:\n                # Update the distribution\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    shared_resources=[repository],\n                    exclusive_resources=self.distros_uris,\n                    args=(distro.pk, self.app_label, self.distribution_serializer_name),\n                    kwargs={\n                        \"data\": distribution_data,\n                        \"partial\": True,\n                    },\n                )\n        except self.distribution_model_cls.DoesNotExist:\n            # Dispatch a task to create the distribution\n            distribution_data[\"name\"] = upstream_distribution[\"name\"]\n            dispatch(\n                general_create,\n                task_group=self.task_group,\n                shared_resources=[repository],\n                exclusive_resources=self.distros_uris,\n                args=(self.app_label, self.distribution_serializer_name),\n                kwargs={\"data\": distribution_data},\n            )\n\n    def sync_params(self, repository, remote):\n        \"\"\"This method returns a dict that will be passed as kwargs to the sync task.\"\"\"\n        raise NotImplementedError(\"Each replicator must supply its own sync params.\")\n\n    def sync(self, repository, remote):\n        dispatch(\n            self.sync_task,\n            task_group=self.task_group,\n            shared_resources=[remote],\n            exclusive_resources=[repository],\n            kwargs=self.sync_params(repository, remote),\n        )\n\n    def remove_missing(self, names):\n        # Remove all distributions with names not present in the list of names\n        # Perform this in an extra task, because we hold a big lock here.\n        distribution_ids = [\n            (distribution.pk, self.app_label, self.distribution_serializer_name)\n            for distribution in self.distribution_model_cls.objects.filter(\n                pulp_domain=self.domain\n            ).exclude(name__in=names)\n        ]\n        if distribution_ids:\n            dispatch(\n                general_multi_delete,\n                task_group=self.task_group,\n                exclusive_resources=self.distros_uris,\n                args=(distribution_ids,),\n            )\n\n        # Remove all the repositories and remotes of the missing distributions\n        repositories = list(\n            self.repository_model_cls.objects.filter(\n                pulp_domain=self.domain, user_hidden=False\n            ).exclude(name__in=names)\n        )\n        repository_ids = [\n            (repo.pk, self.app_label, self.repository_serializer_name) for repo in repositories\n        ]\n\n        remotes = list(\n            self.remote_model_cls.objects.filter(pulp_domain=self.domain).exclude(name__in=names)\n        )\n        remote_ids = [\n            (remote.pk, self.app_label, self.remote_serializer_name) for remote in remotes\n        ]\n\n        if repository_ids or remote_ids:\n            dispatch(\n                general_multi_delete,\n                task_group=self.task_group,\n                exclusive_resources=repositories + remotes,\n                args=(repository_ids + remote_ids,),\n            )\n", "path": "pulpcore/app/replica.py"}], "after_files": [{"content": "from django.db.models import Model\nimport logging\n\nfrom pulp_glue.common.context import PulpContext\nfrom pulpcore.tasking.tasks import dispatch\nfrom pulpcore.app.tasks.base import (\n    general_update,\n    general_create,\n    general_multi_delete,\n)\nfrom pulpcore.plugin.util import get_url, get_domain\n\n_logger = logging.getLogger(__name__)\n\n\nclass ReplicaContext(PulpContext):\n    def __init__(self, **kwargs):\n        super().__init__(**kwargs)\n        self.out_buf = \"\"\n        self.err_buf = \"\"\n\n    def echo(self, message: str, nl: bool = True, err: bool = False) -> None:\n        if err:\n            self.err_buf += message\n            if nl:\n                _logger.warn(\"{}\", self.err_buf)\n                self.err_buf = \"\"\n        else:\n            self.out_buf += message\n            if nl:\n                _logger.info(\"{}\", self.out_buf)\n                self.out_buf = \"\"\n\n\nclass Replicator:\n    distribution_ctx_cls = None\n    repository_ctx_cls = None\n    publication_ctx_cls = None\n    remote_model_cls = None\n    repository_model_cls = None\n    distribution_model_cls = None\n    distribution_serializer_name = None\n    repository_serializer_name = None\n    remote_serializer_name = None\n    app_label = None\n    sync_task = None\n\n    def __init__(self, pulp_ctx, task_group, tls_settings):\n        \"\"\"\n        :param pulp_ctx: PulpReplicaContext\n        :param task_group: TaskGroup\n        :param ca_cert: str\n        \"\"\"\n        self.pulp_ctx = pulp_ctx\n        self.task_group = task_group\n        self.tls_settings = tls_settings\n        self.domain = get_domain()\n        self.distros_uris = [f\"pdrn:{self.domain.pulp_id}:distributions\"]\n\n    @staticmethod\n    def needs_update(fields_dict, model_instance):\n        \"\"\"\n        Compares a Model instance's attributes against a dictionary where keys are attribute\n        names and values are expected values.\n        \"\"\"\n        needs_update = False\n        for field_name, value in fields_dict.items():\n            if isinstance(getattr(model_instance, field_name), Model):\n                if get_url(getattr(model_instance, field_name)) != value:\n                    needs_update = True\n            elif getattr(model_instance, field_name) != value:\n                needs_update = True\n        return needs_update\n\n    def upstream_distributions(self, labels=None):\n        if labels:\n            params = {\"pulp_label_select\": labels}\n        else:\n            params = {}\n        offset = 0\n        list_size = 100\n        while True:\n            distributions = self.distribution_ctx_cls(self.pulp_ctx).list(list_size, offset, params)\n            for distro in distributions:\n                yield distro\n            if len(distributions) < list_size:\n                break\n            offset += list_size\n\n    def url(self, upstream_distribution):\n        return upstream_distribution[\"base_url\"]\n\n    def remote_extra_fields(self, upstream_distribution):\n        return {}\n\n    def create_or_update_remote(self, upstream_distribution):\n        if not upstream_distribution.get(\"repository\") and not upstream_distribution.get(\n            \"publication\"\n        ):\n            return None\n        url = self.url(upstream_distribution)\n        remote_fields_dict = {\"url\": url}\n        remote_fields_dict.update(self.tls_settings)\n        remote_fields_dict.update(self.remote_extra_fields(upstream_distribution))\n\n        # Check if there is a remote pointing to this distribution\n        try:\n            remote = self.remote_model_cls.objects.get(\n                name=upstream_distribution[\"name\"], pulp_domain=self.domain\n            )\n            needs_update = self.needs_update(remote_fields_dict, remote)\n            if needs_update:\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    exclusive_resources=[remote],\n                    args=(remote.pk, self.app_label, self.remote_serializer_name),\n                    kwargs={\"data\": remote_fields_dict, \"partial\": True},\n                )\n        except self.remote_model_cls.DoesNotExist:\n            # Create the remote\n            remote = self.remote_model_cls(name=upstream_distribution[\"name\"], **remote_fields_dict)\n            remote.save()\n\n        return remote\n\n    def repository_extra_fields(self, remote):\n        return {}\n\n    def create_or_update_repository(self, remote):\n        try:\n            repository = self.repository_model_cls.objects.get(\n                name=remote.name, pulp_domain=self.domain\n            )\n            repo_fields_dict = self.repository_extra_fields(remote)\n            needs_update = self.needs_update(repo_fields_dict, repository)\n            if needs_update:\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    exclusive_resources=[repository],\n                    args=(repository.pk, self.app_label, self.repository_serializer_name),\n                    kwargs={\"data\": repo_fields_dict, \"partial\": True},\n                )\n        except self.repository_model_cls.DoesNotExist:\n            repository = self.repository_model_cls(\n                name=remote.name, **self.repository_extra_fields(remote)\n            )\n            repository.save()\n        return repository\n\n    def distribution_data(self, repository, upstream_distribution):\n        \"\"\"\n        Return the fields that need to be updated/cleared on distributions for idempotence.\n        \"\"\"\n        return {\n            \"repository\": get_url(repository),\n            \"publication\": None,\n            \"base_path\": upstream_distribution[\"base_path\"],\n        }\n\n    def create_or_update_distribution(self, repository, upstream_distribution):\n        distribution_data = self.distribution_data(repository, upstream_distribution)\n        try:\n            distro = self.distribution_model_cls.objects.get(\n                name=upstream_distribution[\"name\"], pulp_domain=self.domain\n            )\n            # Check that the distribution has the right repository associated\n            needs_update = self.needs_update(distribution_data, distro)\n            if needs_update:\n                # Update the distribution\n                dispatch(\n                    general_update,\n                    task_group=self.task_group,\n                    shared_resources=[repository],\n                    exclusive_resources=self.distros_uris,\n                    args=(distro.pk, self.app_label, self.distribution_serializer_name),\n                    kwargs={\n                        \"data\": distribution_data,\n                        \"partial\": True,\n                    },\n                )\n        except self.distribution_model_cls.DoesNotExist:\n            # Dispatch a task to create the distribution\n            distribution_data[\"name\"] = upstream_distribution[\"name\"]\n            dispatch(\n                general_create,\n                task_group=self.task_group,\n                shared_resources=[repository],\n                exclusive_resources=self.distros_uris,\n                args=(self.app_label, self.distribution_serializer_name),\n                kwargs={\"data\": distribution_data},\n            )\n\n    def sync_params(self, repository, remote):\n        \"\"\"This method returns a dict that will be passed as kwargs to the sync task.\"\"\"\n        raise NotImplementedError(\"Each replicator must supply its own sync params.\")\n\n    def sync(self, repository, remote):\n        dispatch(\n            self.sync_task,\n            task_group=self.task_group,\n            shared_resources=[remote],\n            exclusive_resources=[repository],\n            kwargs=self.sync_params(repository, remote),\n        )\n\n    def remove_missing(self, names):\n        # Remove all distributions with names not present in the list of names\n        # Perform this in an extra task, because we hold a big lock here.\n        distribution_ids = [\n            (distribution.pk, self.app_label, self.distribution_serializer_name)\n            for distribution in self.distribution_model_cls.objects.filter(\n                pulp_domain=self.domain\n            ).exclude(name__in=names)\n        ]\n        if distribution_ids:\n            dispatch(\n                general_multi_delete,\n                task_group=self.task_group,\n                exclusive_resources=self.distros_uris,\n                args=(distribution_ids,),\n            )\n\n        # Remove all the repositories and remotes of the missing distributions\n        repositories = list(\n            self.repository_model_cls.objects.filter(\n                pulp_domain=self.domain, user_hidden=False\n            ).exclude(name__in=names)\n        )\n        repository_ids = [\n            (repo.pk, self.app_label, self.repository_serializer_name) for repo in repositories\n        ]\n\n        remotes = list(\n            self.remote_model_cls.objects.filter(pulp_domain=self.domain).exclude(name__in=names)\n        )\n        remote_ids = [\n            (remote.pk, self.app_label, self.remote_serializer_name) for remote in remotes\n        ]\n\n        if repository_ids or remote_ids:\n            dispatch(\n                general_multi_delete,\n                task_group=self.task_group,\n                exclusive_resources=repositories + remotes,\n                args=(repository_ids + remote_ids,),\n            )\n", "path": "pulpcore/app/replica.py"}]}