<commit_before>import os
import argparse
from time import time
from time import sleep
from sqlalchemy import text


from app import db
from app import logger

from queue_main import DbQueue
from pub import Pub
from util import run_sql



class DbQueuePub(DbQueue):
    def table_name(self, job_type):
        table_name = "pub"
        return table_name

    def process_name(self, job_type):
        process_name = "update" # formation name is from Procfile
        return process_name

    def worker_run(self, **kwargs):
        single_obj_id = kwargs.get("id", None)
        chunk = kwargs.get("chunk", 25)
        limit = kwargs.get("limit", 10)
        queue_table = "pub"
        run_class = Pub
        run_method = "update"

        if single_obj_id:
            limit = 1
        else:
            if not limit:
                limit = 1000
            text_query_pattern = """WITH picked_from_queue AS (
                       SELECT *
                       FROM   {queue_table}
                       WHERE  started is null
                       ORDER BY updated asc
                       --ORDER BY rand
                   LIMIT  {chunk}
                   FOR UPDATE SKIP LOCKED
                   )
                UPDATE {queue_table} queue_rows_to_update
                SET    started=now()
                FROM   picked_from_queue
                WHERE picked_from_queue.id = queue_rows_to_update.id
                RETURNING picked_from_queue.id;"""
            text_query = text_query_pattern.format(
                limit=limit,
                chunk=chunk,
                queue_table=queue_table
            )
            logger.info(u"the queue query is:\n{}".format(text_query))

        index = 0
        start_time = time()
        while True:
            new_loop_start_time = time()
            if single_obj_id:
                objects = [run_class.query.filter(run_class.id == single_obj_id).first()]
            else:
                # logger.info(u"looking for new jobs")
                # objects = run_class.query.from_statement(text(text_query)).execution_options(autocommit=True).all()
                id_rows =  db.engine.execute(text(text_query)).fetchall()
                ids = [row[0] for row in id_rows]<commit_msg>add updated is null for now<commit_after>import os
import argparse
from time import time
from time import sleep
from sqlalchemy import text


from app import db
from app import logger

from queue_main import DbQueue
from pub import Pub
from util import run_sql



class DbQueuePub(DbQueue):
    def table_name(self, job_type):
        table_name = "pub"
        return table_name

    def process_name(self, job_type):
        process_name = "update" # formation name is from Procfile
        return process_name

    def worker_run(self, **kwargs):
        single_obj_id = kwargs.get("id", None)
        chunk = kwargs.get("chunk", 25)
        limit = kwargs.get("limit", 10)
        queue_table = "pub"
        run_class = Pub
        run_method = "update"

        if single_obj_id:
            limit = 1
        else:
            if not limit:
                limit = 1000
            text_query_pattern = """WITH picked_from_queue AS (
                       SELECT *
                       FROM   {queue_table}
                       WHERE  started is null
                       and updated is null
                       ORDER BY updated asc
                       --ORDER BY rand
                   LIMIT  {chunk}
                   FOR UPDATE SKIP LOCKED
                   )
                UPDATE {queue_table} queue_rows_to_update
                SET    started=now()
                FROM   picked_from_queue
                WHERE picked_from_queue.id = queue_rows_to_update.id
                RETURNING picked_from_queue.id;"""
            text_query = text_query_pattern.format(
                limit=limit,
                chunk=chunk,
                queue_table=queue_table
            )
            logger.info(u"the queue query is:\n{}".format(text_query))

        index = 0
        start_time = time()
        while True:
            new_loop_start_time = time()
            if single_obj_id:
                objects = [run_class.query.filter(run_class.id == single_obj_id).first()]
            else:
                # logger.info(u"looking for new jobs")
                # objects = run_class.query.from_statement(text(text_query)).execution_options(autocommit=True).all()
                id_rows =  db.engine.execute(text(text_query)).fetchall()
                ids = [row[0] for row in id_rows]
{'fraction_non_alphanumeric': 0.07247325911373062, 'fraction_numerical': 0.004802444881030343, 'mean_word_length': 1.5989790130459445, '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': '5672317', 'n_tokens_mistral': 1294, 'n_tokens_neox': 1202, 'n_words': 324}