def _ensure_logical_slots_primary(self, slots: Dict[str, Any]) -> None:
        """Create any missing logical replication *slots* on the primary.

        If the logical slot already exists, copy state information into the replication slots structure stored in the
        class instance.

        :param slots: Slots that should exist are supplied in a dictionary, mapping slot name to any attributes.
                      The method will only consider slots that have a value that is a dictionary with a key ``type``
                      with a value that is ``logical``.

        """
        # Group logical slots to be created by database name
        logical_slots: Dict[str, Dict[str, Dict[str, Any]]] = defaultdict(dict)
        for name, value in slots.items():
            if value['type'] == 'logical':
                if self._replication_slots.get(name, {}).get('datoid'):
                    self._copy_items(self._replication_slots[name], value)
                else:
                    logical_slots[value['database']][name] = value

        # Create new logical slots
        for database, values in logical_slots.items():
            with self.get_local_connection_cursor(dbname=database) as cur:
                for name, value in values.items():
                    try:
                        cur.execute("SELECT pg_catalog.pg_create_logical_replication_slot(%s, %s)"
                                    " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
                                    " WHERE slot_type = 'logical' AND slot_name = %s)",
                                    (name, value['plugin'], name))
                    except Exception as e:
                        logger.error("Failed to create logical replication slot '%s' plugin='%s': %r",
                                     name, value['plugin'], e)
                        slots.pop(name)
                    self._schedule_load_slots = True
Create any missing logical replication *slots* on the primary.

        If the logical slot already exists, copy state information into the replication slots structure stored in the
        class instance.

        :param slots: Slots that should exist are supplied in a dictionary, mapping slot name to any attributes.
                      The method will only consider slots that have a value that is a dictionary with a key ``type``
                      with a value that is ``logical``.

        
patroni/postgresql/slots.py