"""Configures a Kafka Connector for Postgres Station data"""
import json
import logging

import requests
from settings import Settings

logger = logging.getLogger(__name__)


KAFKA_CONNECT_URL = f"{Settings.URLs.KAFKA_CONNECT_URL}/connectors"
CONNECTOR_NAME = "stations"


def configure_connector():
    """Starts and configures the Kafka Connect connector"""
    logging.debug("Creating or updating kafka connect connector...")

    resp = requests.get(f"{KAFKA_CONNECT_URL}/{CONNECTOR_NAME}")
    if resp.status_code == 200:
        logging.debug("Connector already created skipping recreation")
        return

    config = {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "false",
        "topic.prefix": "com.connect.transportation.",
        "connection.url": "jdbc:postgresql://postgres:5432/cta",
        "connection.user": "cta_admin",
        "connection.password": "chicago",
        "batch.max.rows": "500",
        "table.whitelist": "stations",
        "poll.interval.ms": "5000",  # Poll every 5 seconds
        "mode": "incrementing",
        "incrementing.column.name": "stop_id",
    }
    # TODO: Complete the Kafka Connect Config below.
    # Directions: Use the JDBC Source Connector to connect to Postgres. Load the `stations` table
    # using incrementing mode, with `stop_id` as the incrementing column name.
    # Make sure to think about what an appropriate topic prefix would be, and how frequently Kafka
    # Connect should run this connector (hint: not very often!)

    data = json.dumps({"name": CONNECTOR_NAME, "config": config})

    resp = requests.post(
        KAFKA_CONNECT_URL,
        headers={"Content-Type": "application/json"},
        data=data,
    )

    # Ensure a healthy response was given
    resp.raise_for_status()
    logging.info("-------Connector created successfully-------")


if __name__ == "__main__":
    configure_connector()

[((8, 9, 8, 36), 'logging.getLogger', 'logging.getLogger', ({(8, 27, 8, 35): '__name__'}, {}), '(__name__)', False, 'import logging\n'), ((17, 4, 17, 68), 'logging.debug', 'logging.debug', ({(17, 18, 17, 67): '"""Creating or updating kafka connect connector..."""'}, {}), "('Creating or updating kafka connect connector...')", False, 'import logging\n'), ((19, 11, 19, 64), 'requests.get', 'requests.get', ({(19, 24, 19, 63): 'f"""{KAFKA_CONNECT_URL}/{CONNECTOR_NAME}"""'}, {}), "(f'{KAFKA_CONNECT_URL}/{CONNECTOR_NAME}')", False, 'import requests\n'), ((46, 11, 46, 65), 'json.dumps', 'json.dumps', ({(46, 22, 46, 64): "{'name': CONNECTOR_NAME, 'config': config}"}, {}), "({'name': CONNECTOR_NAME, 'config': config})", False, 'import json\n'), ((48, 11, 52, 5), 'requests.post', 'requests.post', (), '', False, 'import requests\n'), ((56, 4, 56, 64), 'logging.info', 'logging.info', ({(56, 17, 56, 63): '"""-------Connector created successfully-------"""'}, {}), "('-------Connector created successfully-------')", False, 'import logging\n'), ((21, 8, 21, 70), 'logging.debug', 'logging.debug', ({(21, 22, 21, 69): '"""Connector already created skipping recreation"""'}, {}), "('Connector already created skipping recreation')", False, 'import logging\n')]