benjamintli/cosqa-llm-filtered-hard-negatives huggingface dataset:
Query: how to capture failure in behave python

Relevant code:
def handle_test(self, command, **options):
        """Send a test error to APM Server"""
        # can't be async for testing
        config = {"async_mode": False}
        for key in ("service_name", "secret_token"):
            if options.get(key):
                config[key] = options[key]
        client = DjangoClient(**config)
        client.error_logger = ColoredLogger(self.stderr)
        client.logger = ColoredLogger(self.stderr)
        self.write(
            "Trying to send a test error to APM Server using these settings:\n\n"
            "SERVICE_NAME:\t%s\n"
            "SECRET_TOKEN:\t%s\n"
            "SERVER:\t\t%s\n\n" % (client.config.service_name, client.config.secret_token, client.config.server_url)
        )

        try:
            raise TestException("Hi there!")
        except TestException:
            client.capture_exception()
            if not client.error_logger.errors:
                self.write(
                    "Success! We tracked the error successfully! You should be"
                    " able to see it in a few seconds at the above URL"
                )
        finally:
            client.close()

Similar but not relevant code:
def on_error(e):  # pragma: no cover
    """Error handler

    RuntimeError or ValueError exceptions raised by commands will be handled
    by this function.
    """
    exname = {'RuntimeError': 'Runtime error', 'Value Error': 'Value error'}
    sys.stderr.write('{}: {}\n'.format(exname[e.__class__.__name__], str(e)))
    sys.stderr.write('See file slam_error.log for additional details.\n')
    sys.exit(1)

Similar but not relevant code:
def lambda_failure_response(*args):
        """
        Helper function to create a Lambda Failure Response

        :return: A Flask Response
        """
        response_data = jsonify(ServiceErrorResponses._LAMBDA_FAILURE)
        return make_response(response_data, ServiceErrorResponses.HTTP_STATUS_CODE_502)

Similar but not relevant code:
def prepare_for_reraise(error, exc_info=None):
    """Prepares the exception for re-raising with reraise method.

    This method attaches type and traceback info to the error object
    so that reraise can properly reraise it using this info.

    """
    if not hasattr(error, "_type_"):
        if exc_info is None:
            exc_info = sys.exc_info()
        error._type_ = exc_info[0]
        error._traceback = exc_info[2]
    return error

Similar but not relevant code:
def raise_os_error(_errno, path=None):
    """
    Helper for raising the correct exception under Python 3 while still
    being able to raise the same common exception class in Python 2.7.
    """

    msg = "%s: '%s'" % (strerror(_errno), path) if path else strerror(_errno)
    raise OSError(_errno, msg)
Query: python load jsonlines file to object

Relevant code:
def load(cls, fp, **kwargs):
    """wrapper for :py:func:`json.load`"""
    json_obj = json.load(fp, **kwargs)
    return parse(cls, json_obj)

Similar but not relevant code:
def open_json(file_name):
    """
    returns json contents as string
    """
    with open(file_name, "r") as json_data:
        data = json.load(json_data)
        return data

Similar but not relevant code:
def from_file(file_path) -> dict:
        """ Load JSON file """
        with io.open(file_path, 'r', encoding='utf-8') as json_stream:
            return Json.parse(json_stream, True)

Similar but not relevant code:
def _loadfilepath(self, filepath, **kwargs):
        """This loads a geojson file into a geojson python
        dictionary using the json module.
        
        Note: to load with a different text encoding use the encoding argument.
        """
        with open(filepath, "r") as f:
            data = json.load(f, **kwargs)
        return data

Similar but not relevant code:
def read_json(location):
    """Open and load JSON from file.

    location (Path): Path to JSON file.
    RETURNS (dict): Loaded JSON content.
    """
    location = ensure_path(location)
    with location.open('r', encoding='utf8') as f:
        return ujson.load(f)
Query: python load rds file

Relevant code:
def load_graph_from_rdf(fname):
    """ reads an RDF file into a graph """
    print("reading RDF from " + fname + "....")
    store = Graph()
    store.parse(fname, format="n3")
    print("Loaded " + str(len(store)) + " tuples")
    return store

Similar but not relevant code:
def load_db(file, db, verbose=True):
    """
    Load :class:`mongomock.database.Database` from a local file.

    :param file: file path.
    :param db: instance of :class:`mongomock.database.Database`.
    :param verbose: bool, toggle on log.
    :return: loaded db.
    """
    db_data = json.load(file, verbose=verbose)
    return _load(db_data, db)

Similar but not relevant code:
def list_rds(region, filter_by_kwargs):
    """List all RDS thingys."""
    conn = boto.rds.connect_to_region(region)
    instances = conn.get_all_dbinstances()
    return lookup(instances, filter_by=filter_by_kwargs)

Similar but not relevant code:
def read_las(source, closefd=True):
    """ Entry point for reading las data in pylas

    Reads the whole file into memory.

    >>> las = read_las("pylastests/simple.las")
    >>> las.classification
    array([1, 1, 1, ..., 1, 1, 1], dtype=uint8)

    Parameters
    ----------
    source : str or io.BytesIO
        The source to read data from

    closefd: bool
            if True and the source is a stream, the function will close it
            after it is done reading


    Returns
    -------
    pylas.lasdatas.base.LasBase
        The object you can interact with to get access to the LAS points & VLRs
    """
    with open_las(source, closefd=closefd) as reader:
        return reader.read()

Similar but not relevant code:
def connect_rds(aws_access_key_id=None, aws_secret_access_key=None, **kwargs):
    """
    :type aws_access_key_id: string
    :param aws_access_key_id: Your AWS Access Key ID

    :type aws_secret_access_key: string
    :param aws_secret_access_key: Your AWS Secret Access Key

    :rtype: :class:`boto.rds.RDSConnection`
    :return: A connection to RDS
    """
    from boto.rds import RDSConnection
    return RDSConnection(aws_access_key_id, aws_secret_access_key, **kwargs)
Query: python locate tuple in list

Relevant code:
def tuple_search(t, i, v):
    """
    Search tuple array by index and value
    :param t: tuple array
    :param i: index of the value in each tuple
    :param v: value
    :return: the first tuple in the array with the specific index / value
    """
    for e in t:
        if e[i] == v:
            return e
    return None

Similar but not relevant code:
def tuple_check(*args, func=None):
    """Check if arguments are tuple type."""
    func = func or inspect.stack()[2][3]
    for var in args:
        if not isinstance(var, (tuple, collections.abc.Sequence)):
            name = type(var).__name__
            raise TupleError(
                f'Function {func} expected tuple, {name} got instead.')

Similar but not relevant code:
def option2tuple(opt):
    """Return a tuple of option, taking possible presence of level into account"""

    if isinstance(opt[0], int):
        tup = opt[1], opt[2:]
    else:
        tup = opt[0], opt[1:]

    return tup

Similar but not relevant code:
def isnamedtuple(obj):
    """Heuristic check if an object is a namedtuple."""
    return isinstance(obj, tuple) \
           and hasattr(obj, "_fields") \
           and hasattr(obj, "_asdict") \
           and callable(obj._asdict)
Query: python location of substring in string

Relevant code:
def _find(string, sub_string, start_index):
    """Return index of sub_string in string.

    Raise TokenError if sub_string is not found.
    """
    result = string.find(sub_string, start_index)
    if result == -1:
        raise TokenError("expected '{0}'".format(sub_string))
    return result

Similar but not relevant code:
def split(s):
  """Uses dynamic programming to infer the location of spaces in a string without spaces."""
  l = [_split(x) for x in _SPLIT_RE.split(s)]
  return [item for sublist in l for item in sublist]

Similar but not relevant code:
def long_substr(data):
    """Return the longest common substring in a list of strings.
    
    Credit: http://stackoverflow.com/questions/2892931/longest-common-substring-from-more-than-two-strings-python
    """
    substr = ''
    if len(data) > 1 and len(data[0]) > 0:
        for i in range(len(data[0])):
            for j in range(len(data[0])-i+1):
                if j > len(substr) and all(data[0][i:i+j] in x for x in data):
                    substr = data[0][i:i+j]
    elif len(data) == 1:
        substr = data[0]
    return substr

Similar but not relevant code:
def valid_substitution(strlen, index):
    """
    skip performing substitutions that are outside the bounds of the string
    """
    values = index[0]
    return all([strlen > i for i in values])
Query: python lock file delete

Relevant code:
def assert_lock(fname):
    """
    If file is locked then terminate program else lock file.
    """

    if not set_lock(fname):
        logger.error('File {} is already locked. Terminating.'.format(fname))
        sys.exit()

Similar but not relevant code:
def Unlock(fd, path):
  """Release the lock on the file.

  Args:
    fd: int, the file descriptor of the file to unlock.
    path: string, the name of the file to lock.

  Raises:
    IOError, raised from flock while attempting to release a file lock.
  """
  try:
    fcntl.flock(fd, fcntl.LOCK_UN | fcntl.LOCK_NB)
  except IOError as e:
    if e.errno == errno.EWOULDBLOCK:
      raise IOError('Exception unlocking %s. Locked by another process.' % path)
    else:
      raise IOError('Exception unlocking %s. %s.' % (path, str(e)))

Similar but not relevant code:
def __enter__(self):
        """Acquire a lock on the output file, prevents collisions between multiple runs."""
        self.fd = open(self.filename, 'a')
        fcntl.lockf(self.fd, fcntl.LOCK_EX)
        return self.fd

Similar but not relevant code:
def acquire_nix(lock_file):  # pragma: no cover
    """Acquire a lock file on linux or osx."""
    fd = os.open(lock_file, OPEN_MODE)

    try:
        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except (IOError, OSError):
        os.close(fd)
    else:
        return fd

Similar but not relevant code:
def lock_file(f, block=False):
    """
    If block=False (the default), die hard and fast if another process has
    already grabbed the lock for this file.

    If block=True, wait for the lock to be released, then continue.
    """
    try:
        flags = fcntl.LOCK_EX
        if not block:
            flags |= fcntl.LOCK_NB
        fcntl.flock(f.fileno(), flags)
    except IOError as e:
        if e.errno in (errno.EACCES, errno.EAGAIN):
            raise SystemExit("ERROR: %s is locked by another process." %
                             f.name)
        raise
Query: how to change spaces to underscores python

Relevant code:
def normalise_string(string):
    """ Strips trailing whitespace from string, lowercases it and replaces
        spaces with underscores
    """
    string = (string.strip()).lower()
    return re.sub(r'\W+', '_', string)

Similar but not relevant code:
def camelcase_underscore(name):
    """ Convert camelcase names to underscore """
    s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
    return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()

Similar but not relevant code:
def mixedcase(path):
    """Removes underscores and capitalizes the neighbouring character"""
    words = path.split('_')
    return words[0] + ''.join(word.title() for word in words[1:])

Similar but not relevant code:
def us2mc(string):
    """Transform an underscore_case string to a mixedCase string"""
    return re.sub(r'_([a-z])', lambda m: (m.group(1).upper()), string)
Query: how to change text print color in python

Relevant code:
def printc(cls, txt, color=colors.red):
        """Print in color."""
        print(cls.color_txt(txt, color))

Similar but not relevant code:
def _update_fontcolor(self, fontcolor):
        """Updates text font color button

        Parameters
        ----------

        fontcolor: Integer
        \tText color in integer RGB format

        """

        textcolor = wx.SystemSettings_GetColour(wx.SYS_COLOUR_WINDOWTEXT)
        textcolor.SetRGB(fontcolor)

        self.textcolor_choice.SetColour(textcolor)
Query: python logging formatter color

Relevant code:
def clog(color):
    """Same to ``log``, but this one centralizes the message first."""
    logger = log(color)
    return lambda msg: logger(centralize(msg).rstrip())

Similar but not relevant code:
def print_log(text, *colors):
    """Print a log message to standard error."""
    sys.stderr.write(sprint("{}: {}".format(script_name, text), *colors) + "\n")

Similar but not relevant code:
def handle_logging(self):
        """
        To allow devs to log as early as possible, logging will already be
        handled here
        """

        configure_logging(self.get_scrapy_options())

        # Disable duplicates
        self.__scrapy_options["LOG_ENABLED"] = False

        # Now, after log-level is correctly set, lets log them.
        for msg in self.log_output:
            if msg["level"] is "error":
                self.log.error(msg["msg"])
            elif msg["level"] is "info":
                self.log.info(msg["msg"])
            elif msg["level"] is "debug":
                self.log.debug(msg["msg"])

Similar but not relevant code:
def pylog(self, *args, **kwargs):
        """Display all available logging information."""
        printerr(self.name, args, kwargs, traceback.format_exc())

Similar but not relevant code:
def info(self, message, *args, **kwargs):
        """More important level : default for print and save
        """
        self._log(logging.INFO, message, *args, **kwargs)
Query: how to change time zone of a column to ist in python

Relevant code:
def execute_cast_simple_literal_to_timestamp(op, data, type, **kwargs):
    """Cast integer and strings to timestamps"""
    return pd.Timestamp(data, tz=type.timezone)

Similar but not relevant code:
def normalize(self, dt, is_dst=False):
        """Correct the timezone information on the given datetime"""
        if dt.tzinfo is self:
            return dt
        if dt.tzinfo is None:
            raise ValueError('Naive time - no tzinfo set')
        return dt.astimezone(self)

Similar but not relevant code:
def make_aware(value, timezone):
    """
    Makes a naive datetime.datetime in a given time zone aware.
    """
    if hasattr(timezone, 'localize') and value not in (datetime.datetime.min, datetime.datetime.max):
        # available for pytz time zones
        return timezone.localize(value, is_dst=None)
    else:
        # may be wrong around DST changes
        return value.replace(tzinfo=timezone)

Similar but not relevant code:
def get_timezone() -> Tuple[datetime.tzinfo, str]:
    """Discover the current time zone and it's standard string representation (for source{d})."""
    dt = get_datetime_now().astimezone()
    tzstr = dt.strftime("%z")
    tzstr = tzstr[:-2] + ":" + tzstr[-2:]
    return dt.tzinfo, tzstr

Similar but not relevant code:
def timestamp(format=DATEFMT, timezone='Africa/Johannesburg'):
    """ Return current datetime with timezone applied
        [all timezones] print sorted(pytz.all_timezones) """

    return formatdate(datetime.now(tz=pytz.timezone(timezone)))

Similar but not relevant code:
def current_offset(local_tz=None):
    """
    Returns current utcoffset for a timezone. Uses
    DEFAULT_LOCAL_TZ by default. That value can be
    changed at runtime using the func below.
    """
    if local_tz is None:
        local_tz = DEFAULT_LOCAL_TZ
    dt = local_tz.localize(datetime.now())
    return dt.utcoffset()
Query: python logging get file handler name

Relevant code:
def find_console_handler(logger):
    """Return a stream handler, if it exists."""
    for handler in logger.handlers:
        if (isinstance(handler, logging.StreamHandler) and
                handler.stream == sys.stderr):
            return handler

Similar but not relevant code:
def logger(message, level=10):
    """Handle logging."""
    logging.getLogger(__name__).log(level, str(message))

Similar but not relevant code:
def log(logger, level, message):
    """Logs message to stderr if logging isn't initialized."""

    if logger.parent.name != 'root':
        logger.log(level, message)
    else:
        print(message, file=sys.stderr)

Similar but not relevant code:
def setupLogFile(self):
		"""Set up the logging file for a new session- include date and some whitespace"""
		self.logWrite("\n###############################################")
		self.logWrite("calcpkg.py log from " + str(datetime.datetime.now()))
		self.changeLogging(True)

Similar but not relevant code:
def handle_logging(self):
        """
        To allow devs to log as early as possible, logging will already be
        handled here
        """

        configure_logging(self.get_scrapy_options())

        # Disable duplicates
        self.__scrapy_options["LOG_ENABLED"] = False

        # Now, after log-level is correctly set, lets log them.
        for msg in self.log_output:
            if msg["level"] is "error":
                self.log.error(msg["msg"])
            elif msg["level"] is "info":
                self.log.info(msg["msg"])
            elif msg["level"] is "debug":
                self.log.debug(msg["msg"])
Query: how to change window title python

Relevant code:
def title(msg):
    """Sets the title of the console window."""
    if sys.platform.startswith("win"):
        ctypes.windll.kernel32.SetConsoleTitleW(tounicode(msg))

Similar but not relevant code:
def getFileDialogTitle(msg, title):
    """
    Create nicely-formatted string based on arguments msg and title
    :param msg: the msg to be displayed
    :param title: the window title
    :return: None
    """
    if msg and title:
        return "%s - %s" % (title, msg)
    if msg and not title:
        return str(msg)
    if title and not msg:
        return str(title)
    return None

Similar but not relevant code:
def inFocus(self):
        """Set GUI on-top flag"""
        previous_flags = self.window.flags()
        self.window.setFlags(previous_flags |
                             QtCore.Qt.WindowStaysOnTopHint)

Similar but not relevant code:
def get_active_window():
    """Get the currently focused window
    """
    active_win = None
    default = wnck.screen_get_default()
    while gtk.events_pending():
        gtk.main_iteration(False)
    window_list = default.get_windows()
    if len(window_list) == 0:
        print "No Windows Found"
    for win in window_list:
        if win.is_active():
            active_win = win.get_name()
    return active_win

Similar but not relevant code:
def win32_refresh_window(cls):
        """
        Call win32 API to refresh the whole Window.

        This is sometimes necessary when the application paints background
        for completion menus. When the menu disappears, it leaves traces due
        to a bug in the Windows Console. Sending a repaint request solves it.
        """
        # Get console handle
        handle = windll.kernel32.GetConsoleWindow()

        RDW_INVALIDATE = 0x0001
        windll.user32.RedrawWindow(handle, None, None, c_uint(RDW_INVALIDATE))
Query: python logging not print immediately

Relevant code:
def _debug_log(self, msg):
        """Debug log messages if debug=True"""
        if not self.debug:
            return
        sys.stderr.write('{}\n'.format(msg))

Similar but not relevant code:
def handle_logging(self):
        """
        To allow devs to log as early as possible, logging will already be
        handled here
        """

        configure_logging(self.get_scrapy_options())

        # Disable duplicates
        self.__scrapy_options["LOG_ENABLED"] = False

        # Now, after log-level is correctly set, lets log them.
        for msg in self.log_output:
            if msg["level"] is "error":
                self.log.error(msg["msg"])
            elif msg["level"] is "info":
                self.log.info(msg["msg"])
            elif msg["level"] is "debug":
                self.log.debug(msg["msg"])

Similar but not relevant code:
def print_log(text, *colors):
    """Print a log message to standard error."""
    sys.stderr.write(sprint("{}: {}".format(script_name, text), *colors) + "\n")

Similar but not relevant code:
def log_exception(exc_info=None, stream=None):
    """Log the 'exc_info' tuple in the server log."""
    exc_info = exc_info or sys.exc_info()
    stream = stream or sys.stderr
    try:
        from traceback import print_exception
        print_exception(exc_info[0], exc_info[1], exc_info[2], None, stream)
        stream.flush()
    finally:
        exc_info = None

Similar but not relevant code:
def pout(msg, log=None):
    """Print 'msg' to stdout, and option 'log' at info level."""
    _print(msg, sys.stdout, log_func=log.info if log else None)
Query: python logging nothing after delete log file

Relevant code:
def close_log(log, verbose=True):
    """Close log

    This method closes and active logging.Logger instance.

    Parameters
    ----------
    log : logging.Logger
        Logging instance

    """

    if verbose:
        print('Closing log file:', log.name)

    # Send closing message.
    log.info('The log file has been closed.')

    # Remove all handlers from log.
    [log.removeHandler(handler) for handler in log.handlers]

Similar but not relevant code:
def stoplog(self):
        """ Stop logging.
    
        @return: 1 on success and 0 on error
        @rtype: integer
        """
        if self._file_logger:
            self.logger.removeHandler(_file_logger)
            self._file_logger = None
        return 1

Similar but not relevant code:
def setupLogFile(self):
		"""Set up the logging file for a new session- include date and some whitespace"""
		self.logWrite("\n###############################################")
		self.logWrite("calcpkg.py log from " + str(datetime.datetime.now()))
		self.changeLogging(True)
Query: how to check column types in python

Relevant code:
def make_kind_check(python_types, numpy_kind):
    """
    Make a function that checks whether a scalar or array is of a given kind
    (e.g. float, int, datetime, timedelta).
    """
    def check(value):
        if hasattr(value, 'dtype'):
            return value.dtype.kind == numpy_kind
        return isinstance(value, python_types)
    return check

Similar but not relevant code:
def columns_equal(a: Column, b: Column) -> bool:
    """
    Are two SQLAlchemy columns are equal? Checks based on:

    - column ``name``
    - column ``type`` (see :func:`column_types_equal`)
    - ``nullable``
    """
    return (
        a.name == b.name and
        column_types_equal(a.type, b.type) and
        a.nullable == b.nullable
    )
Query: how to check content of a file is empty or not in python

Relevant code:
def file_empty(fp):
    """Determine if a file is empty or not."""
    # for python 2 we need to use a homemade peek()
    if six.PY2:
        contents = fp.read()
        fp.seek(0)
        return not bool(contents)

    else:
        return not fp.peek()

Similar but not relevant code:
def allZero(buffer):
    """
    Tries to determine if a buffer is empty.
    
    @type buffer: str
    @param buffer: Buffer to test if it is empty.
        
    @rtype: bool
    @return: C{True} if the given buffer is empty, i.e. full of zeros,
        C{False} if it doesn't.
    """
    allZero = True
    for byte in buffer:
        if byte != "\x00":
            allZero = False
            break
    return allZero

Similar but not relevant code:
def is_non_empty_string(input_string):
    """
    Validate if non empty string

    :param input_string: Input is a *str*.
    :return: True if input is string and non empty.
       Raise :exc:`Exception` otherwise.
    """
    try:
        if not input_string.strip():
            raise ValueError()
    except AttributeError as error:
        raise TypeError(error)

    return True