<commit_before># -*- coding: utf-8 -*-
#
# Copyright 2016 dpa-infocom GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import copy
import logging
from livebridge.components import get_converter, get_post, get_source, get_db_client


logger = logging.getLogger(__name__)


class LiveBridge(object):

    def __init__(self, config):
        self.targets = []
        self.api_client = None
        self.config = config
        self.source_id = self.config.get("source_id")
        self.endpoint = self.config.get("endpoint")
        self.label = self.config.get("label", "-")
        self.db = get_db_client()

    def __repr__(self):
        return "<LiveBridge [{}] {} {}>".format(self.label, self.endpoint, self.source_id)

    @property
    def source(self):
        if self.api_client:
            return self.api_client
        # setup api client
        self.api_client = get_source(self.config)
        return self.api_client

    def add_target(self, target):
        logger.debug("Adding target {} to {}".format(target, self))
        self.targets.append(target)

    async def listen_ws(self):
        return asyncio.Task(self.source.listen(self.new_posts))

    async def check_posts(self):
        try:
            # set last_updated
            if not self.source.last_updated:
                self.source.last_updated = await self.db.get_last_updated(self.source_id)
                logger.info("LAST UPDATED: {} {}".format(self.source.last_updated, self.source))
            posts = await self.source.poll()
            if posts:
                await self.new_posts(posts)
        except Exception as e:
            logger.error("Fatal checking {}{}".format(getattr(self, "endpoint", "-"), self.source_id))
            logger.exception(e)
        return True

    async def _process_post(self, target, source_post):
        post = copy.deepcopy(source_post)
        # update last updated property in source client
        self.source.last_updated = post.updated
        # handle post in target
        await target.handle_post(post)

    async def new_posts(self, posts):
        try:
            logger.info("##### Received {} posts from {}".format(len(posts), self.source))
            for p in posts:
                coros = [self._process_post(t, p) for t in self.targets]
                await asyncio.gather(*coros, return_exceptions=True)
        except Exception as e:
            logger.error("Handling new posts failed [{}]: {}".format(self.source, e))
        return True<commit_msg>Improve error handling for new_posts<commit_after># -*- coding: utf-8 -*-
#
# Copyright 2016 dpa-infocom GmbH
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import copy
import logging
from livebridge.components import get_converter, get_post, get_source, get_db_client


logger = logging.getLogger(__name__)


class LiveBridge(object):

    def __init__(self, config):
        self.targets = []
        self.api_client = None
        self.config = config
        self.source_id = self.config.get("source_id")
        self.endpoint = self.config.get("endpoint")
        self.label = self.config.get("label", "-")
        self.db = get_db_client()

    def __repr__(self):
        return "<LiveBridge [{}] {} {}>".format(self.label, self.endpoint, self.source_id)

    @property
    def source(self):
        if self.api_client:
            return self.api_client
        # setup api client
        self.api_client = get_source(self.config)
        return self.api_client

    def add_target(self, target):
        logger.debug("Adding target {} to {}".format(target, self))
        self.targets.append(target)

    async def listen_ws(self):
        return asyncio.Task(self.source.listen(self.new_posts))

    async def check_posts(self):
        try:
            # set last_updated
            if not self.source.last_updated:
                self.source.last_updated = await self.db.get_last_updated(self.source_id)
                logger.info("LAST UPDATED: {} {}".format(self.source.last_updated, self.source))
            posts = await self.source.poll()
            if posts:
                await self.new_posts(posts)
        except Exception as e:
            logger.error("Fatal checking {}{}".format(getattr(self, "endpoint", "-"), self.source_id))
            logger.exception(e)
        return True

    async def _process_post(self, target, source_post):
        post = copy.deepcopy(source_post)
        # update last updated property in source client
        self.source.last_updated = post.updated
        # handle post in target
        await target.handle_post(post)

    async def log_error_results(self, results):
        for err in [r for r in results if r]:
            try:
                raise err
            except Exception as e:
                logger.error("Handling new posts failed [{}]: {}".format(self.source, e))

    async def new_posts(self, posts):
        logger.info("##### Received {} posts from {}".format(len(posts), self.source))
        for p in posts:
            coros = [self._process_post(t, p) for t in self.targets]
            res = await asyncio.gather(*coros, return_exceptions=True)
            await self.log_error_results(res)
        return True
{'fraction_non_alphanumeric': 0.10461390579942326, 'fraction_numerical': 0.0028836911246395386, 'mean_word_length': 2.610757663389242, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4361051', 'n_tokens_mistral': 1773, 'n_tokens_neox': 1718, 'n_words': 516}