<commit_before>import datetime
import re
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import yaml
from exporters.persistence.base_persistence import BasePersistence

Base = declarative_base()


class Job(Base):
    __tablename__ = 'job'
    id = Column(Integer, primary_key=True)
    last_position = Column(String(250), nullable=False)
    type = Column(String(250), nullable=False)
    last_committed = Column(DateTime)
    job_finished = Column(Boolean)
    configuration = Column(String(50000), nullable=False)


class BaseAlchemyPersistence(BasePersistence):
    supported_options = {
        'user': {'type': basestring},
        'password': {'type': basestring},
        'host': {'type': basestring},
        'port': {'type': int},
        'database': {'type': basestring}
    }

    def __init__(self, options):
        self.engine = None
        super(BaseAlchemyPersistence, self).__init__(options)

    def _db_init(self):
        user = self.read_option('user')
        password = self.read_option('password')
        host = self.read_option('host')
        port = self.read_option('port')
        database = self.read_option('database')
        self.engine = create_engine(
            '{}://{}:{}@{}:{}/{}'.format(self.PROTOCOL, user, password, host, port, database))
        Base.metadata.create_all(self.engine)
        Base.metadata.bind = self.engine
        DBSession = sessionmaker(bind=self.engine)
        self.session = DBSession()

    def get_last_position(self):
        if not self.engine:
            self._db_init()
        job = self.session.query(Job).filter(Job.id == self.job_id).first()
        last_position = 0
        try:
            exec('last_position = {}({})'.format(job.type, job.last_position))
        except:
            pass
        return last_position

    def commit_position(self, last_position=None):
        self.last_position = last_position
        self.session.query(Job).filter(Job.id == self.job_id).update(
            {"last_position": self.last_position, "type": type(self.last_position).__name__,
             "last_committed": datetime.datetime.now()}, synchronize_session='fetch')
        self.session.commit()
        self.logger.debug('Commited batch number ' + str(self.last_position) + ' of job: ' + str(self.job_id))

    def generate_new_job(self):
        if not self.engine:
            self._db_init()
        new_job = Job(last_position='None', type='None', configuration=str(self.configuration))
        self.session.add(new_job)
        self.session.commit()
        self.job_id = new_job.id
        self.logger.debug('Created persistence job with id {} in database {}. Using protocol {}.'
                          .format(new_job.id, self.read_option('database'), self.PROTOCOL) + str(new_job.id))
        return new_job.id

    def delete_instance(self):
        self.session.query(Job).filter(Job.id == self.job_id).update(
            {"job_finished": True, "last_committed": datetime.datetime.now()}, synchronize_session='fetch')
        self.session.commit()
        self.session.close()


    @staticmethod
    def configuration_from_uri(uri, uri_regex):
        """
        returns a configuration object.
        """
        connection_parameters = re.match(uri_regex, uri).groups()
        user, password, host, port, database, job_id = connection_parameters
        engine = create_engine(
            '{}://{}:{}@{}:{}/{}'.format(uri.split('://')[0], user, password, host, port, database))
        Base.metadata.create_all(engine)
        Base.metadata.bind = engine
        DBSession = sessionmaker(bind=engine)
        session = DBSession()
        job = session.query(Job).filter(Job.id == int(job_id)).first()
        configuration = job.configuration
        configuration = yaml.safe_load(configuration)
        configuration['exporter_options']['RESUME'] = True
        configuration['exporter_options']['JOB_ID'] = job_id
        return configuration
<commit_msg>Removes type and cast needs from alchemy persistence.<commit_after>import datetime
import json
import re
from sqlalchemy import create_engine, Column, Integer, String, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import yaml
from exporters.persistence.base_persistence import BasePersistence

Base = declarative_base()


class Job(Base):
    __tablename__ = 'job'
    id = Column(Integer, primary_key=True)
    last_position = Column(String(1500), nullable=False)
    last_committed = Column(DateTime)
    job_finished = Column(Boolean)
    configuration = Column(String(50000), nullable=False)


class BaseAlchemyPersistence(BasePersistence):
    supported_options = {
        'user': {'type': basestring},
        'password': {'type': basestring},
        'host': {'type': basestring},
        'port': {'type': int},
        'database': {'type': basestring}
    }

    def __init__(self, options):
        self.engine = None
        super(BaseAlchemyPersistence, self).__init__(options)

    def _db_init(self):
        user = self.read_option('user')
        password = self.read_option('password')
        host = self.read_option('host')
        port = self.read_option('port')
        database = self.read_option('database')
        self.engine = create_engine(
            '{}://{}:{}@{}:{}/{}'.format(self.PROTOCOL, user, password, host, port, database))
        Base.metadata.create_all(self.engine)
        Base.metadata.bind = self.engine
        DBSession = sessionmaker(bind=self.engine)
        self.session = DBSession()

    def get_last_position(self):
        if not self.engine:
            self._db_init()
        job = self.session.query(Job).filter(Job.id == self.job_id).first()
        last_position = json.loads(job.last_position)
        return last_position

    def commit_position(self, last_position=None):
        self.last_position = last_position
        self.session.query(Job).filter(Job.id == self.job_id).update(
            {"last_position": json.dumps(self.last_position),
             "last_committed": datetime.datetime.now()}, synchronize_session='fetch')
        self.session.commit()
        self.logger.debug('Commited batch number ' + str(self.last_position) + ' of job: ' + str(self.job_id))

    def generate_new_job(self):
        if not self.engine:
            self._db_init()
        new_job = Job(last_position='None', configuration=str(self.configuration))
        self.session.add(new_job)
        self.session.commit()
        self.job_id = new_job.id
        self.logger.debug('Created persistence job with id {} in database {}. Using protocol {}.'
                          .format(new_job.id, self.read_option('database'), self.PROTOCOL) + str(new_job.id))
        return new_job.id

    def delete_instance(self):
        self.session.query(Job).filter(Job.id == self.job_id).update(
            {"job_finished": True, "last_committed": datetime.datetime.now()}, synchronize_session='fetch')
        self.session.commit()
        self.session.close()


    @staticmethod
    def configuration_from_uri(uri, uri_regex):
        """
        returns a configuration object.
        """
        connection_parameters = re.match(uri_regex, uri).groups()
        user, password, host, port, database, job_id = connection_parameters
        engine = create_engine(
            '{}://{}:{}@{}:{}/{}'.format(uri.split('://')[0], user, password, host, port, database))
        Base.metadata.create_all(engine)
        Base.metadata.bind = engine
        DBSession = sessionmaker(bind=engine)
        session = DBSession()
        job = session.query(Job).filter(Job.id == int(job_id)).first()
        configuration = job.configuration
        configuration = yaml.safe_load(configuration)
        configuration['exporter_options']['resume'] = True
        configuration['exporter_options']['job_id'] = job_id
        return configuration

{'fraction_non_alphanumeric': 0.13073194979495464, 'fraction_numerical': 0.0028582080278364607, 'mean_word_length': 3.0099651220727455, 'pattern_counts': {'":': 9, '<': 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': '4258989', 'n_tokens_mistral': 2342, 'n_tokens_neox': 2307, 'n_words': 453}