Operation result

dbt/adapters/default/impl.py

@@ -471,7 +471,7 @@ def acquire_connection(cls, profile, name):
             lock.release()
 
     @classmethod
-    def release_connection(cls, profile, name):
+    def release_connection(cls, profile, name='master'):
         global connections_in_use, connections_available, lock
 
         if connections_in_use.get(name) is None:
@@ -773,3 +773,45 @@ def convert_agate_type(cls, agate_table, col_idx):
         for agate_cls, func in conversions:
             if isinstance(agate_type, agate_cls):
                 return func(agate_table, col_idx)
+
+    ###
+    # Operations involving the manifest
+    ###
+    @classmethod
+    def run_operation(cls, profile, project_cfg, manifest, operation_name,
+                      result_key):
+        """Look the operation identified by operation_name up in the manifest
+        and run it.
+
+        Return an an AttrDict with three attributes: 'table', 'data', and
+            'status'. 'table' is an agate.Table.
+        """
+        operation = manifest.find_operation_by_name(operation_name, None)
+
+        # This causes a reference cycle, as dbt.context.runtime.generate()
+        # ends up calling get_adapter, so the import has to be here.
+        import dbt.context.runtime
+        context = dbt.context.runtime.generate(
+            operation,
+            project_cfg,
+            manifest.to_flat_graph(),
+        )
+
+        # TODO: should I get the return value here in case future operations
+        # want to return some string? Jinja (I think) stringifies the results
+        # so it's not super useful. Status, I guess?
+        operation.generator(context)()
+
+        # This is a lot of magic, have to know the magic name is 'catalog'.
+        # TODO: How can we make this part of the data set? Could we make it
+        # the operation's name/unique ID somehow instead?
+        result = context['load_result'](result_key)
+        return result
+
+    ###
+    # Abstract methods involving the flat graph
+    ###
+    @classmethod
+    def get_catalog(cls, profile, project_cfg, run_operation):
+        raise dbt.exceptions.NotImplementedException(
+            '`get_catalog` is not implemented for this adapter!')

dbt/adapters/postgres/impl.py

@@ -9,6 +9,9 @@
 
 from dbt.logger import GLOBAL_LOGGER as logger
 
+GET_CATALOG_OPERATION_NAME = 'get_catalog_data'
+GET_CATALOG_RESULT_KEY = 'catalog'  # defined in get_catalog() macro
+
 
 class PostgresAdapter(dbt.adapters.default.DefaultAdapter):
 
@@ -201,3 +204,14 @@ def convert_date_type(cls, agate_table, col_idx):
     @classmethod
     def convert_time_type(cls, agate_table, col_idx):
         return "time"
+
+    @classmethod
+    def get_catalog(cls, profile, project_cfg, manifest):
+        results = cls.run_operation(profile, project_cfg, manifest,
+                                    GET_CATALOG_OPERATION_NAME,
+                                    GET_CATALOG_RESULT_KEY)
+
+        schemas = cls.get_existing_schemas(profile, project_cfg)
+        results = results.table.where(lambda r: r['table_schema'] in schemas)
+
+        return results

dbt/clients/jinja.py

@@ -100,6 +100,28 @@ def parse(self, parser):
         return node
 
 
+class OperationExtension(jinja2.ext.Extension):
+    tags = ['operation']
+
+    def parse(self, parser):
+        node = jinja2.nodes.Macro(lineno=next(parser.stream).lineno)
+        operation_name = \
+            parser.parse_assign_target(name_only=True).name
+
+        node.args = []
+        node.defaults = []
+
+        while parser.stream.skip_if('comma'):
+            target = parser.parse_assign_target(name_only=True)
+
+        node.name = dbt.utils.get_operation_macro_name(operation_name)
+
+        node.body = parser.parse_statements(('name:endoperation',),
+                                            drop_needle=True)
+
+        return node
+
+
 def create_macro_capture_env(node):
 
     class ParserMacroCapture(jinja2.Undefined):
@@ -142,6 +164,7 @@ def get_template(string, ctx, node=None, capture_macros=False):
             args['undefined'] = create_macro_capture_env(node)
 
         args['extensions'].append(MaterializationExtension)
+        args['extensions'].append(OperationExtension)
 
         env = MacroFuzzEnvironment(**args)
 

dbt/compat.py

@@ -10,8 +10,10 @@
 
 if WHICH_PYTHON == 2:
     basestring = basestring
+    bigint = long
 else:
     basestring = str
+    bigint = int
 
 
 def to_unicode(s):

dbt/context/common.py

@@ -4,6 +4,8 @@
 
 from dbt.adapters.factory import get_adapter
 from dbt.compat import basestring, to_string
+from dbt.node_types import NodeType
+from dbt.contracts.graph.parsed import ParsedMacro, ParsedNode
 
 import dbt.clients.jinja
 import dbt.clients.agate_helper
@@ -75,6 +77,8 @@ def _add_macros(context, model, flat_graph):
     macros_to_add = {'global': [], 'local': []}
 
     for unique_id, macro in flat_graph.get('macros', {}).items():
+        if macro.get('resource_type') != NodeType.Macro:
+            continue
         package_name = macro.get('package_name')
 
         macro_map = {
@@ -201,8 +205,14 @@ def __init__(self, model, context, overrides):
         self.overrides = overrides
 
         if isinstance(model, dict) and model.get('unique_id'):
-            local_vars = model.get('config', {}).get('vars')
+            local_vars = model.get('config', {}).get('vars', {})
             self.model_name = model.get('name')
+        elif isinstance(model, ParsedMacro):
+            local_vars = {}  # macros have no config
+            self.model_name = model.name
+        elif isinstance(model, ParsedNode):
+            local_vars = model.config.get('vars', {})
+            self.model_name = model.name
         else:
             # still used for wrapping
             self.model_name = model.nice_name

dbt/contracts/graph/parsed.py

@@ -182,7 +182,10 @@
                 'maxLength': 127,
             },
             'resource_type': {
-                'enum': [NodeType.Macro],
+                'enum': [
+                    NodeType.Macro,
+                    NodeType.Operation,
+                ],
             },
             'unique_id': {
                 'type': 'string',
@@ -344,6 +347,29 @@ def serialize(self):
             'child_map': forward_edges,
         }
 
+    def _find_by_name(self, name, package, subgraph, nodetype):
+        """
+
+        Find a node by its given name in the appropraite sugraph.
+        """
+        if subgraph == 'nodes':
+            search = self.nodes
+        elif subgraph == 'macros':
+            search = self.macros
+        else:
+            raise NotImplementedError(
+                'subgraph search for {} not implemented'.format(subgraph)
+            )
+        return dbt.utils.find_in_subgraph_by_name(
+            search,
+            name,
+            package,
+            nodetype)
+
+    def find_operation_by_name(self, name, package):
+        return self._find_by_name(name, package, 'macros',
+                                  [NodeType.Operation])
+
     def to_flat_graph(self):
         """Convert the parsed manifest to the 'flat graph' that the compiler
         expects.

dbt/contracts/graph/unparsed.py

@@ -63,6 +63,9 @@
                     NodeType.Model,
                     NodeType.Test,
                     NodeType.Analysis,
+                    # Note: Hooks fail if you remove this, even though it's
+                    # also allowed in ParsedMacro, which seems wrong.
+                    # Maybe need to move hook operations into macros?
                     NodeType.Operation,
                     NodeType.Seed,
                     # we need this if parse_node is going to handle archives.

dbt/include/global_project/macros/adapters/common.sql

@@ -66,3 +66,61 @@
     {{ column_list_for_create_table(columns) }}
   );
 {% endmacro %}
+
+
+{% macro get_catalog() -%}
+  {{ adapter_macro('get_catalog') }}
+{%- endmacro %}
+
+
+{% macro default__get_catalog() -%}
+
+  {% set typename = adapter.type() %}
+  {% set msg -%}
+    get_catalog not implemented for {{ typename }}
+  {%- endset %}
+
+  {{ exceptions.raise_compiler_error(msg) }}
+{% endmacro %}
+
+
+{% macro postgres__get_catalog() -%}
+  {%- call statement('catalog', fetch_result=True) -%}
+    with tables as (
+      select
+          table_schema,
+          table_name,
+          table_type
+
+      from information_schema.tables
+
+      ),
+
+      columns as (
+
+          select
+              table_schema,
+              table_name,
+              null as table_comment,
+
+              column_name,
+              ordinal_position as column_index,
+              data_type as column_type,
+              null as column_comment
+
+
+          from information_schema.columns
+
+      )
+
+      select *
+      from tables
+      join columns using (table_schema, table_name)
+
+      where table_schema != 'information_schema'
+        and table_schema not like 'pg_%'
+  {%- endcall -%}
+  {# There's no point in returning anything as the jinja macro stuff calls #}
+  {# str() on all returns. To get the results, you'll need to use #}
+  {# context['load_result']('catalog') #}
+{%- endmacro %}

dbt/include/global_project/macros/operations/catalog/get_catalog.sql

@@ -0,0 +1,4 @@
+{% operation get_catalog_data %}
+    {% set catalog = dbt.get_catalog() %}
+    {{ return(catalog) }}
+{% endoperation %}

dbt/loader.py

@@ -12,6 +12,7 @@ class GraphLoader(object):
     @classmethod
     def load_all(cls, root_project, all_projects):
         macros = MacroLoader.load_all(root_project, all_projects)
+        macros.update(OperationLoader.load_all(root_project, all_projects))
         nodes = {}
         for loader in cls._LOADERS:
             nodes.update(loader.load_all(root_project, all_projects, macros))
@@ -85,6 +86,20 @@ def load_project(cls, root_project, all_projects, project, project_name,
                 macros=macros)
 
 
+class OperationLoader(ResourceLoader):
+
+    @classmethod
+    def load_project(cls, root_project, all_projects, project, project_name,
+                     macros):
+        return dbt.parser.load_and_parse_macros(
+            package_name=project_name,
+            root_project=root_project,
+            all_projects=all_projects,
+            root_dir=project.get('project-root'),
+            relative_dirs=project.get('macro-paths', []),
+            resource_type=NodeType.Operation)
+
+
 class AnalysisLoader(ResourceLoader):
 
     @classmethod

dbt/main.py

@@ -17,6 +17,7 @@
 import dbt.task.seed as seed_task
 import dbt.task.test as test_task
 import dbt.task.archive as archive_task
+import dbt.task.generate as generate_task
 
 import dbt.tracking
 import dbt.config as config
@@ -414,6 +415,15 @@ def parse_args(args):
     )
     seed_sub.set_defaults(cls=seed_task.SeedTask, which='seed')
 
+    catalog_sub = subs.add_parser('catalog', parents=[base_subparser])
+    catalog_subs = catalog_sub.add_subparsers()
+    # it might look like catalog_sub is the correct parents entry, but that
+    # will cause weird errors about 'conflicting option strings'.
+    generate_sub = catalog_subs.add_parser('generate',
+                                           parents=[base_subparser])
+    generate_sub.set_defaults(cls=generate_task.GenerateTask,
+                              which='generate')
+
     sub = subs.add_parser('test', parents=[base_subparser])
     sub.add_argument(
         '--data',

dbt/node_runners.py

@@ -3,6 +3,7 @@
 from dbt.exceptions import NotImplementedException
 from dbt.utils import get_nodes_by_tags
 from dbt.node_types import NodeType, RunHookType
+from dbt.adapters.factory import get_adapter
 
 import dbt.clients.jinja
 import dbt.context.runtime
@@ -61,6 +62,26 @@ def skipped(self):
         return self.skip
 
 
+class RunOperationResult(RunModelResult):
+    def __init__(self, node, error=None, skip=False, status=None,
+                 failed=None, execution_time=0, returned=None):
+        super(RunOperationResult, self).__init__(node, error, skip, status,
+                                                 failed, execution_time)
+        self.returned = returned
+
+    @property
+    def errored(self):
+        return self.error is not None
+
+    @property
+    def failed(self):
+        return self.fail
+
+    @property
+    def skipped(self):
+        return self.skip
+
+
 class BaseRunner(object):
     print_header = True