<commit_before>
import sys, os, copy, time, hashlib

import random

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import eagerload
import psycopg2, psycopg2.extensions

from api0 import db_from_engine, postgres_db

JOBID = 'jobman.id'
EXPERIMENT = 'jobman.experiment'
#using the dictionary to store these is too slow
STATUS = 'jobman.status'
HASH = 'jobman.hash'
PRIORITY = 'jobman.sql.priority'

HOST = 'jobman.sql.hostname'
HOST_WORKDIR = 'jobman.sql.host_workdir'
PUSH_ERROR = 'jobman.sql.push_error'

START = 0
RUNNING = 1
DONE = 2
ERR_START = 3
ERR_SYNC = 4
ERR_RUN = 5
CANCELED = -1
FUCKED_UP = 666

_TEST_CONCURRENCY = False

def postgres_serial(user, password, host, port, database, poolclass=sqlalchemy.pool.NullPool, **kwargs):
    """Return a DbHandle instance that communicates with a postgres database at transaction
    isolation_level 'SERIALIZABLE'.

    :param user: a username in the database
    :param password: <PASSWORD> the username
    :param host: the network address of the host on which the postgres server is running
    :param port: the port on which to connect to host
    :param database: a database served by the postgres server

    """
    this = postgres_serial

    if not hasattr(this,'engine'):
        def connect():
            c = psycopg2.connect(user=user, password=password, database=database, host=host, port=port)
            c.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE)
            return c
        this.engine = create_engine('postgres://'
                ,creator=connect
                ,poolclass=poolclass
                )

    db = db_from_engine(this.engine, **kwargs)
    db._is_serialized_session_db = True
    return db

def book_dct_postgres_serial(db, retry_max_sleep=10.0, verbose=1):
    """Find a trial in the lisa_db with status START.

    A trial will be returned with status=RUNNING.

    Returns None if no such trial exists in DB.

    This function uses a serial access to the lisadb to guarantee that no other
    process will retrieve the same dct.  It is designed to facilitate writing
    a "consumer" for a Producer-Consumer pattern based on the database.

    """
    print >> sys.stderr, """#TODO: use the priority field, not the status."""
    print >> sys.stderr, """#TODO: ignore entries with key PUSH_ERROR."""

    s = db.session() #open a new session

    # NB. we need the query and attribute update to be in the same transaction
    try:
        assert s.autocommit == False
    except AttributeError:
        pass

    dcts_seen = set([])
    keep_trying = True

    dct = None
    while (dct is None) and keep_trying:
        #build a query
        q = s.query(db._Dict)

        #N.B.
        # use dedicated column to retrieve jobs, not the dictionary keyval pair
        # This should be much faster.
        q = q.filter(db._Dict.status==START)
        q = q.order_by(db._Dict.priority.desc())

        # this doesn't seem to work, hene the string hack below
        q = q.options(eagerload('_attrs')) #hard-coded in api0

        #try to reserve a dct
        try:
            #first() may raise psycopg2.ProgrammingError
            dct = q.first()

            if dct is not None:
                assert (dct not in dcts_seen)
                if verbose: print 'book_unstarted_dct retrieved, ', dct
                dct._set_in_session(STATUS, RUNNING, s)
                if 1:
                    if _TEST_CONCURRENCY:
                        print >> sys.stderr, 'SLEEPING BEFORE BOOKING'
                        time.sleep(10)

                    #commit() may raise psycopg2.ProgrammingError
                    s.commit()
                else:
                    print >> sys.stderr, 'DEBUG MODE: NOT RESERVING JOB!', dct
                #if we get this far, the job is ours!
            else:
                # no jobs are left
                keep_trying = False
        except (psycopg2.OperationalError,
                sqlalchemy.exceptions.ProgrammingError,
                sqlalchemy.exc.DBAPIError), e:
            #either the first() or the commit() raised
            s.rollback() # docs say to do this (or close) after commit raises exception
            if verbose: print 'caught exception', e
            if dct:
                # first() succeeded, commit() failed
                dcts_seen.add(dct)
                dct = None
            wait = random.random(1)*retry_max_sleep
            if verbose: print 'another process stole our dct. Waiting %f secs' % wait
            print 'waiting for %i second' % wait
            time.sleep(wait)

    if dct:
        str(dct) # for loading of attrs in UGLY WAY!!!
    s.close()
    return dct

def book_dct_non_postgres(db):
    print >> sys.stderr, """#TODO: use the priority field, not the status."""
    print >> sys.stderr, """#TODO: ignore entries with key self.push_error."""

    raise NotImplementedError()


###########
# Connect
###########

def parse_dbstring(dbstring):
    """Unpacks a dbstring of the form postgres://username[:password]@hostname[:port]/dbname/tablename

    :rtype: tuple of strings
    :returns: username, password, hostname, port, dbname, tablename

    :note: If the password is not given in the dbstring, this function attempts to retrieve it using
    >>> password = get_password(hostname, dbname)

    :note: port defaults to 5432 (postgres default).<commit_msg>Fixed call to random.random(), no argument should be given.<commit_after>
import sys, os, copy, time, hashlib

import random

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import eagerload
import psycopg2, psycopg2.extensions

from api0 import db_from_engine, postgres_db

JOBID = 'jobman.id'
EXPERIMENT = 'jobman.experiment'
#using the dictionary to store these is too slow
STATUS = 'jobman.status'
HASH = 'jobman.hash'
PRIORITY = 'jobman.sql.priority'

HOST = 'jobman.sql.hostname'
HOST_WORKDIR = 'jobman.sql.host_workdir'
PUSH_ERROR = 'jobman.sql.push_error'

START = 0
RUNNING = 1
DONE = 2
ERR_START = 3
ERR_SYNC = 4
ERR_RUN = 5
CANCELED = -1
FUCKED_UP = 666

_TEST_CONCURRENCY = False

def postgres_serial(user, password, host, port, database, poolclass=sqlalchemy.pool.NullPool, **kwargs):
    """Return a DbHandle instance that communicates with a postgres database at transaction
    isolation_level 'SERIALIZABLE'.

    :param user: a username in the database
    :param password: the password for the username
    :param host: the network address of the host on which the postgres server is running
    :param port: the port on which to connect to host
    :param database: a database served by the postgres server

    """
    this = postgres_serial

    if not hasattr(this,'engine'):
        def connect():
            c = psycopg2.connect(user=user, password=password, database=database, host=host, port=port)
            c.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_SERIALIZABLE)
            return c
        this.engine = create_engine('postgres://'
                ,creator=connect
                ,poolclass=poolclass
                )

    db = db_from_engine(this.engine, **kwargs)
    db._is_serialized_session_db = True
    return db

def book_dct_postgres_serial(db, retry_max_sleep=10.0, verbose=1):
    """Find a trial in the lisa_db with status START.

    A trial will be returned with status=RUNNING.

    Returns None if no such trial exists in DB.

    This function uses a serial access to the lisadb to guarantee that no other
    process will retrieve the same dct.  It is designed to facilitate writing
    a "consumer" for a Producer-Consumer pattern based on the database.

    """
    print >> sys.stderr, """#TODO: use the priority field, not the status."""
    print >> sys.stderr, """#TODO: ignore entries with key PUSH_ERROR."""

    s = db.session() #open a new session

    # NB. we need the query and attribute update to be in the same transaction
    try:
        assert s.autocommit == False
    except AttributeError:
        pass

    dcts_seen = set([])
    keep_trying = True

    dct = None
    while (dct is None) and keep_trying:
        #build a query
        q = s.query(db._Dict)

        #N.B.
        # use dedicated column to retrieve jobs, not the dictionary keyval pair
        # This should be much faster.
        q = q.filter(db._Dict.status==START)
        q = q.order_by(db._Dict.priority.desc())

        # this doesn't seem to work, hene the string hack below
        q = q.options(eagerload('_attrs')) #hard-coded in api0

        #try to reserve a dct
        try:
            #first() may raise psycopg2.ProgrammingError
            dct = q.first()

            if dct is not None:
                assert (dct not in dcts_seen)
                if verbose: print 'book_unstarted_dct retrieved, ', dct
                dct._set_in_session(STATUS, RUNNING, s)
                if 1:
                    if _TEST_CONCURRENCY:
                        print >> sys.stderr, 'SLEEPING BEFORE BOOKING'
                        time.sleep(10)

                    #commit() may raise psycopg2.ProgrammingError
                    s.commit()
                else:
                    print >> sys.stderr, 'DEBUG MODE: NOT RESERVING JOB!', dct
                #if we get this far, the job is ours!
            else:
                # no jobs are left
                keep_trying = False
        except (psycopg2.OperationalError,
                sqlalchemy.exceptions.ProgrammingError,
                sqlalchemy.exc.DBAPIError), e:
            #either the first() or the commit() raised
            s.rollback() # docs say to do this (or close) after commit raises exception
            if verbose: print 'caught exception', e
            if dct:
                # first() succeeded, commit() failed
                dcts_seen.add(dct)
                dct = None
            wait = random.random()*retry_max_sleep
            if verbose: print 'another process stole our dct. Waiting %f secs' % wait
            print 'waiting for %i second' % wait
            time.sleep(wait)

    if dct:
        str(dct) # for loading of attrs in UGLY WAY!!!
    s.close()
    return dct

def book_dct_non_postgres(db):
    print >> sys.stderr, """#TODO: use the priority field, not the status."""
    print >> sys.stderr, """#TODO: ignore entries with key self.push_error."""

    raise NotImplementedError()


###########
# Connect
###########

def parse_dbstring(dbstring):
    """Unpacks a dbstring of the form postgres://username[:password]@hostname[:port]/dbname/tablename

    :rtype: tuple of strings
    :returns: username, password, hostname, port, dbname, tablename

    :note: If the password is not given in the dbstring, this function attempts to retrieve it using
    >>> password = get_password(hostname, dbname)

    :note: port defaults to 5432 (postgres default).
{'fraction_non_alphanumeric': 0.08811908481117339, 'fraction_numerical': 0.005605072130846274, 'mean_word_length': 2.410842995926042, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 34, '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': '173811', 'n_tokens_mistral': 3310, 'n_tokens_neox': 3159, 'n_words': 1093}