Team Ai
Datasetpublic

codekingpro/portable-devtools

sourceHugging Faceupdated 5mo agoView on Hugging Face
1likes14kdownloads
compiler.py1999 linesDownload Raw Back to jinja2
1"""Compiles nodes from the parser into Python code."""2 3import typing as t4from contextlib import contextmanager5from functools import update_wrapper6from io import StringIO7from itertools import chain8from keyword import iskeyword as is_python_keyword9 10from markupsafe import escape11from markupsafe import Markup12 13from . import nodes14from .exceptions import TemplateAssertionError15from .idtracking import Symbols16from .idtracking import VAR_LOAD_ALIAS17from .idtracking import VAR_LOAD_PARAMETER18from .idtracking import VAR_LOAD_RESOLVE19from .idtracking import VAR_LOAD_UNDEFINED20from .nodes import EvalContext21from .optimizer import Optimizer22from .utils import _PassArg23from .utils import concat24from .visitor import NodeVisitor25 26if t.TYPE_CHECKING:27    import typing_extensions as te28 29    from .environment import Environment30 31F = t.TypeVar("F", bound=t.Callable[..., t.Any])32 33operators = {34    "eq": "==",35    "ne": "!=",36    "gt": ">",37    "gteq": ">=",38    "lt": "<",39    "lteq": "<=",40    "in": "in",41    "notin": "not in",42}43 44 45def optimizeconst(f: F) -> F:46    def new_func(47        self: "CodeGenerator", node: nodes.Expr, frame: "Frame", **kwargs: t.Any48    ) -> t.Any:49        # Only optimize if the frame is not volatile50        if self.optimizer is not None and not frame.eval_ctx.volatile:51            new_node = self.optimizer.visit(node, frame.eval_ctx)52 53            if new_node != node:54                return self.visit(new_node, frame)55 56        return f(self, node, frame, **kwargs)57 58    return update_wrapper(new_func, f)  # type: ignore[return-value]59 60 61def _make_binop(op: str) -> t.Callable[["CodeGenerator", nodes.BinExpr, "Frame"], None]:62    @optimizeconst63    def visitor(self: "CodeGenerator", node: nodes.BinExpr, frame: Frame) -> None:64        if (65            self.environment.sandboxed and op in self.environment.intercepted_binops  # type: ignore66        ):67            self.write(f"environment.call_binop(context, {op!r}, ")68            self.visit(node.left, frame)69            self.write(", ")70            self.visit(node.right, frame)71        else:72            self.write("(")73            self.visit(node.left, frame)74            self.write(f" {op} ")75            self.visit(node.right, frame)76 77        self.write(")")78 79    return visitor80 81 82def _make_unop(83    op: str,84) -> t.Callable[["CodeGenerator", nodes.UnaryExpr, "Frame"], None]:85    @optimizeconst86    def visitor(self: "CodeGenerator", node: nodes.UnaryExpr, frame: Frame) -> None:87        if (88            self.environment.sandboxed and op in self.environment.intercepted_unops  # type: ignore89        ):90            self.write(f"environment.call_unop(context, {op!r}, ")91            self.visit(node.node, frame)92        else:93            self.write("(" + op)94            self.visit(node.node, frame)95 96        self.write(")")97 98    return visitor99 100 101def generate(102    node: nodes.Template,103    environment: "Environment",104    name: t.Optional[str],105    filename: t.Optional[str],106    stream: t.Optional[t.TextIO] = None,107    defer_init: bool = False,108    optimized: bool = True,109) -> t.Optional[str]:110    """Generate the python source for a node tree."""111    if not isinstance(node, nodes.Template):112        raise TypeError("Can't compile non template nodes")113 114    generator = environment.code_generator_class(115        environment, name, filename, stream, defer_init, optimized116    )117    generator.visit(node)118 119    if stream is None:120        return generator.stream.getvalue()  # type: ignore121 122    return None123 124 125def has_safe_repr(value: t.Any) -> bool:126    """Does the node have a safe representation?"""127    if value is None or value is NotImplemented or value is Ellipsis:128        return True129 130    if type(value) in {bool, int, float, complex, range, str, Markup}:131        return True132 133    if type(value) in {tuple, list, set, frozenset}:134        return all(has_safe_repr(v) for v in value)135 136    if type(value) is dict:  # noqa E721137        return all(has_safe_repr(k) and has_safe_repr(v) for k, v in value.items())138 139    return False140 141 142def find_undeclared(143    nodes: t.Iterable[nodes.Node], names: t.Iterable[str]144) -> t.Set[str]:145    """Check if the names passed are accessed undeclared.  The return value146    is a set of all the undeclared names from the sequence of names found.147    """148    visitor = UndeclaredNameVisitor(names)149    try:150        for node in nodes:151            visitor.visit(node)152    except VisitorExit:153        pass154    return visitor.undeclared155 156 157class MacroRef:158    def __init__(self, node: t.Union[nodes.Macro, nodes.CallBlock]) -> None:159        self.node = node160        self.accesses_caller = False161        self.accesses_kwargs = False162        self.accesses_varargs = False163 164 165class Frame:166    """Holds compile time information for us."""167 168    def __init__(169        self,170        eval_ctx: EvalContext,171        parent: t.Optional["Frame"] = None,172        level: t.Optional[int] = None,173    ) -> None:174        self.eval_ctx = eval_ctx175 176        # the parent of this frame177        self.parent = parent178 179        if parent is None:180            self.symbols = Symbols(level=level)181 182            # in some dynamic inheritance situations the compiler needs to add183            # write tests around output statements.184            self.require_output_check = False185 186            # inside some tags we are using a buffer rather than yield statements.187            # this for example affects {% filter %} or {% macro %}.  If a frame188            # is buffered this variable points to the name of the list used as189            # buffer.190            self.buffer: t.Optional[str] = None191 192            # the name of the block we're in, otherwise None.193            self.block: t.Optional[str] = None194 195        else:196            self.symbols = Symbols(parent.symbols, level=level)197            self.require_output_check = parent.require_output_check198            self.buffer = parent.buffer199            self.block = parent.block200 201        # a toplevel frame is the root + soft frames such as if conditions.202        self.toplevel = False203 204        # the root frame is basically just the outermost frame, so no if205        # conditions.  This information is used to optimize inheritance206        # situations.207        self.rootlevel = False208 209        # variables set inside of loops and blocks should not affect outer frames,210        # but they still needs to be kept track of as part of the active context.211        self.loop_frame = False212        self.block_frame = False213 214        # track whether the frame is being used in an if-statement or conditional215        # expression as it determines which errors should be raised during runtime216        # or compile time.217        self.soft_frame = False218 219    def copy(self) -> "te.Self":220        """Create a copy of the current one."""221        rv = object.__new__(self.__class__)222        rv.__dict__.update(self.__dict__)223        rv.symbols = self.symbols.copy()224        return rv225 226    def inner(self, isolated: bool = False) -> "Frame":227        """Return an inner frame."""228        if isolated:229            return Frame(self.eval_ctx, level=self.symbols.level + 1)230        return Frame(self.eval_ctx, self)231 232    def soft(self) -> "te.Self":233        """Return a soft frame.  A soft frame may not be modified as234        standalone thing as it shares the resources with the frame it235        was created of, but it's not a rootlevel frame any longer.236 237        This is only used to implement if-statements and conditional238        expressions.239        """240        rv = self.copy()241        rv.rootlevel = False242        rv.soft_frame = True243        return rv244 245    __copy__ = copy246 247 248class VisitorExit(RuntimeError):249    """Exception used by the `UndeclaredNameVisitor` to signal a stop."""250 251 252class DependencyFinderVisitor(NodeVisitor):253    """A visitor that collects filter and test calls."""254 255    def __init__(self) -> None:256        self.filters: t.Set[str] = set()257        self.tests: t.Set[str] = set()258 259    def visit_Filter(self, node: nodes.Filter) -> None:260        self.generic_visit(node)261        self.filters.add(node.name)262 263    def visit_Test(self, node: nodes.Test) -> None:264        self.generic_visit(node)265        self.tests.add(node.name)266 267    def visit_Block(self, node: nodes.Block) -> None:268        """Stop visiting at blocks."""269 270 271class UndeclaredNameVisitor(NodeVisitor):272    """A visitor that checks if a name is accessed without being273    declared.  This is different from the frame visitor as it will274    not stop at closure frames.275    """276 277    def __init__(self, names: t.Iterable[str]) -> None:278        self.names = set(names)279        self.undeclared: t.Set[str] = set()280 281    def visit_Name(self, node: nodes.Name) -> None:282        if node.ctx == "load" and node.name in self.names:283            self.undeclared.add(node.name)284            if self.undeclared == self.names:285                raise VisitorExit()286        else:287            self.names.discard(node.name)288 289    def visit_Block(self, node: nodes.Block) -> None:290        """Stop visiting a blocks."""291 292 293class CompilerExit(Exception):294    """Raised if the compiler encountered a situation where it just295    doesn't make sense to further process the code.  Any block that296    raises such an exception is not further processed.297    """298 299 300class CodeGenerator(NodeVisitor):301    def __init__(302        self,303        environment: "Environment",304        name: t.Optional[str],305        filename: t.Optional[str],306        stream: t.Optional[t.TextIO] = None,307        defer_init: bool = False,308        optimized: bool = True,309    ) -> None:310        if stream is None:311            stream = StringIO()312        self.environment = environment313        self.name = name314        self.filename = filename315        self.stream = stream316        self.created_block_context = False317        self.defer_init = defer_init318        self.optimizer: t.Optional[Optimizer] = None319 320        if optimized:321            self.optimizer = Optimizer(environment)322 323        # aliases for imports324        self.import_aliases: t.Dict[str, str] = {}325 326        # a registry for all blocks.  Because blocks are moved out327        # into the global python scope they are registered here328        self.blocks: t.Dict[str, nodes.Block] = {}329 330        # the number of extends statements so far331        self.extends_so_far = 0332 333        # some templates have a rootlevel extends.  In this case we334        # can safely assume that we're a child template and do some335        # more optimizations.336        self.has_known_extends = False337 338        # the current line number339        self.code_lineno = 1340 341        # registry of all filters and tests (global, not block local)342        self.tests: t.Dict[str, str] = {}343        self.filters: t.Dict[str, str] = {}344 345        # the debug information346        self.debug_info: t.List[t.Tuple[int, int]] = []347        self._write_debug_info: t.Optional[int] = None348 349        # the number of new lines before the next write()350        self._new_lines = 0351 352        # the line number of the last written statement353        self._last_line = 0354 355        # true if nothing was written so far.356        self._first_write = True357 358        # used by the `temporary_identifier` method to get new359        # unique, temporary identifier360        self._last_identifier = 0361 362        # the current indentation363        self._indentation = 0364 365        # Tracks toplevel assignments366        self._assign_stack: t.List[t.Set[str]] = []367 368        # Tracks parameter definition blocks369        self._param_def_block: t.List[t.Set[str]] = []370 371        # Tracks the current context.372        self._context_reference_stack = ["context"]373 374    @property375    def optimized(self) -> bool:376        return self.optimizer is not None377 378    # -- Various compilation helpers379 380    def fail(self, msg: str, lineno: int) -> "te.NoReturn":381        """Fail with a :exc:`TemplateAssertionError`."""382        raise TemplateAssertionError(msg, lineno, self.name, self.filename)383 384    def temporary_identifier(self) -> str:385        """Get a new unique identifier."""386        self._last_identifier += 1387        return f"t_{self._last_identifier}"388 389    def buffer(self, frame: Frame) -> None:390        """Enable buffering for the frame from that point onwards."""391        frame.buffer = self.temporary_identifier()392        self.writeline(f"{frame.buffer} = []")393 394    def return_buffer_contents(395        self, frame: Frame, force_unescaped: bool = False396    ) -> None:397        """Return the buffer contents of the frame."""398        if not force_unescaped:399            if frame.eval_ctx.volatile:400                self.writeline("if context.eval_ctx.autoescape:")401                self.indent()402                self.writeline(f"return Markup(concat({frame.buffer}))")403                self.outdent()404                self.writeline("else:")405                self.indent()406                self.writeline(f"return concat({frame.buffer})")407                self.outdent()408                return409            elif frame.eval_ctx.autoescape:410                self.writeline(f"return Markup(concat({frame.buffer}))")411                return412        self.writeline(f"return concat({frame.buffer})")413 414    def indent(self) -> None:415        """Indent by one."""416        self._indentation += 1417 418    def outdent(self, step: int = 1) -> None:419        """Outdent by step."""420        self._indentation -= step421 422    def start_write(self, frame: Frame, node: t.Optional[nodes.Node] = None) -> None:423        """Yield or write into the frame buffer."""424        if frame.buffer is None:425            self.writeline("yield ", node)426        else:427            self.writeline(f"{frame.buffer}.append(", node)428 429    def end_write(self, frame: Frame) -> None:430        """End the writing process started by `start_write`."""431        if frame.buffer is not None:432            self.write(")")433 434    def simple_write(435        self, s: str, frame: Frame, node: t.Optional[nodes.Node] = None436    ) -> None:437        """Simple shortcut for start_write + write + end_write."""438        self.start_write(frame, node)439        self.write(s)440        self.end_write(frame)441 442    def blockvisit(self, nodes: t.Iterable[nodes.Node], frame: Frame) -> None:443        """Visit a list of nodes as block in a frame.  If the current frame444        is no buffer a dummy ``if 0: yield None`` is written automatically.445        """446        try:447            self.writeline("pass")448            for node in nodes:449                self.visit(node, frame)450        except CompilerExit:451            pass452 453    def write(self, x: str) -> None:454        """Write a string into the output stream."""455        if self._new_lines:456            if not self._first_write:457                self.stream.write("\n" * self._new_lines)458                self.code_lineno += self._new_lines459                if self._write_debug_info is not None:460                    self.debug_info.append((self._write_debug_info, self.code_lineno))461                    self._write_debug_info = None462            self._first_write = False463            self.stream.write("    " * self._indentation)464            self._new_lines = 0465        self.stream.write(x)466 467    def writeline(468        self, x: str, node: t.Optional[nodes.Node] = None, extra: int = 0469    ) -> None:470        """Combination of newline and write."""471        self.newline(node, extra)472        self.write(x)473 474    def newline(self, node: t.Optional[nodes.Node] = None, extra: int = 0) -> None:475        """Add one or more newlines before the next write."""476        self._new_lines = max(self._new_lines, 1 + extra)477        if node is not None and node.lineno != self._last_line:478            self._write_debug_info = node.lineno479            self._last_line = node.lineno480 481    def signature(482        self,483        node: t.Union[nodes.Call, nodes.Filter, nodes.Test],484        frame: Frame,485        extra_kwargs: t.Optional[t.Mapping[str, t.Any]] = None,486    ) -> None:487        """Writes a function call to the stream for the current node.488        A leading comma is added automatically.  The extra keyword489        arguments may not include python keywords otherwise a syntax490        error could occur.  The extra keyword arguments should be given491        as python dict.492        """493        # if any of the given keyword arguments is a python keyword494        # we have to make sure that no invalid call is created.495        kwarg_workaround = any(496            is_python_keyword(t.cast(str, k))497            for k in chain((x.key for x in node.kwargs), extra_kwargs or ())498        )499 500        for arg in node.args:501            self.write(", ")502            self.visit(arg, frame)503 504        if not kwarg_workaround:505            for kwarg in node.kwargs:506                self.write(", ")507                self.visit(kwarg, frame)508            if extra_kwargs is not None:509                for key, value in extra_kwargs.items():510                    self.write(f", {key}={value}")511        if node.dyn_args:512            self.write(", *")513            self.visit(node.dyn_args, frame)514 515        if kwarg_workaround:516            if node.dyn_kwargs is not None:517                self.write(", **dict({")518            else:519                self.write(", **{")520            for kwarg in node.kwargs:521                self.write(f"{kwarg.key!r}: ")522                self.visit(kwarg.value, frame)523                self.write(", ")524            if extra_kwargs is not None:525                for key, value in extra_kwargs.items():526                    self.write(f"{key!r}: {value}, ")527            if node.dyn_kwargs is not None:528                self.write("}, **")529                self.visit(node.dyn_kwargs, frame)530                self.write(")")531            else:532                self.write("}")533 534        elif node.dyn_kwargs is not None:535            self.write(", **")536            self.visit(node.dyn_kwargs, frame)537 538    def pull_dependencies(self, nodes: t.Iterable[nodes.Node]) -> None:539        """Find all filter and test names used in the template and540        assign them to variables in the compiled namespace. Checking541        that the names are registered with the environment is done when542        compiling the Filter and Test nodes. If the node is in an If or543        CondExpr node, the check is done at runtime instead.544 545        .. versionchanged:: 3.0546            Filters and tests in If and CondExpr nodes are checked at547            runtime instead of compile time.548        """549        visitor = DependencyFinderVisitor()550 551        for node in nodes:552            visitor.visit(node)553 554        for id_map, names, dependency in (555            (self.filters, visitor.filters, "filters"),556            (557                self.tests,558                visitor.tests,559                "tests",560            ),561        ):562            for name in sorted(names):563                if name not in id_map:564                    id_map[name] = self.temporary_identifier()565 566                # add check during runtime that dependencies used inside of executed567                # blocks are defined, as this step may be skipped during compile time568                self.writeline("try:")569                self.indent()570                self.writeline(f"{id_map[name]} = environment.{dependency}[{name!r}]")571                self.outdent()572                self.writeline("except KeyError:")573                self.indent()574                self.writeline("@internalcode")575                self.writeline(f"def {id_map[name]}(*unused):")576                self.indent()577                self.writeline(578                    f'raise TemplateRuntimeError("No {dependency[:-1]}'579                    f' named {name!r} found.")'580                )581                self.outdent()582                self.outdent()583 584    def enter_frame(self, frame: Frame) -> None:585        undefs = []586        for target, (action, param) in frame.symbols.loads.items():587            if action == VAR_LOAD_PARAMETER:588                pass589            elif action == VAR_LOAD_RESOLVE:590                self.writeline(f"{target} = {self.get_resolve_func()}({param!r})")591            elif action == VAR_LOAD_ALIAS:592                self.writeline(f"{target} = {param}")593            elif action == VAR_LOAD_UNDEFINED:594                undefs.append(target)595            else:596                raise NotImplementedError("unknown load instruction")597        if undefs:598            self.writeline(f"{' = '.join(undefs)} = missing")599 600    def leave_frame(self, frame: Frame, with_python_scope: bool = False) -> None:601        if not with_python_scope:602            undefs = []603            for target in frame.symbols.loads:604                undefs.append(target)605            if undefs:606                self.writeline(f"{' = '.join(undefs)} = missing")607 608    def choose_async(self, async_value: str = "async ", sync_value: str = "") -> str:609        return async_value if self.environment.is_async else sync_value610 611    def func(self, name: str) -> str:612        return f"{self.choose_async()}def {name}"613 614    def macro_body(615        self, node: t.Union[nodes.Macro, nodes.CallBlock], frame: Frame616    ) -> t.Tuple[Frame, MacroRef]:617        """Dump the function def of a macro or call block."""618        frame = frame.inner()619        frame.symbols.analyze_node(node)620        macro_ref = MacroRef(node)621 622        explicit_caller = None623        skip_special_params = set()624        args = []625 626        for idx, arg in enumerate(node.args):627            if arg.name == "caller":628                explicit_caller = idx629            if arg.name in ("kwargs", "varargs"):630                skip_special_params.add(arg.name)631            args.append(frame.symbols.ref(arg.name))632 633        undeclared = find_undeclared(node.body, ("caller", "kwargs", "varargs"))634 635        if "caller" in undeclared:636            # In older Jinja versions there was a bug that allowed caller637            # to retain the special behavior even if it was mentioned in638            # the argument list.  However thankfully this was only really639            # working if it was the last argument.  So we are explicitly640            # checking this now and error out if it is anywhere else in641            # the argument list.642            if explicit_caller is not None:643                try:644                    node.defaults[explicit_caller - len(node.args)]645                except IndexError:646                    self.fail(647                        "When defining macros or call blocks the "648                        'special "caller" argument must be omitted '649                        "or be given a default.",650                        node.lineno,651                    )652            else:653                args.append(frame.symbols.declare_parameter("caller"))654            macro_ref.accesses_caller = True655        if "kwargs" in undeclared and "kwargs" not in skip_special_params:656            args.append(frame.symbols.declare_parameter("kwargs"))657            macro_ref.accesses_kwargs = True658        if "varargs" in undeclared and "varargs" not in skip_special_params:659            args.append(frame.symbols.declare_parameter("varargs"))660            macro_ref.accesses_varargs = True661 662        # macros are delayed, they never require output checks663        frame.require_output_check = False664        frame.symbols.analyze_node(node)665        self.writeline(f"{self.func('macro')}({', '.join(args)}):", node)666        self.indent()667 668        self.buffer(frame)669        self.enter_frame(frame)670 671        self.push_parameter_definitions(frame)672        for idx, arg in enumerate(node.args):673            ref = frame.symbols.ref(arg.name)674            self.writeline(f"if {ref} is missing:")675            self.indent()676            try:677                default = node.defaults[idx - len(node.args)]678            except IndexError:679                self.writeline(680                    f'{ref} = undefined("parameter {arg.name!r} was not provided",'681                    f" name={arg.name!r})"682                )683            else:684                self.writeline(f"{ref} = ")685                self.visit(default, frame)686            self.mark_parameter_stored(ref)687            self.outdent()688        self.pop_parameter_definitions()689 690        self.blockvisit(node.body, frame)691        self.return_buffer_contents(frame, force_unescaped=True)692        self.leave_frame(frame, with_python_scope=True)693        self.outdent()694 695        return frame, macro_ref696 697    def macro_def(self, macro_ref: MacroRef, frame: Frame) -> None:698        """Dump the macro definition for the def created by macro_body."""699        arg_tuple = ", ".join(repr(x.name) for x in macro_ref.node.args)700        name = getattr(macro_ref.node, "name", None)701        if len(macro_ref.node.args) == 1:702            arg_tuple += ","703        self.write(704            f"Macro(environment, macro, {name!r}, ({arg_tuple}),"705            f" {macro_ref.accesses_kwargs!r}, {macro_ref.accesses_varargs!r},"706            f" {macro_ref.accesses_caller!r}, context.eval_ctx.autoescape)"707        )708 709    def position(self, node: nodes.Node) -> str:710        """Return a human readable position for the node."""711        rv = f"line {node.lineno}"712        if self.name is not None:713            rv = f"{rv} in {self.name!r}"714        return rv715 716    def dump_local_context(self, frame: Frame) -> str:717        items_kv = ", ".join(718            f"{name!r}: {target}"719            for name, target in frame.symbols.dump_stores().items()720        )721        return f"{{{items_kv}}}"722 723    def write_commons(self) -> None:724        """Writes a common preamble that is used by root and block functions.725        Primarily this sets up common local helpers and enforces a generator726        through a dead branch.727        """728        self.writeline("resolve = context.resolve_or_missing")729        self.writeline("undefined = environment.undefined")730        self.writeline("concat = environment.concat")731        # always use the standard Undefined class for the implicit else of732        # conditional expressions733        self.writeline("cond_expr_undefined = Undefined")734        self.writeline("if 0: yield None")735 736    def push_parameter_definitions(self, frame: Frame) -> None:737        """Pushes all parameter targets from the given frame into a local738        stack that permits tracking of yet to be assigned parameters.  In739        particular this enables the optimization from `visit_Name` to skip740        undefined expressions for parameters in macros as macros can reference741        otherwise unbound parameters.742        """743        self._param_def_block.append(frame.symbols.dump_param_targets())744 745    def pop_parameter_definitions(self) -> None:746        """Pops the current parameter definitions set."""747        self._param_def_block.pop()748 749    def mark_parameter_stored(self, target: str) -> None:750        """Marks a parameter in the current parameter definitions as stored.751        This will skip the enforced undefined checks.752        """753        if self._param_def_block:754            self._param_def_block[-1].discard(target)755 756    def push_context_reference(self, target: str) -> None:757        self._context_reference_stack.append(target)758 759    def pop_context_reference(self) -> None:760        self._context_reference_stack.pop()761 762    def get_context_ref(self) -> str:763        return self._context_reference_stack[-1]764 765    def get_resolve_func(self) -> str:766        target = self._context_reference_stack[-1]767        if target == "context":768            return "resolve"769        return f"{target}.resolve"770 771    def derive_context(self, frame: Frame) -> str:772        return f"{self.get_context_ref()}.derived({self.dump_local_context(frame)})"773 774    def parameter_is_undeclared(self, target: str) -> bool:775        """Checks if a given target is an undeclared parameter."""776        if not self._param_def_block:777            return False778        return target in self._param_def_block[-1]779 780    def push_assign_tracking(self) -> None:781        """Pushes a new layer for assignment tracking."""782        self._assign_stack.append(set())783 784    def pop_assign_tracking(self, frame: Frame) -> None:785        """Pops the topmost level for assignment tracking and updates the786        context variables if necessary.787        """788        vars = self._assign_stack.pop()789        if (790            not frame.block_frame791            and not frame.loop_frame792            and not frame.toplevel793            or not vars794        ):795            return796        public_names = [x for x in vars if x[:1] != "_"]797        if len(vars) == 1:798            name = next(iter(vars))799            ref = frame.symbols.ref(name)800            if frame.loop_frame:801                self.writeline(f"_loop_vars[{name!r}] = {ref}")802                return803            if frame.block_frame:804                self.writeline(f"_block_vars[{name!r}] = {ref}")805                return806            self.writeline(f"context.vars[{name!r}] = {ref}")807        else:808            if frame.loop_frame:809                self.writeline("_loop_vars.update({")810            elif frame.block_frame:811                self.writeline("_block_vars.update({")812            else:813                self.writeline("context.vars.update({")814            for idx, name in enumerate(sorted(vars)):815                if idx:816                    self.write(", ")817                ref = frame.symbols.ref(name)818                self.write(f"{name!r}: {ref}")819            self.write("})")820        if not frame.block_frame and not frame.loop_frame and public_names:821            if len(public_names) == 1:822                self.writeline(f"context.exported_vars.add({public_names[0]!r})")823            else:824                names_str = ", ".join(map(repr, sorted(public_names)))825                self.writeline(f"context.exported_vars.update(({names_str}))")826 827    # -- Statement Visitors828 829    def visit_Template(830        self, node: nodes.Template, frame: t.Optional[Frame] = None831    ) -> None:832        assert frame is None, "no root frame allowed"833        eval_ctx = EvalContext(self.environment, self.name)834 835        from .runtime import async_exported836        from .runtime import exported837 838        if self.environment.is_async:839            exported_names = sorted(exported + async_exported)840        else:841            exported_names = sorted(exported)842 843        self.writeline("from jinja2.runtime import " + ", ".join(exported_names))844 845        # if we want a deferred initialization we cannot move the846        # environment into a local name847        envenv = "" if self.defer_init else ", environment=environment"848 849        # do we have an extends tag at all?  If not, we can save some850        # overhead by just not processing any inheritance code.851        have_extends = node.find(nodes.Extends) is not None852 853        # find all blocks854        for block in node.find_all(nodes.Block):855            if block.name in self.blocks:856                self.fail(f"block {block.name!r} defined twice", block.lineno)857            self.blocks[block.name] = block858 859        # find all imports and import them860        for import_ in node.find_all(nodes.ImportedName):861            if import_.importname not in self.import_aliases:862                imp = import_.importname863                self.import_aliases[imp] = alias = self.temporary_identifier()864                if "." in imp:865                    module, obj = imp.rsplit(".", 1)866                    self.writeline(f"from {module} import {obj} as {alias}")867                else:868                    self.writeline(f"import {imp} as {alias}")869 870        # add the load name871        self.writeline(f"name = {self.name!r}")872 873        # generate the root render function.874        self.writeline(875            f"{self.func('root')}(context, missing=missing{envenv}):", extra=1876        )877        self.indent()878        self.write_commons()879 880        # process the root881        frame = Frame(eval_ctx)882        if "self" in find_undeclared(node.body, ("self",)):883            ref = frame.symbols.declare_parameter("self")884            self.writeline(f"{ref} = TemplateReference(context)")885        frame.symbols.analyze_node(node)886        frame.toplevel = frame.rootlevel = True887        frame.require_output_check = have_extends and not self.has_known_extends888        if have_extends:889            self.writeline("parent_template = None")890        self.enter_frame(frame)891        self.pull_dependencies(node.body)892        self.blockvisit(node.body, frame)893        self.leave_frame(frame, with_python_scope=True)894        self.outdent()895 896        # make sure that the parent root is called.897        if have_extends:898            if not self.has_known_extends:899                self.indent()900                self.writeline("if parent_template is not None:")901            self.indent()902            if not self.environment.is_async:903                self.writeline("yield from parent_template.root_render_func(context)")904            else:905                self.writeline("agen = parent_template.root_render_func(context)")906                self.writeline("try:")907                self.indent()908                self.writeline("async for event in agen:")909                self.indent()910                self.writeline("yield event")911                self.outdent()912                self.outdent()913                self.writeline("finally: await agen.aclose()")914            self.outdent(1 + (not self.has_known_extends))915 916        # at this point we now have the blocks collected and can visit them too.917        for name, block in self.blocks.items():918            self.writeline(919                f"{self.func('block_' + name)}(context, missing=missing{envenv}):",920                block,921                1,922            )923            self.indent()924            self.write_commons()925            # It's important that we do not make this frame a child of the926            # toplevel template.  This would cause a variety of927            # interesting issues with identifier tracking.928            block_frame = Frame(eval_ctx)929            block_frame.block_frame = True930            undeclared = find_undeclared(block.body, ("self", "super"))931            if "self" in undeclared:932                ref = block_frame.symbols.declare_parameter("self")933                self.writeline(f"{ref} = TemplateReference(context)")934            if "super" in undeclared:935                ref = block_frame.symbols.declare_parameter("super")936                self.writeline(f"{ref} = context.super({name!r}, block_{name})")937            block_frame.symbols.analyze_node(block)938            block_frame.block = name939            self.writeline("_block_vars = {}")940            self.enter_frame(block_frame)941            self.pull_dependencies(block.body)942            self.blockvisit(block.body, block_frame)943            self.leave_frame(block_frame, with_python_scope=True)944            self.outdent()945 946        blocks_kv_str = ", ".join(f"{x!r}: block_{x}" for x in self.blocks)947        self.writeline(f"blocks = {{{blocks_kv_str}}}", extra=1)948        debug_kv_str = "&".join(f"{k}={v}" for k, v in self.debug_info)949        self.writeline(f"debug_info = {debug_kv_str!r}")950 951    def visit_Block(self, node: nodes.Block, frame: Frame) -> None:952        """Call a block and register it for the template."""953        level = 0954        if frame.toplevel:955            # if we know that we are a child template, there is no need to956            # check if we are one957            if self.has_known_extends:958                return959            if self.extends_so_far > 0:960                self.writeline("if parent_template is None:")961                self.indent()962                level += 1963 964        if node.scoped:965            context = self.derive_context(frame)966        else:967            context = self.get_context_ref()968 969        if node.required:970            self.writeline(f"if len(context.blocks[{node.name!r}]) <= 1:", node)971            self.indent()972            self.writeline(973                f'raise TemplateRuntimeError("Required block {node.name!r} not found")',974                node,975            )976            self.outdent()977 978        if not self.environment.is_async and frame.buffer is None:979            self.writeline(980                f"yield from context.blocks[{node.name!r}][0]({context})", node981            )982        else:983            self.writeline(f"gen = context.blocks[{node.name!r}][0]({context})")984            self.writeline("try:")985            self.indent()986            self.writeline(987                f"{self.choose_async()}for event in gen:",988                node,989            )990            self.indent()991            self.simple_write("event", frame)992            self.outdent()993            self.outdent()994            self.writeline(995                f"finally: {self.choose_async('await gen.aclose()', 'gen.close()')}"996            )997 998        self.outdent(level)999 1000    def visit_Extends(self, node: nodes.Extends, frame: Frame) -> None:1001        """Calls the extender."""1002        if not frame.toplevel:1003            self.fail("cannot use extend from a non top-level scope", node.lineno)1004 1005        # if the number of extends statements in general is zero so1006        # far, we don't have to add a check if something extended1007        # the template before this one.1008        if self.extends_so_far > 0:1009            # if we have a known extends we just add a template runtime1010            # error into the generated code.  We could catch that at compile1011            # time too, but i welcome it not to confuse users by throwing the1012            # same error at different times just "because we can".1013            if not self.has_known_extends:1014                self.writeline("if parent_template is not None:")1015                self.indent()1016            self.writeline('raise TemplateRuntimeError("extended multiple times")')1017 1018            # if we have a known extends already we don't need that code here1019            # as we know that the template execution will end here.1020            if self.has_known_extends:1021                raise CompilerExit()1022            else:1023                self.outdent()1024 1025        self.writeline("parent_template = environment.get_template(", node)1026        self.visit(node.template, frame)1027        self.write(f", {self.name!r})")1028        self.writeline("for name, parent_block in parent_template.blocks.items():")1029        self.indent()1030        self.writeline("context.blocks.setdefault(name, []).append(parent_block)")1031        self.outdent()1032 1033        # if this extends statement was in the root level we can take1034        # advantage of that information and simplify the generated code1035        # in the top level from this point onwards1036        if frame.rootlevel:1037            self.has_known_extends = True1038 1039        # and now we have one more1040        self.extends_so_far += 11041 1042    def visit_Include(self, node: nodes.Include, frame: Frame) -> None:1043        """Handles includes."""1044        if node.ignore_missing:1045            self.writeline("try:")1046            self.indent()1047 1048        func_name = "get_or_select_template"1049        if isinstance(node.template, nodes.Const):1050            if isinstance(node.template.value, str):1051                func_name = "get_template"1052            elif isinstance(node.template.value, (tuple, list)):1053                func_name = "select_template"1054        elif isinstance(node.template, (nodes.Tuple, nodes.List)):1055            func_name = "select_template"1056 1057        self.writeline(f"template = environment.{func_name}(", node)1058        self.visit(node.template, frame)1059        self.write(f", {self.name!r})")1060        if node.ignore_missing:1061            self.outdent()1062            self.writeline("except TemplateNotFound:")1063            self.indent()1064            self.writeline("pass")1065            self.outdent()1066            self.writeline("else:")1067            self.indent()1068 1069        def loop_body() -> None:1070            self.indent()1071            self.simple_write("event", frame)1072            self.outdent()1073 1074        if node.with_context:1075            self.writeline(1076                f"gen = template.root_render_func("1077                "template.new_context(context.get_all(), True,"1078                f" {self.dump_local_context(frame)}))"1079            )1080            self.writeline("try:")1081            self.indent()1082            self.writeline(f"{self.choose_async()}for event in gen:")1083            loop_body()1084            self.outdent()1085            self.writeline(1086                f"finally: {self.choose_async('await gen.aclose()', 'gen.close()')}"1087            )1088        elif self.environment.is_async:1089            self.writeline(1090                "for event in (await template._get_default_module_async())"1091                "._body_stream:"1092            )1093            loop_body()1094        else:1095            self.writeline("yield from template._get_default_module()._body_stream")1096 1097        if node.ignore_missing:1098            self.outdent()1099 1100    def _import_common(1101        self, node: t.Union[nodes.Import, nodes.FromImport], frame: Frame1102    ) -> None:1103        self.write(f"{self.choose_async('await ')}environment.get_template(")1104        self.visit(node.template, frame)1105        self.write(f", {self.name!r}).")1106 1107        if node.with_context:1108            f_name = f"make_module{self.choose_async('_async')}"1109            self.write(1110                f"{f_name}(context.get_all(), True, {self.dump_local_context(frame)})"1111            )1112        else:1113            self.write(f"_get_default_module{self.choose_async('_async')}(context)")1114 1115    def visit_Import(self, node: nodes.Import, frame: Frame) -> None:1116        """Visit regular imports."""1117        self.writeline(f"{frame.symbols.ref(node.target)} = ", node)1118        if frame.toplevel:1119            self.write(f"context.vars[{node.target!r}] = ")1120 1121        self._import_common(node, frame)1122 1123        if frame.toplevel and not node.target.startswith("_"):1124            self.writeline(f"context.exported_vars.discard({node.target!r})")1125 1126    def visit_FromImport(self, node: nodes.FromImport, frame: Frame) -> None:1127        """Visit named imports."""1128        self.newline(node)1129        self.write("included_template = ")1130        self._import_common(node, frame)1131        var_names = []1132        discarded_names = []1133        for name in node.names:1134            if isinstance(name, tuple):1135                name, alias = name1136            else:1137                alias = name1138            self.writeline(1139                f"{frame.symbols.ref(alias)} ="1140                f" getattr(included_template, {name!r}, missing)"1141            )1142            self.writeline(f"if {frame.symbols.ref(alias)} is missing:")1143            self.indent()1144            # The position will contain the template name, and will be formatted1145            # into a string that will be compiled into an f-string. Curly braces1146            # in the name must be replaced with escapes so that they will not be1147            # executed as part of the f-string.1148            position = self.position(node).replace("{", "{{").replace("}", "}}")1149            message = (1150                "the template {included_template.__name__!r}"1151                f" (imported on {position})"1152                f" does not export the requested name {name!r}"1153            )1154            self.writeline(1155                f"{frame.symbols.ref(alias)} = undefined(f{message!r}, name={name!r})"1156            )1157            self.outdent()1158            if frame.toplevel:1159                var_names.append(alias)1160                if not alias.startswith("_"):1161                    discarded_names.append(alias)1162 1163        if var_names:1164            if len(var_names) == 1:1165                name = var_names[0]1166                self.writeline(f"context.vars[{name!r}] = {frame.symbols.ref(name)}")1167            else:1168                names_kv = ", ".join(1169                    f"{name!r}: {frame.symbols.ref(name)}" for name in var_names1170                )1171                self.writeline(f"context.vars.update({{{names_kv}}})")1172        if discarded_names:1173            if len(discarded_names) == 1:1174                self.writeline(f"context.exported_vars.discard({discarded_names[0]!r})")1175            else:1176                names_str = ", ".join(map(repr, discarded_names))1177                self.writeline(1178                    f"context.exported_vars.difference_update(({names_str}))"1179                )1180 1181    def visit_For(self, node: nodes.For, frame: Frame) -> None:1182        loop_frame = frame.inner()1183        loop_frame.loop_frame = True1184        test_frame = frame.inner()1185        else_frame = frame.inner()1186 1187        # try to figure out if we have an extended loop.  An extended loop1188        # is necessary if the loop is in recursive mode if the special loop1189        # variable is accessed in the body if the body is a scoped block.1190        extended_loop = (1191            node.recursive1192            or "loop"1193            in find_undeclared(node.iter_child_nodes(only=("body",)), ("loop",))1194            or any(block.scoped for block in node.find_all(nodes.Block))1195        )1196 1197        loop_ref = None1198        if extended_loop:1199            loop_ref = loop_frame.symbols.declare_parameter("loop")1200 

Showing the first 1,200 of 1999 lines. Download the file for the rest.

codekingpro/portable-devtools · Team Ai