"""Configures KSQL to combine station and turnstile data"""
import json
import logging

import requests

import topic_check


logger = logging.getLogger(__name__)


KSQL_URL = "http://localhost:8088"

KSQL_STATEMENT = """
CREATE TABLE turnstile (
    station_id INT,
    station_name VARCHAR,
    line VARCHAR
) WITH (
    KAFKA_TOPIC='cta.trains.monitor.turnstile',
    VALUE_FORMAT='AVRO',
    KEY='station_id'
);

CREATE TABLE turnstile_summary
WITH (
        KAFKA_TOPIC='cta.trains.monitor.turnstile.summarized'
      , VALUE_FORMAT='json'
    ) 
) AS
        SELECT station_id, count(station_id) as count
        FROM turnstile
        GROUP BY station_id
;
"""


def execute_statement():
    """Executes the KSQL statement against the KSQL API"""
    if topic_check.topic_exists("cta.trains.monitor.turnstile.summarized"):
	logger.info("Turnstile summary topic already exists")
        return

    logger.debug("executing ksql statement...")
    resp = requests.post(
        f"{KSQL_URL}/ksql",
        headers={"Content-Type": "application/vnd.ksql.v1+json"},
        data=json.dumps(
            {
                "ksql": KSQL_STATEMENT,
                "streamsProperties": {"ksql.streams.auto.offset.reset": "earliest"},
            }
        ),
    )

    # Check whether response status reflect successfull run
    resp.raise_for_status()
    if resp.status_code == 200:
        logger.info("Turnstile ksql tables created successfully")
    else:
        logger.warning("Error creating Turnstile ksql tables")
        logger.info(f"Response status code: {resp.status_code}")

 __name__ == "__main__":
    execute_statement()
