<commit_before>#! /usr/bin/env python

from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from json import load, dump
from pika import BlockingConnection, ConnectionParameters
from psycopg2 import connect
from SocketServer import ForkingMixIn
from sys import argv
from time import time as now
from traceback import print_exc
from urlparse import urlparse, parse_qs

ALL_TOPICS_SQL = 'SELECT DISTINCT topic FROM facts;'
LAST_TEN_SQL = '''
    SELECT id, topic, ts, content FROM
        (SELECT id, topic, round(extract(epoch from ts)) AS ts, content
             FROM facts WHERE topic = %s ORDER BY id DESC LIMIT 10)
        AS lastten
    ORDER BY id ASC;
'''
SINCE_ID_SQL = '''
SELECT id, topic, round(extract(epoch from ts)), content
    FROM facts WHERE topic = %s and id >= %s;
'''
QUEUE_URL_TEMPLATE = '%s/queues/%s'

with open(argv[1]) as config_file:
    config = load(config_file)

class Alarm:
    def __init__(self, duration):
        if duration is not None:
            self.alarm_time = now() + duration
        else:
            self.alarm_time = None

    def is_ringing(self):
        return self.alarm_time is not None and now() > self.alarm_time

class MyHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed_url = urlparse(self.path)
        q_comp = parse_qs(parsed_url.query)
        path_elements = parsed_url.path.split('/')
        if len(path_elements) < 2:
            self.bad_request()
        elif path_elements[1] == 'queues' and len(path_elements) > 2:
            self.get_from_queue(path_elements[2])
        elif path_elements[1] == 'topics':
            if len(path_elements) == 2 or path_elements[2] == '':
                self.all_topics()
            elif 'from_id' in q_comp and q_comp['from_id'][0].isdigit():
                self.facts_since(path_elements[2], int(q_comp['from_id'][0]))
            elif len(q_comp) == 0:
                self.facts_since(path_elements[2])
            else:
                self.bad_request()
        else:
            self.bad_request()
    
    def do_POST(self):
        path_elements = self.path.split('/')
        if len(path_elements) < 3 or path_elements[1] != 'topics':
            self.bad_request()
        elif len(path_elements) < 4 or path_elements[3] == '':
            self.publish(path_elements[2])
        elif path_elements[3] == 'queue':
            self.make_queue(path_elements[2])
        else:
            self.bad_request()

    def bad_request(self):
        self.send_response(400)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()

    def all_topics(self):
        """Return a list of all topics."""
        conn = connect(host=config['pg_host'], database=config['pg_database'],
                       user=config['pg_user'], password=config['pg_password'])
        cursor = conn.cursor()
        try:
            cursor.execute(ALL_TOPICS_SQL)
            topics = [row[0] for row in cursor.fetchall()]
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            dump(topics, self.wfile)
        except Exception as e:
            print 'Exception: %s' % e.message
            print_exc()
        finally:
            cursor.close()
            conn.close()

    def facts_since(self, topic, ident=None):
        print "topic %s, id %s" % (topic, ident)
        """Get all facts for topic since the given id, or last 10 if None"""
        conn = connect(host=config['pg_host'], database=config['pg_database'],
                       user=config['pg_user'], password=config['<PASSWORD>'])
        cursor = conn.cursor()
        try:
            if ident is None:
                cursor.execute(LAST_TEN_SQL, (topic,))
            else:
                cursor.execute(SINCE_ID_SQL, (topic, ident))
            facts = cursor.fetchall()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            dump(facts, self.wfile)
        except Exception as e:
            print 'Exception: %s' % e.message
            print_exc()
        finally:
            cursor.close()
            conn.close()

    def make_queue(self, topic):
        """Create a queue for the given topic."""
        connection = BlockingConnection(ConnectionParameters(host=config['rabbit_host'],
                                                             port=config['rabbit_port']))
        channel = connection.channel()
        channel.exchange_declare(exchange=config['exchange'], type='topic')
        queue = channel.queue_declare().method.queue
        channel.queue_bind(exchange=config['exchange'],
                           queue=queue,
                           routing_key=topic)        
        connection.close()

        out = {'queue_name': queue,
               'queue_url': QUEUE_URL_TEMPLATE % (config['web_url'], queue,)}
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        dump(out, self.wfile)

    def wait_on_queue(self, channel, queue):
        alarm = Alarm(10)
        while True:
            fact = channel.basic_get(queue=queue, no_ack=True)[2]
            if fact is not None:
                return fact
            if alarm.is_ringing():
                return None

    def get_from_queue(self, queue):
        """Get fact from front of queue."""
        connection = BlockingConnection(ConnectionParameters(host=config['rabbit_host'],
                                                             port=config['rabbit_port']))
        channel = connection.channel()
        channel.exchange_declare(exchange=config['exchange'], type='topic')
        fact = self.wait_on_queue(channel, queue)
        connection.close()

        if fact is not None:
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(fact)
        else:
            self.send_response(204)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()

    def publish(self, topic):
        """Publish the given fact."""
        content_len = int(self.headers.getheader('content-length', 0))
        fact = self.rfile.read(content_len)

        connection = BlockingConnection(ConnectionParameters(host=config['rabbit_host'],
                                                             port=config['rabbit_port']))
        channel = connection.channel()
        channel.exchange_declare(exchange=config['exchange'], type='topic')

        channel.basic_publish(exchange=config['exchange'],
                              routing_key=topic,
                              body=fact)
        connection.close()

        self.send_response(202)
        self.send_header("Content-Type", "text/plain")
        self.end_headers()

class ForkingServer(ForkingMixIn, HTTPServer):
    pass

if __name__ == '__main__':
    httpd = ForkingServer((config['web_host'], config['web_port']), MyHandler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()
<commit_msg>revised API and simplified httpserver<commit_after>#! /usr/bin/env python

from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from json import load, dumps
from pika import BlockingConnection, ConnectionParameters
from psycopg2 import connect
from SocketServer import ForkingMixIn
from sys import argv
from time import time as now
from traceback import print_exc
from urlparse import urlparse, parse_qs

ALL_TOPICS_SQL = 'SELECT DISTINCT topic FROM facts;'
LAST_TEN_SQL = '''
    SELECT id, topic, ts, content FROM
        (SELECT id, topic, round(extract(epoch from ts)) AS ts, content
             FROM facts WHERE topic = %s ORDER BY id DESC LIMIT 10)
        AS lastten
    ORDER BY id ASC;
'''
AFTER_ID_SQL = '''
SELECT id, topic, round(extract(epoch from ts)), content
    FROM facts WHERE topic = %s and id > %s;
'''
RETRIEVAL_URL_TEMPLATE = '%s/topics/%s/facts?subscription_id=/%s'

with open(argv[1]) as config_file:
    config = load(config_file)

class Alarm:
    def __init__(self, duration):
        if duration is not None:
            self.alarm_time = now() + duration
        else:
            self.alarm_time = None

    def is_ringing(self):
        return self.alarm_time is not None and now() > self.alarm_time

def run_sql(sql, parameters):
    """Run SQL and return result"""
    conn = connect(host=config['pg_host'], database=config['pg_database'],
                   user=config['pg_user'], password=config['pg_password'])
    cursor = conn.cursor()
    try:
        cursor.execute(sql, parameters)
        return cursor.fetchall()
    except Exception as e:
        print 'Exception: %s' % e.message
        print_exc()
        return []
    finally:
        cursor.close()
        conn.close()    

def rabbit_connect():
    host=config['rabbit_host']
    port=config['rabbit_port']
    return BlockingConnection(ConnectionParameters(host=host, port=port))

def rabbit_channel():
    connection = rabbit_connect()
    channel = connection.channel()
    channel.exchange_declare(exchange=config['exchange'], type='topic')
    return channel

def close_rabbit(channel):
    channel.connection.close()

def all_topics(respond):
    """Return a list of all topics."""
    topics = [row[0] for row in run_sql(ALL_TOPICS_SQL)]
    respond(code=200, content_type='application/json', content=dumps(topics))

def last_ten(respond, topic):
    """Get last ten facts on topic"""
    facts = run_sql(LAST_TEN_SQL, (topic,))
    respond(code=200, content_type='application/json', content=dumps(facts))

def facts_after(respond, topic, id):
    """Get all facts for topic after the given id"""
    facts = run_sql(AFTER_ID_SQL, (topic, id))
    respond(code=200, content_type='application/json', content=dumps(facts))

def subscribe(respond, topic):
    """Create a subscription for the given topic."""
    channel = rabbit_channel()
    queue = channel.queue_declare().method.queue
    channel.queue_bind(exchange=config['exchange'],
                       queue=queue,
                       routing_key=topic)        
    close_rabbit(channel)
    
    retrieval_url = RETRIEVAL_URL_TEMPLATE % (config['web_url'], topic, queue)}
    content = {'subscription_name': queue, 'retrieval_url': retrieval_url}
    respond(code=200, content_type='application/json', content=content)
    
def wait_on_queue(queue):
    channel = rabbit_channel()
    alarm = Alarm(10)
    while True:
        fact = channel.basic_get(queue=queue, no_ack=True)[2]
        close_rabbit(channel)
        if fact is not None:
            return fact
        if alarm.is_ringing():
            return None

def get_from_queue(respond, queue):
    """Get fact from front of queue."""
    fact = self.wait_on_queue(queue)
    if fact is not None:
        respond(code=200, content_type='application/json', content=fact)
    else:
        respond(code=204, content_type='text/plain', content='')
        
def publish(respond, topic, fact):
    """Publish the given fact."""
    channel = rabbit_channel()    
    channel.basic_publish(exchange=config['exchange'],
                          routing_key=topic,
                          body=fact)
    close_rabbit(channel)

    respond(code=202, content_type='text/plain', content=''))

def respond(handler, code, content_type, content):
    handler.send_response(code)
    handler.send_header("Content-Type", content_type)
    handler.end_headers()
    handler.wfile.write(content)

def extract_from_path(path_elements):
    if len(path_elements) < 4 or path_elements[1] != 'topics' \
            or path_elements[2] == '':
        return None, None
    else:
        return path_elements[2:4]

def extract_query_component(name, query_components):
    return query_components.get(name, [None])[0]

class MyHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        parsed_url = urlparse(self.path)
        query_components = parse_qs(parsed_url.query)

        after_id = extract_query_component('after_id')
        sub_id = extract_query_component('subscription_id')
        topic, res = extract_from_path(parsed_url.path.split('/'))

        respond = lambda c, ct, con: respond(self, c, ct, con)

        if topic is None:
            all_topics(respond)
        elif res == 'facts' and after_id is not None and after_id.isdigit():
            facts_after(respond, topic, int(after_id))
        elif res == 'facts' and sub_id is not None:
            get_from_queue(respond, sub_id)
        elif res == 'facts':
            last_ten(respond, topic)
        else:
            respond(400, 'text/plain', '')

    def do_POST(self):
        content_len = int(self.headers.getheader('content-length', 0))
        data = self.rfile.read(content_len)

        parsed_url = urlparse(self.path)
        query_components = parse_qs(parsed_url.query)
        topic, res = extract_from_path(parsed_url.path.split('/'))

        respond = lambda c, ct, con: respond(self, c, ct, con)

        if res == 'subscription':
            subscribe(respond, topic)
        elif res == 'facts':
            publish(respond, topic, data)
        else:
            respond(400, 'text/plain', '')

class ForkingServer(ForkingMixIn, HTTPServer):
    pass

if __name__ == '__main__':
    httpd = ForkingServer((config['web_host'], config['web_port']), MyHandler)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    httpd.server_close()

{'fraction_non_alphanumeric': 0.11320061705722471, 'fraction_numerical': 0.006831704987879233, 'mean_word_length': 2.5333506358681546, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1730553', 'n_tokens_mistral': 4046, 'n_tokens_neox': 3924, 'n_words': 933}