<commit_before>"""
    consumer.py

    Copyright (c) 2013-2014 Snowplow Analytics Ltd. All rights reserved.

    This program is licensed to you under the Apache License Version 2.0,
    and you may not use this file except in compliance with the Apache License
    Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
    http://www.apache.org/licenses/LICENSE-2.0.

    Unless required by applicable law or agreed to in writing,
    software distributed under the Apache License Version 2.0 is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
    express or implied. See the Apache License Version 2.0 for the specific
    language governing permissions and limitations there under.

    Authors: <NAME>, <NAME>, <NAME>
    Copyright: Copyright (c) 2013-2014 Snowplow Analytics Ltd
    License: Apache License Version 2.0
"""

import requests
import json
import threading
import celery
from celery import Celery
from celery.contrib.methods import task
import redis
import logging
from contracts import contract, new_contract

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

DEFAULT_MAX_LENGTH = 10
THREAD_TIMEOUT = 10

new_contract("protocol", lambda x: x == "http" or x == "https")

new_contract("method", lambda x: x == "get" or x == "post")

new_contract("function", lambda x: hasattr(x, "__call__"))

new_contract("redis", lambda x: isinstance(x, (redis.Redis, redis.StrictRedis)))

try:
    # Check whether a custom Celery configuration module named "snowplow_celery_config" exists
    import snowplow_celery_config
    app = Celery()
    app.config_from_object(snowplow_celery_config)

except ImportError:
    # Otherwise configure Celery with default settings
    app = Celery("Snowplow", broker="redis://guest@localhost//")


class Consumer(object):
    """
        Synchronously send Snowplow events to a Snowplow as_collector_uri
        Supports both GET and POST requests
    """

    @contract
    def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_size=None, on_success=None, on_failure=None):
        """
            :param endpoint:    The collector URL. Don't include "http://" - this is done automatically.
            :type  endpoint:    string
            :param protocol:    The protocol to use - http or https. Defaults to http.
            :type  protocol:    protocol
            :param port:        The collector port to connect to
            :type  port:        int | None
            :param method:      The HTTP request method
            :type  method:      method
            :param buffer_size: The maximum number of queued events before the buffer is flushed. Default is 10.
            :type  buffer_size: int | None
            :param on_success:  Callback executed after every HTTP request in a flush has status code 200
                                Gets passed the number of events flushed.
            :type  on_success:  function | None
            :param on_failure:  Callback executed if at least one HTTP request in a flush has status code 200
                                Gets passed two arguments:
                                1) The number of events which were successfully sent
                                2) If method is "post": The unsent data in string form;
                                   If method is "get":  An array of dictionaries corresponding to the unsent events' payloads
            :type  on_failure:  function | None            
        """
        self.endpoint = Consumer.as_collector_uri(endpoint, protocol, port)

        self.method = method

        if buffer_size is None:
            if method == "post":
                buffer_size = DEFAULT_MAX_LENGTH
            else:
                buffer_size = 1
        self.buffer_size = buffer_size
        self.buffer = []

        self.on_success = on_success
        self.on_failure = on_failure

        self.threads = []

    @staticmethod
    @contract
    def as_collector_uri(endpoint, protocol="http", port=None):
        """
            :param endpoint:  The raw endpoint provided by the user
            :type  endpoint:  string
            :param protocol:  The protocol to use - http or https
            :type  protocol:  protocol            
            :param port:      The collector port to connect to
            :type  port:      int | None            
            :rtype:           string
        """
        if port is None:
            return protocol + "://" + endpoint + "/i"
        else:
            return protocol + "://" + endpoint + ":" + str(port) + "/i"

    @contract
    def input(self, payload):
        """
            Adds an event to the buffer.
            If the maximum size has been reached, flushes the buffer.

            :param payload:   The name-value pairs for the event
            :type  payload:   dict(string:*)
        """
        self.buffer.append(payload)
        if len(self.buffer) >= self.buffer_size:
            return self.flush()

    @task(name="Flush")
    def flush(self):
        """
            Sends all events in the buffer to the collector.
        """
        if self.method == "post":
            data = json.dumps(self.buffer)
            buffer_length = len(self.buffer)
            self.buffer = []
            status_code = self.http_post(data).status_code
            if status_code == 200 and self.on_success is not None:
                self.on_success(buffer_length)
            elif self.on_failure is not None:
                self.on_failure(0, data)
            return status_code

        elif self.method == "get":
            success_count = 0
            unsent_requests = []
            status_code = None

            while len(self.buffer) > 0:
                payload = self.buffer.pop()
                status_code = self.http_get(payload).status_code<commit_msg>Updated POST payload to conform to vschema (#85)<commit_after>"""
    consumer.py

    Copyright (c) 2013-2014 Snowplow Analytics Ltd. All rights reserved.

    This program is licensed to you under the Apache License Version 2.0,
    and you may not use this file except in compliance with the Apache License
    Version 2.0. You may obtain a copy of the Apache License Version 2.0 at
    http://www.apache.org/licenses/LICENSE-2.0.

    Unless required by applicable law or agreed to in writing,
    software distributed under the Apache License Version 2.0 is distributed on
    an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
    express or implied. See the Apache License Version 2.0 for the specific
    language governing permissions and limitations there under.

    Authors: <NAME>, <NAME>, <NAME>
    Copyright: Copyright (c) 2013-2014 Snowplow Analytics Ltd
    License: Apache License Version 2.0
"""

import requests
import json
import threading
import celery
from celery import Celery
from celery.contrib.methods import task
import redis
import logging
from contracts import contract, new_contract

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

DEFAULT_MAX_LENGTH = 10
THREAD_TIMEOUT = 10
TRACKER_PROTOCOL_SCHEMA = "com.snowplowanalytics/tracker_protocol/1.0.0"

new_contract("protocol", lambda x: x == "http" or x == "https")

new_contract("method", lambda x: x == "get" or x == "post")

new_contract("function", lambda x: hasattr(x, "__call__"))

new_contract("redis", lambda x: isinstance(x, (redis.Redis, redis.StrictRedis)))

try:
    # Check whether a custom Celery configuration module named "snowplow_celery_config" exists
    import snowplow_celery_config
    app = Celery()
    app.config_from_object(snowplow_celery_config)

except ImportError:
    # Otherwise configure Celery with default settings
    app = Celery("Snowplow", broker="redis://guest@localhost//")


class Consumer(object):
    """
        Synchronously send Snowplow events to a Snowplow as_collector_uri
        Supports both GET and POST requests
    """

    @contract
    def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_size=None, on_success=None, on_failure=None):
        """
            :param endpoint:    The collector URL. Don't include "http://" - this is done automatically.
            :type  endpoint:    string
            :param protocol:    The protocol to use - http or https. Defaults to http.
            :type  protocol:    protocol
            :param port:        The collector port to connect to
            :type  port:        int | None
            :param method:      The HTTP request method
            :type  method:      method
            :param buffer_size: The maximum number of queued events before the buffer is flushed. Default is 10.
            :type  buffer_size: int | None
            :param on_success:  Callback executed after every HTTP request in a flush has status code 200
                                Gets passed the number of events flushed.
            :type  on_success:  function | None
            :param on_failure:  Callback executed if at least one HTTP request in a flush has status code 200
                                Gets passed two arguments:
                                1) The number of events which were successfully sent
                                2) If method is "post": The unsent data in string form;
                                   If method is "get":  An array of dictionaries corresponding to the unsent events' payloads
            :type  on_failure:  function | None            
        """
        self.endpoint = Consumer.as_collector_uri(endpoint, protocol, port)

        self.method = method

        if buffer_size is None:
            if method == "post":
                buffer_size = DEFAULT_MAX_LENGTH
            else:
                buffer_size = 1
        self.buffer_size = buffer_size
        self.buffer = []

        self.on_success = on_success
        self.on_failure = on_failure

        self.threads = []

    @staticmethod
    @contract
    def as_collector_uri(endpoint, protocol="http", port=None):
        """
            :param endpoint:  The raw endpoint provided by the user
            :type  endpoint:  string
            :param protocol:  The protocol to use - http or https
            :type  protocol:  protocol            
            :param port:      The collector port to connect to
            :type  port:      int | None            
            :rtype:           string
        """
        if port is None:
            return protocol + "://" + endpoint + "/i"
        else:
            return protocol + "://" + endpoint + ":" + str(port) + "/i"

    @contract
    def input(self, payload):
        """
            Adds an event to the buffer.
            If the maximum size has been reached, flushes the buffer.

            :param payload:   The name-value pairs for the event
            :type  payload:   dict(string:*)
        """
        if self.method == "post":
            self.buffer.append({"schema": TRACKER_PROTOCOL_SCHEMA, "data": payload})
        else:
            self.buffer.append(payload)

        if len(self.buffer) >= self.buffer_size:
            return self.flush()

    @task(name="Flush")
    def flush(self):
        """
            Sends all events in the buffer to the collector.
        """
        if self.method == "post":
            data = json.dumps(self.buffer)
            buffer_length = len(self.buffer)
            self.buffer = []
            status_code = self.http_post(data).status_code
            if status_code == 200 and self.on_success is not None:
                self.on_success(buffer_length)
            elif self.on_failure is not None:
                self.on_failure(0, data)
            return status_code

        elif self.method == "get":
            success_count = 0
            unsent_requests = []
            status_code = None

            while len(self.buffer) > 0:
                payload = self.buffer.pop()
                status_code = self.http_get(payload).status_code
{'fraction_non_alphanumeric': 0.07778515689855554, 'fraction_numerical': 0.008882616636227794, 'mean_word_length': 2.0568383658969807, 'pattern_counts': {'":': 19, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4526349', 'n_tokens_mistral': 3237, 'n_tokens_neox': 3077, 'n_words': 1144}