<commit_before>import asyncio
import datetime
import logging
from abc import abstractmethod

from ..core.base import AbstractEntity
from ..utils import import_name


class AbstractWorker(AbstractEntity):
    @abstractmethod  # pragma: no cover
    async def start(self):
        raise NotImplementedError()

    @abstractmethod  # pragma: no cover
    async def stop(self):
        raise NotImplementedError()

    @abstractmethod  # pragma: no cover
    async def status(self):
        raise NotImplementedError()


class Worker(AbstractWorker):
    """ Worker
        config:
            run: str optional coroutine
            autorun: bool with self.init
            persist: bool if need rerun
            logger: str
            sleep: int time in seconds for sleep between rerun
            sleep_start: int time in seconds for sleep before run
            input: str.path to instance of AbstractReader
            output: str.path to instance of AbstractWriter
    """
    async def init(self):
        self._started_at = None
        self._stoped_at = None
        self._future = None
        if self.config.get('run'):
            run = import_name(self.config.run)
            self.run = lambda value=None: run(self, value)
        self._input = self.context[self.config.get('input')]
        self._output = self.context[self.config.get('output')]
        if self._input is not None or self._output is not None:
            self._persist = True
        else:
            self._persist = self.config.get('persist')
        self._sleep = self.config.get('sleep')
        self._sleep_start = self.config.get('sleep_start')
        self.logger = logging.getLogger(self.config.get('logger', __name__))
        if self.config.get('autorun'):
            await self.start()

    @property
    def input(self):
        return self._input

    @property
    def output(self):
        return self._output

    async def runner(self):
        try:
            if self._sleep_start:
                await asyncio.sleep(self._sleep_start, loop=self.loop)
            while True:
                try:
                    if self._input is not None:
                        args = (await self._input.get(),)
                    else:
                        args = ()
                    result = await self.run(*args)
                    if self._output is not None:
                        await self._output.put(result)
                except BaseException:
                    self.logger.exception('ERROR')
                if not self._persist:
                    return
                if self._sleep:
                    await asyncio.sleep(self._sleep, loop=self.loop)
        finally:
            self._stoped_at = datetime.datetime.now()

    async def run(self, value=None):
        raise NotImplementedError()

    def running(self):<commit_msg>reraise CancelledError in worker<commit_after>import asyncio
import datetime
import logging
from abc import abstractmethod

from ..core.base import AbstractEntity
from ..utils import import_name


class AbstractWorker(AbstractEntity):
    @abstractmethod  # pragma: no cover
    async def start(self):
        raise NotImplementedError()

    @abstractmethod  # pragma: no cover
    async def stop(self):
        raise NotImplementedError()

    @abstractmethod  # pragma: no cover
    async def status(self):
        raise NotImplementedError()


class Worker(AbstractWorker):
    """ Worker
        config:
            run: str optional coroutine
            autorun: bool with self.init
            persist: bool if need rerun
            logger: str
            sleep: int time in seconds for sleep between rerun
            sleep_start: int time in seconds for sleep before run
            input: str.path to instance of AbstractReader
            output: str.path to instance of AbstractWriter
    """
    async def init(self):
        self._started_at = None
        self._stoped_at = None
        self._future = None
        if self.config.get('run'):
            run = import_name(self.config.run)
            self.run = lambda value=None: run(self, value)
        self._input = self.context[self.config.get('input')]
        self._output = self.context[self.config.get('output')]
        if self._input is not None or self._output is not None:
            self._persist = True
        else:
            self._persist = self.config.get('persist')
        self._sleep = self.config.get('sleep')
        self._sleep_start = self.config.get('sleep_start')
        self.logger = logging.getLogger(self.config.get('logger', __name__))
        if self.config.get('autorun'):
            await self.start()

    @property
    def input(self):
        return self._input

    @property
    def output(self):
        return self._output

    async def runner(self):
        try:
            if self._sleep_start:
                await asyncio.sleep(self._sleep_start, loop=self.loop)
            while True:
                try:
                    if self._input is not None:
                        args = (await self._input.get(),)
                    else:
                        args = ()
                    result = await self.run(*args)
                    if self._output is not None:
                        await self._output.put(result)
                except asyncio.CancelledError:
                    raise
                except BaseException:
                    self.logger.exception('ERROR')
                if not self._persist:
                    return
                if self._sleep:
                    await asyncio.sleep(self._sleep, loop=self.loop)
        finally:
            self._stoped_at = datetime.datetime.now()

    async def run(self, value=None):
        raise NotImplementedError()

    def running(self):
{'fraction_non_alphanumeric': 0.08413793103448276, 'fraction_numerical': 0.0, 'mean_word_length': 1.7650142993326978, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2817793', 'n_tokens_mistral': 1516, 'n_tokens_neox': 1479, 'n_words': 423}