Source code for astropy.utils.parsing

# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""
Wrappers for PLY to provide thread safety.
"""

import contextlib
import functools
import os
import re
import threading

__all__ = ["lex", "ThreadSafeParser", "yacc"]


_TAB_HEADER = """# -*- coding: utf-8 -*-
# Licensed under a 3-clause BSD style license - see LICENSE.rst

# This file was automatically generated from ply. To re-generate this file,
# remove it from this folder, then build astropy and run the tests in-place:
#
#   python setup.py build_ext --inplace
#   pytest {package}
#
# You can then commit the changes to this file.

"""
_LOCK = threading.RLock()


def _add_tab_header(filename, package):
    with open(filename) as f:
        contents = f.read()

    with open(filename, "w") as f:
        f.write(_TAB_HEADER.format(package=package))
        f.write(contents)


@contextlib.contextmanager
def _patch_get_caller_module_dict(module):
    """Temporarily replace the module's get_caller_module_dict.

    This is a function inside ``ply.lex`` and ``ply.yacc`` (each has a copy)
    that is used to retrieve the caller's local symbols. Here, we patch the
    function to instead retrieve the grandparent's local symbols to account
    for a wrapper layer.
    """
    original = module.get_caller_module_dict

    @functools.wraps(original)
    def wrapper(levels):
        # Add 2, not 1, because the wrapper itself adds another level
        return original(levels + 2)

    module.get_caller_module_dict = wrapper
    yield
    module.get_caller_module_dict = original


def lex(lextab, package, reflags=int(re.VERBOSE)):
    """Create a lexer from local variables.

    It automatically compiles the lexer in optimized mode, writing to
    ``lextab`` in the same directory as the calling file.

    This function is thread-safe. The returned lexer is *not* thread-safe, but
    if it is used exclusively with a single parser returned by :func:`yacc`
    then it will be safe.

    It is only intended to work with lexers defined within the calling
    function, rather than at class or module scope.

    Parameters
    ----------
    lextab : str
        Name for the file to write with the generated tables, if it does not
        already exist (without ``.py`` suffix).
    package : str
        Name of a test package which should be run with pytest to regenerate
        the output file. This is inserted into a comment in the generated
        file.
    reflags : int
        Passed to ``ply.lex``.
    """
    from ply import lex

    caller_file = lex.get_caller_module_dict(2)["__file__"]
    lextab_filename = os.path.join(os.path.dirname(caller_file), lextab + ".py")
    with _LOCK:
        lextab_exists = os.path.exists(lextab_filename)
        with _patch_get_caller_module_dict(lex):
            lexer = lex.lex(
                optimize=True,
                lextab=lextab,
                outputdir=os.path.dirname(caller_file),
                reflags=reflags,
            )
        if not lextab_exists:
            _add_tab_header(lextab_filename, package)
        return lexer


class ThreadSafeParser:
    """Wrap a parser produced by ``ply.yacc.yacc``.

    It provides a :meth:`parse` method that is thread-safe.
    """

    def __init__(self, parser):
        self.parser = parser
        self._lock = threading.RLock()

    def parse(self, *args, **kwargs):
        """Run the wrapped parser, with a lock to ensure serialization."""
        with self._lock:
            return self.parser.parse(*args, **kwargs)


def yacc(tabmodule, package):
    """Create a parser from local variables.

    It automatically compiles the parser in optimized mode, writing to
    ``tabmodule`` in the same directory as the calling file.

    This function is thread-safe, and the returned parser is also thread-safe,
    provided that it does not share a lexer with any other parser.

    It is only intended to work with parsers defined within the calling
    function, rather than at class or module scope.

    Parameters
    ----------
    tabmodule : str
        Name for the file to write with the generated tables, if it does not
        already exist (without ``.py`` suffix).
    package : str
        Name of a test package which should be run with pytest to regenerate
        the output file. This is inserted into a comment in the generated
        file.
    """
    from ply import yacc

    caller_file = yacc.get_caller_module_dict(2)["__file__"]
    tab_filename = os.path.join(os.path.dirname(caller_file), tabmodule + ".py")
    with _LOCK:
        tab_exists = os.path.exists(tab_filename)
        with _patch_get_caller_module_dict(yacc):
            parser = yacc.yacc(
                tabmodule=tabmodule,
                outputdir=os.path.dirname(caller_file),
                debug=False,
                optimize=True,
                write_tables=True,
            )
        if not tab_exists:
            _add_tab_header(tab_filename, package)

    return ThreadSafeParser(parser)