Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Tag for
{% for %}
...
{% endfor %}
"""
from varname import namedtuple
from ..tagmgr import register_tag
from ..tag import Tag
from ..tagfrag import try_render
ForLoop = namedtuple( # pylint: disable=invalid-name
['first', 'index', 'index0', 'last',
'length', 'rindex', 'rindex0']
)
@register_tag
class TagFor(Tag):
"""Class for for tag"""
SYNTAX = r"""
inner_tag: tag_for
!tag_for: $tagnames VAR "in" atom for_args*
?for_args: for_limit_arg | for_offset_arg | for_reversed_arg
for_limit_arg: "limit" ":" (int|var)
for_offset_arg: "offset" ":" (int|var)
for_reversed_arg: "reversed"
"""
"""Tag cycle
{% for %}
...
{% cycle "one", "two", "three" %}
...
{% endfor %}
"""
from varname import namedtuple
from lark import v_args
from ...tagmgr import register_tag
from ..tagparser import Tag, TagTransformer
from ..tagfrag import TagFrag, TagFragConst
CycleObject = namedtuple( # pylint: disable=invalid-name
['group', 'args', 'at']
)
class TagFragCycle(TagFrag):
def render(self, local_envs, global_envs):
group, args = self.data
group = group.render(local_envs, global_envs)
args = [arg.render(local_envs, global_envs) for arg in args]
return CycleObject(group, args, at=[0])
@v_args(inline=True)
class TransformerCycle(TagTransformer):
"""Transformer for fragment parsing cycle tag"""
atom = TagTransformer.tags__atom
{{ product.title }}
{% endtablerow %}
"""
import importlib
from varname import namedtuple
from lark import v_args
from ...tagmgr import register_tag
from ..tagparser import Tag, TagTransformer
from ..tagfrag import TagFrag, TagFragConst
# tagfor module, we cannot do: from . import for
# since for is a keyword
tagfor = importlib.import_module('.for', package=__package__)
TablerowObject = namedtuple( # pylint: disable=invalid-name
['itername', 'obj', 'limit', 'offset', 'cols']
)
class TagFragTablerow(TagFrag):
def render(self, local_envs, global_envs):
itername, expr, args = self.data
obj = expr.render(local_envs, global_envs)
args = dict(args)
limit = args.get('limit', None)
if limit:
limit = limit.render(local_envs, global_envs)
offset = args.get('offset', None)
if offset:
from functools import partial
from collections import deque, OrderedDict
from varname import namedtuple
from diot import Diot
from lark import v_args, Lark, Transformer as LarkTransformer
from lark.exceptions import VisitError as LarkVisitError
# load all shared tags
from . import tags # pylint: disable=unused-import
from ..tagmgr import get_tag
from ..config import LIQUID_LOG_INDENT
from ..exceptions import (
TagUnclosed, EndTagUnexpected,
TagWrongPosition
)
TagContext = namedtuple(['template_name', # pylint: disable=invalid-name
'context_getter',
'line',
'column',
'logger'])
@v_args(inline=True)
class Transformer(LarkTransformer):
"""Transformer class to transform the trees/tokens
Attributes:
_stacks (deque): The stack used to handle the relationships between
tags
_direct_tags (list): The direct tags of the ROOT tag
"""
def __init__(self, config, template_info):
"""Tag tabrow
{% tablerow product in collection.products %}
{{ product.title }}
{% endtablerow %}
<table></table>
"""
from varname import namedtuple
from ..tagmgr import register_tag
from ..tag import Tag
from ..tagfrag import try_render
ForLoop = namedtuple( # pylint: disable=invalid-name
['first', 'index', 'index0', 'last',
'length', 'rindex', 'rindex0']
)
@register_tag
class TagTablerow(Tag):
"""Class for tablerow tag"""
SYNTAX = r"""
inner_tag: tag_tablerow
!tag_tablerow: $tagnames VAR "in" atom tablerow_args*
?tablerow_args: tablerow_limit_arg | tablerow_offset_arg | tablerow_cols_arg
tablerow_limit_arg: "limit" ":" (int|var)
tablerow_offset_arg: "offset" ":" (int|var)
tablerow_cols_arg: "cols" ":" (int|var)
"""
def copy(self, id=None, tag=None, depends=True):
"""@API
Like `proc`'s `copy` function, copy a procset.
Each processes will be copied.
@params:
id (str): Use a different id if you don't
want to use the variant name
tag (str): The new tag of all copied processes
depends (bool): Whether to copy the dependencies or not.
Default: True
- dependences for processes in starts will not be copied
@returns:
(ProcSet): The new procset
"""
id = id or varname()
ret = self.__class__(*self.procs.values(),
id=id,
tag=tag,
copy=True,
depends=False)
if depends:
for proc in ret.procs.values():
proc.depends = [
ret.procs[dep.id] if dep is self.procs[dep.id] else dep
for dep in self.procs[proc.id].depends
]
ret.starts.add(Proxy(ret.procs[proc.id] for proc in self.starts))
ret.ends.add(Proxy(ret.procs[proc.id] for proc in self.ends))
def __init__(self, *procs, id=None, tag=None, copy=True, depends=True):
"""@API
Constructor
@params:
*procs (Proc) : the set of processes
**kwargs: Other arguments to instantiate a `ProcSet`
depends (bool): Whether auto deduce depends.
Default: `True`
id (str): The id of the procset.
Default: `None` (the variable name)
tag (str): The tag of the processes. Default: `None`
copy (bool): Whether copy the processes or just use them.
Default: `True`
"""
self.__dict__['id'] = id or varname()
self.__dict__['tag'] = tag
self.__dict__['starts'] = Proxy()
self.__dict__['ends'] = Proxy()
self.__dict__['delegates'] = OrderedDiot(diot_nest=False)
self.__dict__['procs'] = OrderedDiot(diot_nest=False)
self.__dict__['modules'] = Diot(diot_nest=False)
# save initial states before a module is called
# states will be resumed before each module is called
self.__dict__['initials'] = Diot(diot_nest=False)
prevproc = None
for proc in procs:
assert hasattr(proc, 'id') and hasattr(proc, 'tag'), (
'Argument has to be a Proc object: %r.' % proc
)
if copy:
def copy(self, id=None, **kwargs): # pylint: disable=redefined-builtin, invalid-name
"""@API
Copy a process to a new one
Depends and nexts will be copied
@params:
id: The id of the new process
kwargs: Other arguments for constructing a process
"""
newid = id or varname()
raw_attrs = {key: try_deepcopy(value)
for key, value in self.__attrs_property_raw__.items()
if key not in ('input', 'id', 'channel', 'jobs',
'runtime_config', 'depends', 'nexts')}
# attr.ib not in __attrs_property_raw__
raw_attrs.update({
# only keep the keys of input
'input': (','.join(self._input)
if isinstance(self._input, list)
else ','.join(self._input.keys())
if isinstance(self._input, dict)
else self._input),
'desc': self.desc,
'envs': try_deepcopy(self.envs),
'nthread': self.nthread,
})
default=attr.Factory(lambda: varname(caller=2)),
repr=False,