ronantakizawa/codereview-bench huggingface dataset:
Repository: spotify/luigi
File: luigi/contrib/batch.py (Python)

Code before:
        """Register a job definition with AWS Batch, using a JSON"""
        with open(json_fpath) as f:
            job_def = json.load(f)
        response = self._client.register_job_definition(**job_def)
        status_code = response['ResponseMetadata']['HTTPStatusCode']
        if status_code != 200:
            msg = 'Register job definition request received status code {0}:\n{1}'
            raise Exception(msg.format(status_code, response))
        return response


class BatchTask(luigi.Task):

    """
    Base class for an Amazon Batch job

    Amazon Batch requires you to register "job definitions", which are JSON 
    descriptions for how to issue the ``docker run`` command. This Luigi Task 
    requires a pre-registered Batch jobDefinition name passed as a Parameter

    :param job_definition: name of pre-registered jobDefinition
    :param job_name: name of specific job, for tracking in the queue and logs.

    """
    job_definition = luigi.Parameter()
    job_name = luigi.Parameter(default=None)

    def run(self):
        bc = BatchClient()
        job_id = bc.submit_job(self.job_definition, self.parameters,
            job_name=self.job_name)
        bc.wait_on_job(job_id)

    @property
    def parameters():
        """Override to return a dict of parameters for the Batch Task"""
        return {}


Review comment (suggestion):
This can use the new `OptionalParameter` to avoid the String default is None warning (however, this feature has not yet been released to stable yet)

Code after:
        """Register a job definition with AWS Batch, using a JSON"""
        with open(json_fpath) as f:
            job_def = json.load(f)
        response = self._client.register_job_definition(**job_def)
        status_code = response['ResponseMetadata']['HTTPStatusCode']
        if status_code != 200:
            msg = 'Register job definition request received status code {0}:\n{1}'
            raise Exception(msg.format(status_code, response))
        return response


class BatchTask(luigi.Task):

    """
    Base class for an Amazon Batch job

    Amazon Batch requires you to register "job definitions", which are JSON
    descriptions for how to issue the ``docker run`` command. This Luigi Task
    requires a pre-registered Batch jobDefinition name passed as a Parameter

    :param job_definition (str): name of pre-registered jobDefinition
    :param job_name: name of specific job, for tracking in the queue and logs.

    """
    job_definition = luigi.Parameter()
    job_name = luigi.OptionalParameter(default=None)
    poll_time = luigi.IntParameter(default=POLL_TIME)

    def run(self):
        bc = BatchClient(self.poll_time)
        job_id = bc.submit_job(
            self.job_definition,
            self.parameters,
            job_name=self.job_name)
        bc.wait_on_job(job_id)

    @property
    def parameters(self):
        """Override to return a dict of parameters for the Batch Task"""
        return {}
Repository: spotify/luigi
File: luigi/contrib/datadog_metric.py (Python)

Code before:

        api.Event.create(**params)

    def _send_gauge(self, metric_name, value, tags=[]):
        all_tags = tags + self.default_tags()

        namespaced_metric = "{namespace}.{metric_name}".format(namespace=self._config.metric_namespace,
                                                               metric_name=metric_name)
        statsd.gauge(namespaced_metric, value, tags=all_tags)

    def _send_increment(self, metric_name, value=1, tags=[]):
        all_tags = tags + self.default_tags()

        namespaced_metric = "{namespace}.{metric_name}".format(namespace=self._config.metric_namespace,
                                                               metric_name=metric_name)
        statsd.increment(namespaced_metric, value, tags=all_tags)

    def _format_task_params_to_tags(self, task):
        params = []
        for key, value in task.params.items():
            params.append("{key}:{value}".format(key=key, value=value))

        return params

    def default_tags(self):
        default_tags = []

        env_tag = "environment:{environment}".format(environment=self._config.environment)
        default_tags.append(env_tag)

        if self._config.default_tags:
            default_tags = default_tags + str.split(self._config.default_tags, ',')

        return default_tags


Review comment (suggestion):
One of your last commits rose this to my attention: should this be a `@property`? It's utilizing `self._config.default_tags` which is a property. Sorry for not noticing sooner!

Code after:

        api.Event.create(**params)

    def _send_gauge(self, metric_name, value, tags=[]):
        all_tags = tags + self.default_tags

        namespaced_metric = "{namespace}.{metric_name}".format(namespace=self._config.metric_namespace,
                                                               metric_name=metric_name)
        statsd.gauge(namespaced_metric, value, tags=all_tags)

    def _send_increment(self, metric_name, value=1, tags=[]):
        all_tags = tags + self.default_tags

        namespaced_metric = "{namespace}.{metric_name}".format(namespace=self._config.metric_namespace,
                                                               metric_name=metric_name)
        statsd.increment(namespaced_metric, value, tags=all_tags)

    def _format_task_params_to_tags(self, task):
        params = []
        for key, value in task.params.items():
            params.append("{key}:{value}".format(key=key, value=value))

        return params

    @property
    def default_tags(self):
        default_tags = []

        env_tag = "environment:{environment}".format(environment=self._config.environment)
        default_tags.append(env_tag)

        if self._config.default_tags:
            default_tags = default_tags + str.split(self._config.default_tags, ',')

        return default_tags
Repository: spotify/luigi
File: luigi/configuration/base_parser.py (Python)

Code before:
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging


class BaseParser:
    _instance = None
    _config_paths = []

    @classmethod
    def instance(cls, *args, **kwargs):
        """ Singleton getter """
        if cls._instance is None:
            cls._instance = cls(*args, **kwargs)
            loaded = cls._instance.reload()
            logging.getLogger('luigi-interface').info('Loaded %r', loaded)

        return cls._instance

    @classmethod
    def add_config_path(cls, path):
        cls._config_paths.append(path)
        cls.reload()

    @classmethod
    def reload(cls):
        return cls.instance().read(cls._config_paths)


Review comment (suggestion):
side note: am i correct that neither the client nor the server's logging config names are configurable? I've found that interesting - why statically set to `luigi-interface`. Perhaps we don't want that changed in general.

Code after:
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging


# IMPORTANT: don't inherit from `object`!
# ConfigParser have some troubles in this case.
# More info: https://stackoverflow.com/a/19323238
class BaseParser:
    @classmethod
    def instance(cls, *args, **kwargs):
        """ Singleton getter """
        if cls._instance is None:
            cls._instance = cls(*args, **kwargs)
            loaded = cls._instance.reload()
            logging.getLogger('luigi-interface').info('Loaded %r', loaded)

        return cls._instance

    @classmethod
    def add_config_path(cls, path):
        cls._config_paths.append(path)
        cls.reload()

    @classmethod
    def reload(cls):
        return cls.instance().read(cls._config_paths)
Repository: spotify/luigi
File: luigi/configuration/base_parser.py (Python)

Code before:
# -*- coding: utf-8 -*-
#
# Copyright 2012-2015 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging


# IMPORTANT: don't inherit from `object`!
# ConfigParser have some troubles in this case.
class BaseParser:
    @classmethod
    def instance(cls, *args, **kwargs):
        """ Singleton getter """
        if cls._instance is None:
            cls._instance = cls(*args, **kwargs)
            loaded = cls._instance.reload()
            logging.getLogger('luigi-interface').info('Loaded %r', loaded)

        return cls._instance

    @classmethod
    def add_config_path(cls, path):
        cls._config_paths.append(path)
        cls.reload()

    @classmethod
    def reload(cls):
        return cls.instance().read(cls._config_paths)


Review comment (bug):
The issue is explained better [on SO](https://stackoverflow.com/questions/19323144/extending-class-in-python-2-7-usage-of-super)

Code after:
# -*- coding: utf-8 -*-
#
# Copyright 2012-2015 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import logging


# IMPORTANT: don't inherit from `object`!
# ConfigParser have some troubles in this case.
# More info: https://stackoverflow.com/a/19323238
class BaseParser:
    @classmethod
    def instance(cls, *args, **kwargs):
        """ Singleton getter """
        if cls._instance is None:
            cls._instance = cls(*args, **kwargs)
            loaded = cls._instance.reload()
            logging.getLogger('luigi-interface').info('Loaded %r', loaded)

        return cls._instance

    @classmethod
    def add_config_path(cls, path):
        cls._config_paths.append(path)
        cls.reload()

    @classmethod
    def reload(cls):
        return cls.instance().read(cls._config_paths)
Repository: spotify/luigi
File: luigi/configuration/core.py (Python)

Code before:

    return parser_class.instance()


def add_config_path(path):
    """Select config parser by file extension and add path into parser.
    """
    if not os.path.isfile(path):
        warnings.warn("Config file does not exist: {path}".format(path=path))
        return False

    # select parser by file extension
    _base, ext = os.path.splitext(path)
    if ext and ext[1:] in PARSERS:
        parser = ext[1:]
    else:
        parser = PARSER
    parser_class = PARSERS[parser]

    if not parser_class.enabled:
        msg = (
                "Parser not installed yet. "
                "Please, install luigi with required parser:\n"
                "pip install luigi[{parser}]"
            )
        raise ImportError(msg.format(parser=parser))

    if parser != PARSER:
        msg = (
            "Config for {added} parser added, but used {used} parser. "
            "Set up right parser via env var: "
            "export LUIGI_CONFIG_PARSER={added}"
        )
        warnings.warn(msg.format(added=parser, used=PARSER))

    # add config path to parser
    parser_class.add_config_path(path)
    return True


if 'LUIGI_CONFIG_PATH' in os.environ:
    add_config_path(os.environ['LUIGI_CONFIG_PATH'])


Review comment (refactor):
This `if not parser_class.enabled` is checked at least twice here. Can that be extracted into a method or something for DRY programming purposes?

Code after:
    """
    parser_class = PARSERS[parser]
    _check_parser(parser_class, parser)
    return parser_class.instance()


def add_config_path(path):
    """Select config parser by file extension and add path into parser.
    """
    if not os.path.isfile(path):
        warnings.warn("Config file does not exist: {path}".format(path=path))
        return False

    # select parser by file extension
    _base, ext = os.path.splitext(path)
    if ext and ext[1:] in PARSERS:
        parser = ext[1:]
    else:
        parser = PARSER
    parser_class = PARSERS[parser]

    _check_parser(parser_class, parser)
    if parser != PARSER:
        msg = (
            "Config for {added} parser added, but used {used} parser. "
            "Set up right parser via env var: "
            "export LUIGI_CONFIG_PARSER={added}"
        )
        warnings.warn(msg.format(added=parser, used=PARSER))

    # add config path to parser
    parser_class.add_config_path(path)
    return True


if 'LUIGI_CONFIG_PATH' in os.environ:
    add_config_path(os.environ['LUIGI_CONFIG_PATH'])
Repository: spotify/luigi
File: test/mock_test.py (Python)

Code before:
        with t.open('r') as b:
            self.assertEqual(list(b), ['bar'])

    def test_bytes(self):
        t = MockTarget("foo", format=Nop)
        with t.open('wb') as b:
            b.write(b"bar")

        with t.open('rb') as b:
            self.assertEqual(list(b), [b'bar'])

    def test_default_mode_value(self):
        t = MockTarget("foo")
        with t.open('w') as b:
            b.write("bar")

        with t.open() as b:
            self.assertEqual(list(b), ['bar'])

    def test_mode_none_error(self):
        t = MockTarget("foo")
        try:
            with t.open(None) as b:
                b.write("bar")
        except TypeError:
            self.assertRaises(TypeError)

    # That should work in python2 because of the autocast
    # That should work in python3 because the default format is Text
    def test_unicode(self):
        t = MockTarget("foo")
        with t.open('w') as b:
            b.write(u"bar")

        with t.open('r') as b:
            self.assertEqual(b.read(), u'bar')


class MockFileSystemTest(unittest.TestCase):
    fs = MockFileSystem()

    def _touch(self, path):
        t = MockTarget(path)
        with t.open('w'):
            pass

    def setUp(self):
        self.fs.clear()
        self.path = "/tmp/foo"
        self.path2 = "/tmp/bar"
        self.path3 = "/tmp/foobar"

Review comment (suggestion):
with this being in an `except`, we can't guarantee that the assertion is risen

i was expecting

```python
t = MockTarget("foo")
with self.assertRaises(TypeError):
    with t.open(None) as b:
        b.write("bar")
```

Code after:
        with t.open('r') as b:
            self.assertEqual(list(b), ['bar'])

    def test_bytes(self):
        t = MockTarget("foo", format=Nop)
        with t.open('wb') as b:
            b.write(b"bar")

        with t.open('rb') as b:
            self.assertEqual(list(b), [b'bar'])

    def test_default_mode_value(self):
        t = MockTarget("foo")
        with t.open('w') as b:
            b.write("bar")

        with t.open() as b:
            self.assertEqual(list(b), ['bar'])

    def test_mode_none_error(self):
        t = MockTarget("foo")
        with self.assertRaises(TypeError):
            with t.open(None) as b:
                b.write("bar")

    # That should work in python2 because of the autocast
    # That should work in python3 because the default format is Text
    def test_unicode(self):
        t = MockTarget("foo")
        with t.open('w') as b:
            b.write(u"bar")

        with t.open('r') as b:
            self.assertEqual(b.read(), u'bar')


class MockFileSystemTest(unittest.TestCase):
    fs = MockFileSystem()

    def _touch(self, path):
        t = MockTarget(path)
        with t.open('w'):
            pass

    def setUp(self):
        self.fs.clear()
        self.path = "/tmp/foo"
        self.path2 = "/tmp/bar"
        self.path3 = "/tmp/foobar"
        self._touch(self.path)
        self._touch(self.path2)
Repository: spotify/luigi
File: luigi/freezing.py (Python)

Code before:
from collections import OrderedDict, Mapping
import operator
import functools


class FrozenOrderedDict(Mapping):
    """
    It is an immutable wrapper around ordered dictionaries that implements the complete :py:class:`collections.Mapping`
    interface. It can be used as a drop-in replacement for dictionaries where immutability and ordering are desired.
    """

    def __init__(self, *args, **kwargs):
        self.__dict = OrderedDict(*args, **kwargs)
        self.__hash = None

    def __getitem__(self, key):
        return self.__dict[key]

    def __iter__(self):
        return iter(self.__dict)

    def __len__(self):
        return len(self.__dict)

    def __repr__(self):
        # We should use short representation for beautiful console output
        return repr(dict(self.__dict))

    def __hash__(self):
        if self.__hash is None:
            hashes = map(hash, self.items())
            self.__hash = functools.reduce(operator.xor, hashes, 0)

        return self.__hash

    def get_wrapped(self):
        return self.__dict

Review comment (bug):
If you copy pasted this class from somewhere (be it StackOverflow) please include a reference from where.

Code after:
"""Internal-only module with immutable data structures.

Please, do not use it outside of Luigi codebase itself.
"""


from collections import OrderedDict, Mapping
import operator
import functools


class FrozenOrderedDict(Mapping):
    """
    It is an immutable wrapper around ordered dictionaries that implements the complete :py:class:`collections.Mapping`
    interface. It can be used as a drop-in replacement for dictionaries where immutability and ordering are desired.
    """

    def __init__(self, *args, **kwargs):
        self.__dict = OrderedDict(*args, **kwargs)
        self.__hash = None

    def __getitem__(self, key):
        return self.__dict[key]

    def __iter__(self):
        return iter(self.__dict)

    def __len__(self):
        return len(self.__dict)

    def __repr__(self):
        # We should use short representation for beautiful console output
        return repr(dict(self.__dict))

    def __hash__(self):
        if self.__hash is None:
            hashes = map(hash, self.items())
Repository: spotify/luigi
File: luigi/freezing.py (Python)

Code before:
from collections import OrderedDict, Mapping
import operator
import functools


class FrozenOrderedDict(Mapping):
    """
    It is an immutable wrapper around ordered dictionaries that implements the complete :py:class:`collections.Mapping`
    interface. It can be used as a drop-in replacement for dictionaries where immutability and ordering are desired.
    """

    def __init__(self, *args, **kwargs):
        self.__dict = OrderedDict(*args, **kwargs)
        self.__hash = None

    def __getitem__(self, key):
        return self.__dict[key]

    def __iter__(self):
        return iter(self.__dict)

    def __len__(self):
        return len(self.__dict)

    def __repr__(self):
        # We should use short representation for beautiful console output
        return repr(dict(self.__dict))

    def __hash__(self):
        if self.__hash is None:
            hashes = map(hash, self.items())
            self.__hash = functools.reduce(operator.xor, hashes, 0)

Review comment (suggestion):
Add module docs please, and clearly state that nothing in this module is meant to be used by luigi users, it's internal only.

Code after:
"""Internal-only module with immutable data structures.

Please, do not use it outside of Luigi codebase itself.
"""


from collections import OrderedDict, Mapping
import operator
import functools


class FrozenOrderedDict(Mapping):
    """
    It is an immutable wrapper around ordered dictionaries that implements the complete :py:class:`collections.Mapping`
    interface. It can be used as a drop-in replacement for dictionaries where immutability and ordering are desired.
    """

    def __init__(self, *args, **kwargs):
        self.__dict = OrderedDict(*args, **kwargs)
        self.__hash = None

    def __getitem__(self, key):
        return self.__dict[key]

    def __iter__(self):
        return iter(self.__dict)

    def __len__(self):
        return len(self.__dict)

    def __repr__(self):
        # We should use short representation for beautiful console output
Repository: spotify/luigi
File: luigi/contrib/spark.py (Python)

Code before:

        super(SparkSubmitTask, self).run()

    def app_options(self):
        """
        Subclass this method to map your task parameters to the app's arguments

        """
        return []

    def _upd_env_if_not_none(self, env, prop):
        var = getattr(self, prop.lower(), None)
        if var:
            env[prop] = var

    @property
    def pyspark_python(self):
        return None

    @property
    def pyspark_driver_python(self):
        return None

    @property
    def hadoop_user_name(self):
        return os.getlogin()

    @property
    def spark_version(self):
        return "spark"

    @property
    def spark_submit(self):
        return configuration.get_config().get(self.spark_version, 'spark-submit', 'spark-submit')

    @property
    def master(self):
        return configuration.get_config().get(self.spark_version, "master", None)

    @property
    def deploy_mode(self):
        return configuration.get_config().get(self.spark_version, "deploy-mode", None)

    @property
    def jars(self):
        return self._list_config(configuration.get_config().get(self.spark_version, "jars", None))

    @property
    def packages(self):
        return self._list_config(configuration.get_config().get(
            self.spark_version, "packages", None))

Review comment (suggestion):
It is a bit dangerous, in some environments HADOOP_USER_NAME may be already set and this will override it, right?  Probably it would be safer to return None

Code after:
        """
        return []

    @property
    def pyspark_python(self):
        return None

    @property
    def pyspark_driver_python(self):
        return None

    @property
    def hadoop_user_name(self):
        return None

    @property
    def spark_version(self):
        return "spark"

    @property
    def spark_submit(self):
        return configuration.get_config().get(self.spark_version, 'spark-submit', 'spark-submit')

    @property
    def master(self):
        return configuration.get_config().get(self.spark_version, "master", None)

    @property
    def deploy_mode(self):
        return configuration.get_config().get(self.spark_version, "deploy-mode", None)

    @property
    def jars(self):
        return self._list_config(configuration.get_config().get(self.spark_version, "jars", None))

    @property
    def packages(self):
        return self._list_config(configuration.get_config().get(
            self.spark_version, "packages", None))

    @property
    def py_files(self):
        return self._list_config(configuration.get_config().get(
            self.spark_version, "py-files", None))

    @property
    def files(self):
        return self._list_config(configuration.get_config().get(self.spark_version, "files", None))

    @property
    def _conf(self):