def create_replication_slot()

in xray-setup/psycopg2/extras.py [0:0]

• 20 lines of code
• 6 McCabe index (conditional complexity)

    def create_replication_slot(self, slot_name, slot_type=None, output_plugin=None):
        """Create streaming replication slot."""

        command = "CREATE_REPLICATION_SLOT %s " % quote_ident(slot_name, self)

        if slot_type is None:
            slot_type = self.connection.replication_type

        if slot_type == REPLICATION_LOGICAL:
            if output_plugin is None:
                raise psycopg2.ProgrammingError(
                    "output plugin name is required to create "
                    "logical replication slot")

            command += "LOGICAL %s" % quote_ident(output_plugin, self)

        elif slot_type == REPLICATION_PHYSICAL:
            if output_plugin is not None:
                raise psycopg2.ProgrammingError(
                    "cannot specify output plugin name when creating "
                    "physical replication slot")

            command += "PHYSICAL"

        else:
            raise psycopg2.ProgrammingError(
                "unrecognized replication type: %s" % repr(slot_type))

        self.execute(command)