<commit_before>import logging
import sys
import os
from c2corg_api.scripts.es.sync import sync_es
from c2corg_api.search import configure_es_from_config, get_queue_config
from pyramid.scripts.common import parse_vars
from pyramid.paster import get_appsettings, setup_logging
from sqlalchemy import engine_from_config

from kombu.mixins import ConsumerMixin
from sqlalchemy.orm import sessionmaker

log = logging.getLogger('c2corg_api_syncer')


class SyncWorker(ConsumerMixin):
    """Worker that listens to messages in the Redis queue and then synchronizes
    changes that have been made in the Postgres database with ElasticSearch.

    Based on this example:
    http://docs.celeryproject.org/projects/kombu/en/latest/userguide/examples.html
    """

    def __init__(
            self, connection, queue, batch_size, session=None,
            session_factory=None):
        self.connection = connection
        self.queue = queue
        self.batch_size = batch_size
        self.session = session
        self.session_factory = session_factory

    def get_consumers(self, consumer_factory, channel):
        return [consumer_factory(
            queues=[self.queue], callbacks=[self.process_task])]

    def process_task(self, body, message):
        log.info('Sync requested')
        # the sync request is confirmed even though it might fail. in this case
        # a new request will trigger a new sync, or manual interaction is
        # required.
        message.ack()
        try:
            self.sync()
        except:
            log.error('Sync failed', exc_info=True)
        log.info('Waiting on messages')

    def sync(self):<commit_msg>Try to reconnect in syncer<commit_after>import logging
import sys
import os
from c2corg_api.scripts.es.sync import sync_es
from c2corg_api.search import configure_es_from_config, get_queue_config
from pyramid.scripts.common import parse_vars
from pyramid.paster import get_appsettings, setup_logging
from sqlalchemy import engine_from_config

from kombu.mixins import ConsumerMixin
from sqlalchemy.orm import sessionmaker

log = logging.getLogger('c2corg_api_syncer')


class SyncWorker(ConsumerMixin):
    """Worker that listens to messages in the Redis queue and then synchronizes
    changes that have been made in the Postgres database with ElasticSearch.

    Based on this example:
    http://docs.celeryproject.org/projects/kombu/en/latest/userguide/examples.html
    """

    def __init__(
            self, connection, queue, batch_size, session=None,
            session_factory=None):
        self.connection = connection
        self.queue = queue
        self.batch_size = batch_size
        self.session = session
        self.session_factory = session_factory
        self.connect_max_retries = 3

    def get_consumers(self, consumer_factory, channel):
        return [consumer_factory(
            queues=[self.queue], callbacks=[self.process_task])]

    def process_task(self, body, message):
        log.info('Sync requested')
        # the sync request is confirmed even though it might fail. in this case
        # a new request will trigger a new sync, or manual interaction is
        # required.
        message.ack()
        try:
            self.sync()
        except:
            log.error('Sync failed', exc_info=True)
        log.info('Waiting on messages')

    def sync(self):
{'fraction_non_alphanumeric': 0.0718954248366013, 'fraction_numerical': 0.0020796197266785502, 'mean_word_length': 3.121175030599755, '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': '644163', 'n_tokens_mistral': 952, 'n_tokens_neox': 922, 'n_words': 281}