Source code for astropy.utils.parsing

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

from __future__ import annotations

import contextlib
import functools
import re
import threading
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Generator
    from types import ModuleType

    from astropy.extern.ply.lex import Lexer
    from astropy.extern.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())


[docs] 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 astropy.extern.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 )
[docs] 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()
[docs] def parse(self, *args, **kwargs): """Run the wrapped parser, with a lock to ensure serialization.""" with self._lock: return self.parser.parse(*args, **kwargs)
[docs] 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 astropy.extern.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)