<commit_before>import logging

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Index, UniqueConstraint
from sqlalchemy.dialects.postgresql import (ARRAY, JSON)
from sqlalchemy import (Column, Integer, Text, ForeignKey, DateTime, Boolean,
                        func, CheckConstraint, and_)
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.sql.expression import text
from sqlalchemy.orm import relationship, backref, validates
from sqlalchemy.exc import IntegrityError
from croniter import croniter
import utc

import mettle_protocol as mp


Base = declarative_base()

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class Service(Base):
    __tablename__ = 'services'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    description = Column(Text)
    updated_by = Column(Text, nullable=False)

    @validates('name')
    def validate_name(self, key, name):
        # Ensure that the name has no characters that have special meanings in
        # RabbitMQ routing keys.
        assert '.' not in name
        assert '*' not in name
        assert '#' not in name
        return name


    __table_args__ = (
        UniqueConstraint('name'),
    )

    def __repr__(self):
        return 'Service <%s>' % self.name


class NotificationList(Base):
    __tablename__ = 'notification_lists'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    recipients = Column(ARRAY(Text), nullable=False)
    updated_by = Column(Text, nullable=False)
    def __repr__(self):
        return 'NotificationList <%s>' % self.name


class Pipeline(Base):
    __tablename__ = 'pipelines'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    service_id = Column(Integer, ForeignKey('services.id'), nullable=False)
    notification_list_id = Column(Integer, ForeignKey('notification_lists.id'),
                                  nullable=False)
    updated_by = Column(Text, nullable=False)
    active = Column(Boolean, nullable=False, server_default=text('true'))

    retries = Column(Integer, default=3)

    service = relationship("Service", backref=backref('pipelines', order_by=name))
    notification_list = relationship('NotificationList',
                                     backref=backref('pipelines', order_by=name))

    # A pipeline must have either a crontab or a trigger pipeline, but not both.
    crontab = Column(Text)
    chained_from_id = Column(Integer, ForeignKey('pipelines.id'))
    chained_from = relationship(lambda: Pipeline, remote_side=id,
                                backref='chains_to')

    @validates('name')
    def validate_name(self, key, name):
        # Ensure that the name has no characters that have special meanings in
        # RabbitMQ routing keys.
        assert '.' not in name
        assert '*' not in name
        assert '#' not in name
        return name

    @validates('crontab')
    def validate_crontab(self, key, cronspec):
        if cronspec is not None:
            # If the cronspec is not parseable, croniter will raise an exception
            # here.
            croniter(cronspec, utc.now())
        return cronspec

    __table_args__ = (
        UniqueConstraint('name', 'service_id'),
        CheckConstraint('crontab IS NOT NULL OR chained_from_id IS NOT NULL',
                        name='crontab_or_pipeline_check'),
        CheckConstraint('NOT (crontab IS NOT NULL AND chained_from_id IS NOT NULL)',
                        name='crontab_and_pipeline_check'),
    )

    def __repr__(self):
        return 'Pipeline <%s>' % self.name


class PipelineRun(Base):
    __tablename__ = 'pipeline_runs'
    id = Column(Integer, primary_key=True)
    pipeline_id = Column(Integer, ForeignKey('pipelines.id'), nullable=False)
    target_time = Column(DateTime(timezone=True), nullable=False)
    created_time = Column(DateTime(timezone=True), nullable=False,
                  server_default=func.now())
    succeeded = Column(Boolean, nullable=False, default=False,
                       server_default=text('false'))
    started_by = Column(Text, nullable=False) # username or 'timer'

    # These fields are set when we get an ack from the ETL service
    ack_time = Column(DateTime(timezone=True))
    targets = Column(MutableDict.as_mutable(JSON), default={})

    # end_time is set by dispatcher after it has heard in from all job runs
    end_time = Column(DateTime(timezone=True))

    pipeline = relationship("Pipeline", backref=backref('pipeline_runs',
                                                        order_by=created_time))

    def is_ended(self, db):
        if self.ack_time is None:
            return False
        return all(self.target_is_ended(db, t) for t in self.targets)

    def all_targets_succeeded(self, db):
        return all(self.target_is_succeeded(db, t) for t in self.targets)

    def target_is_succeeded(self, db, target):
        job = db.query(Job).filter(Job.pipeline_run==self,
                                   Job.target==target,
                                   Job.succeeded==True).first()
        return job is not None

    def target_is_failed(self, db, target):
        failure_count = db.query(Job).filter(Job.pipeline_run==self,
                                             Job.target==target,
                                             Job.end_time!=None,
                                             Job.succeeded==False).count()
        return failure_count >= self.pipeline.retries

    def target_is_ended(self, db, target):
        return (self.target_is_succeeded(db, target) or
                self.target_is_failed(db, target))

    def target_is_in_progress(self, db, target):
        job = db.query(Job).filter(Job.pipeline_run==self,
                                   Job.target==target,
                                   Job.end_time==None).first()
        return job is not None

    def target_deps_met(self, db, target):
        for dep in self.targets[target]:
            if not self.target_is_succeeded(db, dep):
                return False
        return True

    def target_is_ready(self, db, target):
        # Return true if the target meets these conditions:
        # 1. Are not ended.
        # 2. Do not already have an in-progress job for them in the DB.
        # 3. If they have dependencies, there is a successful job in the DB
        # whose target provides that dependency.
        if self.target_is_ended(db, target):
            return False
        if self.target_is_in_progress(db, target):
            return False
        if not self.target_deps_met(db, target):
            return False
        return True

    def get_ready_targets(self, db):
        return [t for t in self.targets if self.target_is_ready(db, t)]

    def make_job(self, db, target):
        job = Job(
            pipeline_run=self,
            target=target,<commit_msg>better comment on target_is_ready<commit_after>import logging

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Index, UniqueConstraint
from sqlalchemy.dialects.postgresql import (ARRAY, JSON)
from sqlalchemy import (Column, Integer, Text, ForeignKey, DateTime, Boolean,
                        func, CheckConstraint, and_)
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.sql.expression import text
from sqlalchemy.orm import relationship, backref, validates
from sqlalchemy.exc import IntegrityError
from croniter import croniter
import utc

import mettle_protocol as mp


Base = declarative_base()

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)


class Service(Base):
    __tablename__ = 'services'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    description = Column(Text)
    updated_by = Column(Text, nullable=False)

    @validates('name')
    def validate_name(self, key, name):
        # Ensure that the name has no characters that have special meanings in
        # RabbitMQ routing keys.
        assert '.' not in name
        assert '*' not in name
        assert '#' not in name
        return name


    __table_args__ = (
        UniqueConstraint('name'),
    )

    def __repr__(self):
        return 'Service <%s>' % self.name


class NotificationList(Base):
    __tablename__ = 'notification_lists'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    recipients = Column(ARRAY(Text), nullable=False)
    updated_by = Column(Text, nullable=False)
    def __repr__(self):
        return 'NotificationList <%s>' % self.name


class Pipeline(Base):
    __tablename__ = 'pipelines'
    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)
    service_id = Column(Integer, ForeignKey('services.id'), nullable=False)
    notification_list_id = Column(Integer, ForeignKey('notification_lists.id'),
                                  nullable=False)
    updated_by = Column(Text, nullable=False)
    active = Column(Boolean, nullable=False, server_default=text('true'))

    retries = Column(Integer, default=3)

    service = relationship("Service", backref=backref('pipelines', order_by=name))
    notification_list = relationship('NotificationList',
                                     backref=backref('pipelines', order_by=name))

    # A pipeline must have either a crontab or a trigger pipeline, but not both.
    crontab = Column(Text)
    chained_from_id = Column(Integer, ForeignKey('pipelines.id'))
    chained_from = relationship(lambda: Pipeline, remote_side=id,
                                backref='chains_to')

    @validates('name')
    def validate_name(self, key, name):
        # Ensure that the name has no characters that have special meanings in
        # RabbitMQ routing keys.
        assert '.' not in name
        assert '*' not in name
        assert '#' not in name
        return name

    @validates('crontab')
    def validate_crontab(self, key, cronspec):
        if cronspec is not None:
            # If the cronspec is not parseable, croniter will raise an exception
            # here.
            croniter(cronspec, utc.now())
        return cronspec

    __table_args__ = (
        UniqueConstraint('name', 'service_id'),
        CheckConstraint('crontab IS NOT NULL OR chained_from_id IS NOT NULL',
                        name='crontab_or_pipeline_check'),
        CheckConstraint('NOT (crontab IS NOT NULL AND chained_from_id IS NOT NULL)',
                        name='crontab_and_pipeline_check'),
    )

    def __repr__(self):
        return 'Pipeline <%s>' % self.name


class PipelineRun(Base):
    __tablename__ = 'pipeline_runs'
    id = Column(Integer, primary_key=True)
    pipeline_id = Column(Integer, ForeignKey('pipelines.id'), nullable=False)
    target_time = Column(DateTime(timezone=True), nullable=False)
    created_time = Column(DateTime(timezone=True), nullable=False,
                  server_default=func.now())
    succeeded = Column(Boolean, nullable=False, default=False,
                       server_default=text('false'))
    started_by = Column(Text, nullable=False) # username or 'timer'

    # These fields are set when we get an ack from the ETL service
    ack_time = Column(DateTime(timezone=True))
    targets = Column(MutableDict.as_mutable(JSON), default={})

    # end_time is set by dispatcher after it has heard in from all job runs
    end_time = Column(DateTime(timezone=True))

    pipeline = relationship("Pipeline", backref=backref('pipeline_runs',
                                                        order_by=created_time))

    def is_ended(self, db):
        if self.ack_time is None:
            return False
        return all(self.target_is_ended(db, t) for t in self.targets)

    def all_targets_succeeded(self, db):
        return all(self.target_is_succeeded(db, t) for t in self.targets)

    def target_is_succeeded(self, db, target):
        job = db.query(Job).filter(Job.pipeline_run==self,
                                   Job.target==target,
                                   Job.succeeded==True).first()
        return job is not None

    def target_is_failed(self, db, target):
        failure_count = db.query(Job).filter(Job.pipeline_run==self,
                                             Job.target==target,
                                             Job.end_time!=None,
                                             Job.succeeded==False).count()
        return failure_count >= self.pipeline.retries

    def target_is_ended(self, db, target):
        return (self.target_is_succeeded(db, target) or
                self.target_is_failed(db, target))

    def target_is_in_progress(self, db, target):
        job = db.query(Job).filter(Job.pipeline_run==self,
                                   Job.target==target,
                                   Job.end_time==None).first()
        return job is not None

    def target_deps_met(self, db, target):
        for dep in self.targets[target]:
            if not self.target_is_succeeded(db, dep):
                return False
        return True

    def target_is_ready(self, db, target):
        # Return true if the target meets these conditions:
        # 1. Is not ended.
        # 2. Does not already have an in-progress job in the DB.
        # 3. If it has dependencies, there is a successful job in the DB whose
        # target provides that dependency.
        if self.target_is_ended(db, target):
            return False
        if self.target_is_in_progress(db, target):
            return False
        if not self.target_deps_met(db, target):
            return False
        return True

    def get_ready_targets(self, db):
        return [t for t in self.targets if self.target_is_ready(db, t)]

    def make_job(self, db, target):
        job = Job(
            pipeline_run=self,
            target=target,
{'fraction_non_alphanumeric': 0.09083767764050561, 'fraction_numerical': 0.0005713061486824251, 'mean_word_length': 2.492269326683292, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1599039', 'n_tokens_mistral': 4037, 'n_tokens_neox': 3867, 'n_words': 1066}