main: better config handling

Author: balinthaller

flyover/__main__.py

@@ -29,20 +29,11 @@
 
 def run_csv_job(config_file, output_dir):
     carte_loader = CarteLoader()
-    extractors, connections = parse_config(config_file)
-
-    extractor_config = {}
-
-    for extractor, connection in zip(extractors, connections):
-        scope = extractor.get_scope()
-        config = connection.get("config", {})
-        for conf_key, conf_value in config.items():
-            extractor_config[f"{scope}.{conf_key}"] = conf_value
+    extractors, config = parse_config(config_file)
 
     job_config = ConfigFactory.from_dict(
         {"loader.carte.tables_output_path": output_dir,
-         # f'extractor.postgres_metadata.where_clause_suffix: '
-        **extractor_config}
+         **config}
     )
 
     for extractor in extractors:

flyover/utils/config_parser.py

@@ -5,13 +5,31 @@
 
 yaml = YAML()
 
+CONFIG_KEY = "config"
+
 
 def create_glue_connection(conn_dict):
-    return GlueExtractor(conn_dict.get("name", "glue"))
+    return GlueExtractor(conn_dict.get("name", "glue")), {}
 
 
 def create_postgres_connection(conn_dict):
-    return PostgresMetadataExtractor()
+    extractor = PostgresMetadataExtractor()
+    extractor_scope = extractor.get_scope()
+    config = conn_dict.get(CONFIG_KEY, {})
+    included_schemas = config.get("included_schemas", ["public"])
+    schemas_sql_list = "('{schemas}')".format(schemas="', '".join(included_schemas))
+    schema_where_clause = f"where table_schema in {schemas_sql_list}"
+
+    connection_string = config.get("connection_string")
+    if connection_string is None:
+        raise KeyError(
+            "connection_string is a required config item for PostgreSQL connections"
+        )
+
+    return PostgresMetadataExtractor(), {
+        f"{extractor_scope}.{PostgresMetadataExtractor.WHERE_CLAUSE_SUFFIX_KEY}": schema_where_clause,
+        f"{extractor_scope}.extractor.sqlalchemy.conn_string": connection_string,
+    }
 
 
 CONNECTION_FACTORIES = {
@@ -27,12 +45,23 @@ def parse_config(filename):
 
     connections = parsed_data.get("connections", [])
 
-    extractors = [
-        CONNECTION_FACTORIES[conn_dict.get("type")](conn_dict)
-        for conn_dict in connections
-    ]
+    extractors = []
+    config = {}
+
+    for conn_dict in connections:
+        extractor, extractor_config = CONNECTION_FACTORIES[conn_dict.get("type")](
+            conn_dict
+        )
+        custom_config = {}
+        scope = extractor.get_scope()
+        config = conn_dict.get(CONFIG_KEY, {})
+        for conf_key, conf_value in config.items():
+            custom_config[f"{scope}.{conf_key}"] = conf_value
+
+        config = {**config, **extractor_config, **custom_config}
+        extractors.append(extractor)
 
-    return extractors, connections
+    return extractors, config
 
 
 def _read_file(filename: str):

flyover/utils/frontmatter.py

@@ -12,11 +12,11 @@ def parse(filename):
     metadata = {}
     content = ""
 
-    if len(data) >= 1 and data[0] != "":
-        metadata = yaml.load(data[0])
+    if len(data) >= 2 and data[1] != "":
+        metadata = yaml.load(data[1])
 
     if len(data) >= 2:
-        content = data[1]
+        content = data[2]
 
     return metadata, content