Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _inject_runtime_config(adapter, node, extra_context):
wrapped_sql = node.wrapped_sql
context = _node_context(adapter, node)
context.update(extra_context)
sql = dbt.clients.jinja.get_rendered(wrapped_sql, context)
node.wrapped_sql = sql
return node
def collect_docrefs(
target: UnparsedSchemaYaml,
refs: ParserRef,
column_name: Optional[str],
*descriptions: str,
) -> None:
context = {'doc': docs(target, refs.docrefs, column_name)}
for description in descriptions:
get_rendered(description, context)
# Set this temporarily. Not the full config yet (as config() hasn't been
# called from jinja yet). But the Var() call below needs info about project
# level configs b/c they might contain refs. TODO: Restructure this?
config_dict = coalesce(archive_config, {})
config_dict.update(config.config)
node['config'] = config_dict
# Set this temporarily so get_rendered() below has access to a schema
profile = dbt.utils.get_profile_from_project(root_project_config)
default_schema = profile.get('schema', 'public')
node['schema'] = default_schema
context = dbt.context.parser.generate(node, root_project_config,
{"macros": macros})
dbt.clients.jinja.get_rendered(
node.get('raw_sql'), context, node,
capture_macros=True)
# Clean up any open connections opened by adapter functions that hit the db
db_wrapper = context['adapter']
adapter = db_wrapper.adapter
profile = db_wrapper.profile
adapter.release_connection(profile, node.get('name'))
# Special macro defined in the global project
schema_override = config.config.get('schema')
get_schema = context.get('generate_schema_name', lambda x: default_schema)
node['schema'] = get_schema(schema_override)
# Overwrite node config
config_dict = node.get('config', {})
def render_value(self, value, keypath=None):
# keypath is ignored.
# if it wasn't read as a string, ignore it
if not isinstance(value, compat.basestring):
return value
# force the result of rendering into this python version's native
# string type
return compat.to_native_string(get_rendered(value, self.context))
def process_docs_for_node(cls, manifest, current_project, node):
for docref in node.get('docrefs', []):
column_name = docref.get('column_name')
if column_name is None:
description = node.get('description', '')
else:
column = cls._get_node_column(node, column_name)
description = column.get('description', '')
context = {
'doc': docs(node, manifest, current_project, column_name),
}
# At this point, target_doc is a ParsedDocumentation, and we
# know that our documentation string has a 'docs("...")'
# pointing at it. We want to render it.
description = dbt.clients.jinja.get_rendered(description,
context)
# now put it back.
if column_name is None:
node.set('description', description)
else:
column['description'] = description
def __call__(self, var_name, default=_VAR_NOTSET):
self.assert_var_defined(var_name, default)
if var_name not in self.local_vars:
return default
raw = self.local_vars[var_name]
# if bool/int/float/etc are passed in, don't compile anything
if not isinstance(raw, basestring):
return raw
return dbt.clients.jinja.get_rendered(raw, self.context)
data = node.to_dict()
data.update({
'compiled': False,
'compiled_sql': None,
'extra_ctes_injected': False,
'extra_ctes': [],
'injected_sql': None,
})
compiled_node = CompiledNode(**data)
context = dbt.context.runtime.generate(
compiled_node, self.config, manifest)
context.update(extra_context)
compiled_node.compiled_sql = dbt.clients.jinja.get_rendered(
node.get('raw_sql'),
context,
node)
compiled_node.compiled = True
injected_node, _ = prepend_ctes(compiled_node, manifest)
should_wrap = {NodeType.Test, NodeType.Operation}
if injected_node.resource_type in should_wrap:
# data tests get wrapped in count(*)
# TODO : move this somewhere more reasonable
if 'data' in injected_node.tags and \
is_type(injected_node, NodeType.Test):
injected_node.wrapped_sql = (
"select count(*) from (\n{test_sql}\n) sbq").format(
def generate_source_node(self, source, table, path, package_name, root_dir,
refs):
unique_id = self.get_path(NodeType.Source, package_name,
source.name, table.name)
context = {'doc': dbt.context.parser.docs(source, refs.docrefs)}
description = table.get('description', '')
source_description = source.get('description', '')
get_rendered(description, context)
get_rendered(source_description, context)
freshness = dbt.utils.deep_merge(source.get('freshness', {}),
table.get('freshness', {}))
loaded_at_field = table.get('loaded_at_field',
source.get('loaded_at_field'))
# use 'or {}' to allow quoting: null
source_quoting = source.get('quoting') or {}
table_quoting = table.get('quoting') or {}
quoting = dbt.utils.deep_merge(source_quoting, table_quoting)
default_database = self.root_project_config.credentials.database
return ParsedSourceDefinition(
package_name=package_name,