Skip to main content
`Ctrl`+`K`

# 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 re
    import threading
    from collections.abc import Generator
    from pathlib import Path
    from types import ModuleType

    from ply.lex import Lexer
    from ply.yacc import LRParser

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

    _TAB_HEADER = """# 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()

    @contextlib.contextmanager
    def _patch_ply_module(
        module: ModuleType, file: Path, package: str
    ) -> Generator[None, None, None]:
        """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.

        Additionally, a custom header is inserted into any files ``ply`` writes.
        """
        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)

        file_exists = file.exists() or file.with_suffix(".pyc").exists()
        module.get_caller_module_dict = wrapper
        yield
        module.get_caller_module_dict = original
        if not file_exists:
            file.write_text(_TAB_HEADER.format(package=package) + file.read_text())

    def lex(lextab: str, package: str, reflags: int = int(re.VERBOSE)) -> Lexer:
        """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_dir = Path(lex.get_caller_module_dict(2)["__file__"]).parent
        with _LOCK, _patch_ply_module(lex, caller_dir / (lextab + ".py"), package):
            return lex.lex(
                optimize=True, lextab=lextab, outputdir=caller_dir, reflags=reflags
            )

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

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

        def __init__(self, parser: LRParser) -> None:
            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: str, package: str) -> ThreadSafeParser:
        """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_dir = Path(yacc.get_caller_module_dict(2)["__file__"]).parent
        with _LOCK, _patch_ply_module(yacc, caller_dir / (tabmodule + ".py"), package):
            parser = yacc.yacc(
                tabmodule=tabmodule,
                outputdir=caller_dir,
                debug=False,
                optimize=True,
                write_tables=True,
            )
        return ThreadSafeParser(parser)