<commit_before>import json
import uuid
import time
import asyncio

import requests
from requests.exceptions import HTTPError
import aiohttp

def links_as_dict(links):
    result = {}
    for link in links:
        result[link['relation']] = link['uri']
    return result

class StreamEntry:
    def __init__(self, content):
        self._content = content
        self._links = None
        self.summary = content['summary']

    @property
    def links(self):
        if self._links is None:
            self._links = links_as_dict(self._content['links'])
        return self._links

class StreamPage:
    def __init__(self, content):
        self._content = content 
        self._links = None

    @property
    def links(self):
        if self._links is None:
            self._links = links_as_dict(self._content['links'])
        return self._links

    def entries(self):
        for entry_content in reversed(self._content['entries']):
            yield StreamEntry(entry_content)

class EventStoreClient:

    def __init__(self, host, secure=False, port=2113):
        proto = "https" if secure else "http"
        self.base_url = proto + "://" + host + ":" + str(port)

    def post_events(self, stream_name, events):
        url = self.base_url +  "/streams/" + stream_name
        headers = {'Content-Type': 'application/vnd.eventstore.events+json'}
        response = requests.post(url, headers=headers, data=json.dumps(events))
        if response.status_code >= 400 and response.status_code < 500:
            raise ValueError(response.reason)

    def publish_event(self, stream_name, event_type, data, event_id=None):
        if event_id is None:
            event_id = str(uuid.uuid4())

        event = {
            'eventId': event_id,
            'eventType': event_type,
            'data': data
        }
        self.post_events(stream_name, [event])

    @asyncio.coroutine
    def get_stream_page_async(self, uri):
        # print('getting stream page {}'.format(uri))
        headers = {'Accept': 'application/vnd.eventstore.events+json'}
        response = yield from aiohttp.request('get', uri, headers=headers)
        content = yield from response.json()
        # print('received stream page {}'.format(uri))
        return StreamPage(content)

    @asyncio.coroutine
    def get_stream_head_async(self, stream_name):
        uri = '{}/streams/{}'.format(self.base_url, stream_name)
        return (yield from self.get_stream_page_async(uri))

    @asyncio.coroutine
    def fetch_event_async(self, uri):
        # print('getting event data from {}'.format(uri))
        headers = {'Accept': 'application/json'}
        response = yield from aiohttp.request('get', uri, headers=headers)
        content = yield from response.json()
        # print('received content from {}'.format(uri))
        return (uri, content)

    @asyncio.coroutine
    def get_all_events_from_page(self, page):
        coroutines = []
        for entry in page.entries():
            task = asyncio.Task(self.fetch_event_async(entry.links['alternate']))
            coroutines.append(task)

        return (yield from asyncio.gather(*coroutines))

    @asyncio.coroutine
    def get_all_events_async(self, stream_name, on_event):
        q = asyncio.Queue()
        head = yield from self.get_stream_head_async(stream_name)
        last = yield from self.get_stream_page_async(head.links['last'])

        @asyncio.coroutine
        def follow_previous_links():
            current_page = last
            while 'previous' in current_page.links:
                yield from q.put(current_page)
                previous_uri = current_page.links['previous']
                current_page = yield from self.get_stream_page_async(previous_uri)
            yield from q.put(None) # indicate last page

        @asyncio.coroutine
        def fetch_events():
            while True:
                page = yield from q.get()
                if page is None: # last page
                    return
                events = yield from self.get_all_events_from_page(page)
                for event in events:
                    on_event(event)

        asyncio.async(follow_previous_links())
        yield from fetch_events()

    def get_stream_page(self, uri):
        headers = {'Accept': 'application/vnd.eventstore.events+json'}
        response = requests.get(uri, headers=headers)
        response.raise_for_status()
        content = response.json()
        return StreamPage(content)

    def get_stream_head(self, stream_name):
        uri = self.base_url + '/streams/' + stream_name
        return self.get_stream_page(uri)

    def fetch_events(self, stream_entries):
        for entry in stream_entries:
            headers = {'Accept': 'application/json'}
            response = requests.get(entry.links['alternate'], headers=headers)
            content = response.json()
            yield entry.summary, content

    def get_all_events(self, stream_name):
        head = self.get_stream_head(stream_name)
        uri = head.links.get('last', None)
        while uri:
            current_page = self.get_stream_page(uri)
            yield from self.fetch_events(current_page.entries())
            uri = current_page.links.get('previous', None)

    def subscribe(self, stream_name, interval_seconds=1):
        last = None
        while last is None:<commit_msg>Remove old lame synchronous get_all_events<commit_after>import json
import uuid
import time
import asyncio

import requests
from requests.exceptions import HTTPError
import aiohttp

def links_as_dict(links):
    result = {}
    for link in links:
        result[link['relation']] = link['uri']
    return result

class StreamEntry:
    def __init__(self, content):
        self._content = content
        self._links = None
        self.summary = content['summary']

    @property
    def links(self):
        if self._links is None:
            self._links = links_as_dict(self._content['links'])
        return self._links

class StreamPage:
    def __init__(self, content):
        self._content = content 
        self._links = None

    @property
    def links(self):
        if self._links is None:
            self._links = links_as_dict(self._content['links'])
        return self._links

    def entries(self):
        for entry_content in reversed(self._content['entries']):
            yield StreamEntry(entry_content)

class EventStoreClient:

    def __init__(self, host, secure=False, port=2113):
        proto = "https" if secure else "http"
        self.base_url = proto + "://" + host + ":" + str(port)

    def post_events(self, stream_name, events):
        url = self.base_url +  "/streams/" + stream_name
        headers = {'Content-Type': 'application/vnd.eventstore.events+json'}
        response = requests.post(url, headers=headers, data=json.dumps(events))
        if response.status_code >= 400 and response.status_code < 500:
            raise ValueError(response.reason)

    def publish_event(self, stream_name, event_type, data, event_id=None):
        if event_id is None:
            event_id = str(uuid.uuid4())

        event = {
            'eventId': event_id,
            'eventType': event_type,
            'data': data
        }
        self.post_events(stream_name, [event])

    @asyncio.coroutine
    def get_stream_page_async(self, uri):
        # print('getting stream page {}'.format(uri))
        headers = {'Accept': 'application/vnd.eventstore.events+json'}
        response = yield from aiohttp.request('get', uri, headers=headers)
        content = yield from response.json()
        # print('received stream page {}'.format(uri))
        return StreamPage(content)

    @asyncio.coroutine
    def get_stream_head_async(self, stream_name):
        uri = '{}/streams/{}'.format(self.base_url, stream_name)
        return (yield from self.get_stream_page_async(uri))

    @asyncio.coroutine
    def fetch_event_async(self, uri):
        # print('getting event data from {}'.format(uri))
        headers = {'Accept': 'application/json'}
        response = yield from aiohttp.request('get', uri, headers=headers)
        content = yield from response.json()
        # print('received content from {}'.format(uri))
        return (uri, content)

    @asyncio.coroutine
    def get_all_events_from_page(self, page):
        coroutines = []
        for entry in page.entries():
            task = asyncio.Task(self.fetch_event_async(entry.links['alternate']))
            coroutines.append(task)

        return (yield from asyncio.gather(*coroutines))

    @asyncio.coroutine
    def get_all_events_async(self, stream_name, on_event):
        q = asyncio.Queue()
        head = yield from self.get_stream_head_async(stream_name)
        last = yield from self.get_stream_page_async(head.links['last'])

        @asyncio.coroutine
        def follow_previous_links():
            current_page = last
            while 'previous' in current_page.links:
                yield from q.put(current_page)
                previous_uri = current_page.links['previous']
                current_page = yield from self.get_stream_page_async(previous_uri)
            yield from q.put(None) # indicate last page

        @asyncio.coroutine
        def fetch_events():
            while True:
                page = yield from q.get()
                if page is None: # last page
                    return
                events = yield from self.get_all_events_from_page(page)
                for event in events:
                    on_event(event)

        asyncio.async(follow_previous_links())
        yield from fetch_events()

    def get_stream_page(self, uri):
        headers = {'Accept': 'application/vnd.eventstore.events+json'}
        response = requests.get(uri, headers=headers)
        response.raise_for_status()
        content = response.json()
        return StreamPage(content)

    def get_stream_head(self, stream_name):
        uri = self.base_url + '/streams/' + stream_name
        return self.get_stream_page(uri)

    def fetch_events(self, stream_entries):
        for entry in stream_entries:
            headers = {'Accept': 'application/json'}
            response = requests.get(entry.links['alternate'], headers=headers)
            content = response.json()
            yield entry.summary, content

    def subscribe(self, stream_name, interval_seconds=1):
        last = None
        while last is None:
{'fraction_non_alphanumeric': 0.10420055497081619, 'fraction_numerical': 0.0022964309635441584, 'mean_word_length': 2.484, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '323347', 'n_tokens_mistral': 2992, 'n_tokens_neox': 2914, 'n_words': 701}