Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def id_matches(unique_id, target_name, target_package, nodetypes, model):
"""Return True if the unique ID matches the given name, package, and type.
If package is None, any package is allowed.
nodetypes should be a container of NodeTypes that implements the 'in'
operator.
"""
node_type = model.resource_type
node_parts = unique_id.split('.', 2)
if len(node_parts) != 3:
msg = "unique_id {} is malformed".format(unique_id)
dbt.exceptions.raise_compiler_error(msg, model)
resource_type, package_name, node_name = node_parts
if resource_type not in nodetypes:
return False
if node_type == NodeType.Source.value:
if node_name.count('.') != 1:
msg = "{} names must contain exactly 1 '.' character"\
.format(node_type)
dbt.exceptions.raise_compiler_error(msg, model)
else:
if '.' in node_name:
msg = "{} names cannot contain '.' characters".format(node_type)
dbt.exceptions.raise_compiler_error(msg, model)
if target_name != node_name:
def __deepcopy__(self, memo):
path = os.path.join(self.node.root_path,
self.node.original_file_path)
logger.debug(
'dbt encountered an undefined variable, "{}" in node {}.{} '
'(source path: {})'
.format(self.name, self.node.package_name,
self.node.name, path))
# match jinja's message
dbt.exceptions.raise_compiler_error(
"{!r} is undefined".format(self.name),
node=self.node
)
if legacy in names:
extra.append(legacy)
if missing:
if extra:
msg = (
'Snapshot target has ("{}") but not ("{}") - is it an '
'unmigrated previous version archive?'
.format('", "'.join(extra), '", "'.join(missing))
)
else:
msg = (
'Snapshot target is not a snapshot table (missing "{}")'
.format('", "'.join(missing))
)
raise_compiler_error(msg)
def _transform_config(self, config):
for oldkey in ('pre_hook', 'post_hook'):
if oldkey in config:
newkey = oldkey.replace('_', '-')
if newkey in config:
dbt.exceptions.raise_compiler_error(
'Invalid config, has conflicting keys "{}" and "{}"'
.format(oldkey, newkey),
self.model
)
config[newkey] = config.pop(oldkey)
return config
def find_blocks(self, allowed_blocks=None, collect_raw_data=True):
"""Find all top-level blocks in the data."""
if allowed_blocks is None:
allowed_blocks = {'snapshot', 'macro', 'materialization', 'docs'}
for tag in self.tag_parser.find_tags():
if tag.block_type_name in _CONTROL_FLOW_TAGS:
self.stack.append(tag.block_type_name)
elif tag.block_type_name in _CONTROL_FLOW_END_TAGS:
found = None
if self.stack:
found = self.stack.pop()
else:
expected = _CONTROL_FLOW_END_TAGS[tag.block_type_name]
dbt.exceptions.raise_compiler_error((
'Got an unexpected control flow end tag, got {} but '
'never saw a preceeding {} (@ {})'
).format(tag.block_type_name, expected, tag.start))
expected = _CONTROL_FLOW_TAGS[found]
if expected != tag.block_type_name:
dbt.exceptions.raise_compiler_error((
'Got an unexpected control flow end tag, got {} but '
'expected {} next (@ {})'
).format(tag.block_type_name, expected, tag.start))
if tag.block_type_name in allowed_blocks:
if self.stack:
dbt.exceptions.raise_compiler_error((
'Got a block definition inside control flow at {}. '
'All dbt block definitions must be at the top level'
).format(tag.start))
def get_template(string, ctx, node=None, capture_macros=False):
try:
env = get_environment(node, capture_macros)
template_source = str(string)
return env.from_string(template_source, globals=ctx)
except (jinja2.exceptions.TemplateSyntaxError,
jinja2.exceptions.UndefinedError) as e:
e.translated = False
dbt.exceptions.raise_compiler_error(str(e), node)
def _transform_config(self, config):
for oldkey in ('pre_hook', 'post_hook'):
if oldkey in config:
newkey = oldkey.replace('_', '-')
if newkey in config:
dbt.exceptions.raise_compiler_error(
'Invalid config, has conflicting keys "{}" and "{}"'
.format(oldkey, newkey),
self.model
)
config[newkey] = config.pop(oldkey)
return config
def render_template(template, ctx, node=None):
try:
return template.render(ctx)
except (jinja2.exceptions.TemplateSyntaxError,
jinja2.exceptions.UndefinedError) as e:
e.translated = False
dbt.exceptions.raise_compiler_error(str(e), node)
def impl(message_if_exception, func, *args, **kwargs):
try:
return func(*args, **kwargs)
except Exception:
dbt.exceptions.raise_compiler_error(message_if_exception, model)
return impl
key: new_configs[key] for key
in new_configs if key in config_keys
}
for key in self.AppendListFields:
append_fields = self.__get_as_list(relevant_configs, key)
mutable_config[key].extend([
f for f in append_fields if f not in mutable_config[key]
])
for key in self.ExtendDictFields:
dict_val = relevant_configs.get(key, {})
try:
mutable_config[key].update(dict_val)
except (ValueError, TypeError, AttributeError):
dbt.exceptions.raise_compiler_error(
'Invalid config field: "{}" must be a dict'.format(key)
)
for key in (self.ClobberFields | self.AdapterSpecificConfigs):
if key in relevant_configs:
mutable_config[key] = relevant_configs[key]
return relevant_configs