<commit_before>from celery.signals import worker_process_init
from celery import Task, task
import logging

from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import create_engine
from meerkat_abacus.pipeline_worker.pipeline import Pipeline

from meerkat_abacus import util
from meerkat_abacus import config

config_ = config.get_config()


pipeline = None


def configure_worker():
    # load the application configuration
    # db_uri = conf['db_uri']
    global engine
    logging.info("Worker setup")
    engine = create_engine(config_.DATABASE_URL, pool_pre_ping=True)

    global session
    session = scoped_session(sessionmaker(autocommit=False,
                                          autoflush=False,
                                          bind=engine))
    logging.info(session)
    global pipeline
    pipeline = Pipeline(engine, session, config_)


@task(bind=True, name="process_data")
def process_data(self, data_rows):
    if pipeline is None:
        configure_worker()
    logging.info("STARTING task")
    engine.dispose()
    pipeline.process_chunk(data_rows)
    logging.info("ENDING task")


@task
def test_up():
    return True<commit_msg>Fixing registering celery tasks in pipeline worker.<commit_after>import logging

from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import create_engine
from meerkat_abacus.pipeline_worker.pipeline import Pipeline

from meerkat_abacus import config
from meerkat_abacus.pipeline_worker.celery_app import app

config_ = config.get_config()


pipeline = None


def configure_worker():
    # load the application configuration
    # db_uri = conf['db_uri']
    global engine
    logging.info("Worker setup")
    engine = create_engine(config_.DATABASE_URL, pool_pre_ping=True)

    global session
    session = scoped_session(sessionmaker(autocommit=False,
                                          autoflush=False,
                                          bind=engine))
    logging.info(session)
    global pipeline
    pipeline = Pipeline(engine, session, config_)


@app.task(bind=True, name="process_data")
def process_data(self, data_rows):
    if pipeline is None:
        configure_worker()
    logging.info("STARTING task")
    engine.dispose()
    pipeline.process_chunk(data_rows)
    logging.info("ENDING task")


@app.task
def test_up():
    return True
{'fraction_non_alphanumeric': 0.07549557148882328, 'fraction_numerical': 0.0, 'mean_word_length': 3.3443223443223444, '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': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4817489', 'n_tokens_mistral': 676, 'n_tokens_neox': 666, 'n_words': 166}