<commit_before>import boto.datapipeline
import json
from . import base, keywords

class Pipeline(object):
    def __init__(self, name, unique_id, desc=None, region='us-east-1', pipeline_id=None):
        self.name        = name
        self.unique_id   = unique_id
        self.desc        = desc
        self.region_name = region
        self.pipeline_id = pipeline_id
        self.objects     = list()

    @property
    def region(self):
        try:
            return self._region
        except AttributeError:
            self._region = self.connect()
            return self._region

    def payload(self, pipeline_id=None):
        """ Pipeline payload. """
        payload = {
            'pipelineId'      : pipeline_id or self.pipeline_id,
            'pipelineObjects' : self.objects }

        return payload

    def definition(self, schedule,
        scheduleType        = keywords.scheduleType.cron,
        failureAndRerunMode = keywords.failureAndRerunMode.CASCADE,
        pipelineLogUri      = None,
        role                = 'DataPipelineDefaultRole',
        resourceRole        = 'DataPipelineDefaultResourceRole',
        **kwargs):
        default = base.DataPipelineObject(
                        name                = 'Default',
                        id                  = 'Default',
                        scheduleType        = scheduleType,
                        failureAndRerunMode = failureAndRerunMode,
                        pipelineLogUri      = pipelineLogUri,
                        role                = role,
                        resourceRole        = resourceRole,
                        schedule            = schedule,
                        **kwargs )
        return default

    def connect(self, region_name=None, aws_access_key_id=None, aws_secret_access_key=None, **kwargs):
        """ Connect to AWS region.

            Arguments:
                region_name           (str):  Optional AWS region name
                aws_access_key_id     (str):  Optional AWS Access Key
                aws_secret_access_key (str):  Optional AWS Secret Key

            Returns:
                A handle to the boto connection. """
        region_name = region_name or self.region_name
        kwargs.setdefault('aws_access_key_id', aws_access_key_id)
        kwargs.setdefault('aws_secret_access_key', aws_secret_access_key)
        return boto.datapipeline.connect_to_region(region_name, **kwargs)

    def activate(self, pipeline_id=None):
        """ Activate pipeline.

            Arguments:
                pipeline_id (str):  Optional ID of pipeline if called without create()

            Returns:
                boto response. """
        pipeline_id = pipeline_id or self.pipeline_id
        return self.region.activate_pipeline(pipeline_id)

    def update(self, pipeline_id=None):
        """ Update pipeline definition with self.objects.

            Arguments:
                pipeline_id (str):  Optional ID of pipeline if called without create()

            Returns:
                boto response. """
        payload = json.dumps(self.payload(pipeline_id))
        return self.region.make_request(action='PutPipelineDefinition', body=payload)

    def validate(self, pipeline_id=None):
        """ Validate the pipeline definition. """
        payload = json.dumps(self.payload(pipeline_id))
        return self.make_request(action='ValidatePipelineDefinition', body=payload)

    def create(self):
        """ Create pipeline and set pipeline_id. """
        response = self.region.create_pipeline(self.name, self.unique_id, self.desc)<commit_msg>Remove pipeline_id arg from pipeline methods<commit_after>import boto.datapipeline
import json
from . import base, keywords

class Pipeline(object):
    def __init__(self, name, unique_id, desc=None, region='us-east-1', pipeline_id=None):
        self.name        = name
        self.unique_id   = unique_id
        self.desc        = desc
        self.region_name = region
        self.pipeline_id = pipeline_id
        self.objects     = list()

    @property
    def region(self):
        try:
            return self._region
        except AttributeError:
            self._region = self.connect()
            return self._region

    def payload(self, pipeline_id=None):
        """ Pipeline payload. """
        payload = {
            'pipelineId'      : pipeline_id or self.pipeline_id,
            'pipelineObjects' : self.objects }

        return payload

    def definition(self, schedule,
        scheduleType        = keywords.scheduleType.cron,
        failureAndRerunMode = keywords.failureAndRerunMode.CASCADE,
        pipelineLogUri      = None,
        role                = 'DataPipelineDefaultRole',
        resourceRole        = 'DataPipelineDefaultResourceRole',
        **kwargs):
        default = base.DataPipelineObject(
                        name                = 'Default',
                        id                  = 'Default',
                        scheduleType        = scheduleType,
                        failureAndRerunMode = failureAndRerunMode,
                        pipelineLogUri      = pipelineLogUri,
                        role                = role,
                        resourceRole        = resourceRole,
                        schedule            = schedule,
                        **kwargs )
        return default

    def connect(self, region_name=None, aws_access_key_id=None, aws_secret_access_key=None, **kwargs):
        """ Connect to AWS region.

            Arguments:
                region_name           (str):  Optional AWS region name
                aws_access_key_id     (str):  Optional AWS Access Key
                aws_secret_access_key (str):  Optional AWS Secret Key

            Returns:
                A handle to the boto connection. """
        region_name = region_name or self.region_name
        kwargs.setdefault('aws_access_key_id', aws_access_key_id)
        kwargs.setdefault('aws_secret_access_key', aws_secret_access_key)
        return boto.datapipeline.connect_to_region(region_name, **kwargs)

    def activate(self):
        """ Activate pipeline. """
        assert self.pipeline_id is not None, "pipeline_id is None"
        return self.region.activate_pipeline(self.pipeline_id)

    def update(self):
        """ Update pipeline definition with self.objects. """
        assert self.pipeline_id is not None, "pipeline_id is None"
        payload = json.dumps(self.payload(self.pipeline_id))
        return self.region.make_request(action='PutPipelineDefinition', body=payload)

    def validate(self):
        """ Validate the pipeline definition. """
        assert self.pipeline_id is not None, "pipeline_id is None"
        payload = json.dumps(self.payload(self.pipeline_id))
        return self.make_request(action='ValidatePipelineDefinition', body=payload)

    def create(self):
        """ Create pipeline and set pipeline_id. """
        response = self.region.create_pipeline(self.name, self.unique_id, self.desc)
{'fraction_non_alphanumeric': 0.0804369414101291, 'fraction_numerical': 0.0002837281883955171, 'mean_word_length': 1.885796152271797, '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': '7084332', 'n_tokens_mistral': 1800, 'n_tokens_neox': 1772, 'n_words': 417}