Reverted changes that had overwritten the previous linting of this file.

Author: JanPeterDatakind

docker/airflow/dags/run_dot_project.py

@@ -20,11 +20,11 @@
 
 
 def get_object(
-    object_name_in,
-    earliest_date_to_sync,
-    date_field,
-    source_conn_in,
-    columns_to_exclude,
+        object_name_in,
+        earliest_date_to_sync,
+        date_field,
+        source_conn_in,
+        columns_to_exclude,
 ):
     """
 
@@ -47,9 +47,18 @@ def get_object(
 
     connection = BaseHook.get_connection(source_conn_in)
 
-    sql_stmt = "SELECT * FROM " + connection.schema + "." + object_name_in
+    sql_stmt = (
+            "SELECT * FROM "
+            + connection.schema
+            + "."
+            + object_name_in
+    )
     if date_field != None:
-        sql_stmt += " WHERE " + date_field + " >= '" + earliest_date_to_sync + "'"
+        sql_stmt += (" WHERE "
+                     + date_field
+                     + " >= '"
+                     + earliest_date_to_sync
+                     + "'")
     print(sql_stmt)
     pg_hook = PostgresHook(postgres_conn_id=source_conn_in, schema=source_conn_in)
     pg_conn = pg_hook.get_conn()
@@ -58,13 +67,13 @@ def get_object(
     data = cursor.fetchall()
 
     sql_stmt = (
-        "SELECT column_name, data_type FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name = '"
-        + object_name_in
-        + "'"
-        + "AND table_schema = '"
-        + connection.schema
-        + "' "
-        + " ORDER BY ordinal_position "
+            "SELECT column_name, data_type FROM INFORMATION_SCHEMA.COLUMNS WHERE table_name = '"
+            + object_name_in
+            + "'"
+            + "AND table_schema = '"
+            + connection.schema
+            + "' "
+            + " ORDER BY ordinal_position "
     )
     print(sql_stmt)
     pg_hook = PostgresHook(postgres_conn_id=source_conn_in, schema=source_conn_in)
@@ -76,20 +85,16 @@ def get_object(
     # Fail back to views
     if len(columns) == 0:
         sql_stmt = (
-            'SELECT a.attname as "column_name",'
-            + ' pg_catalog.format_type(a.atttypid, a.atttypmod) as "data_type" '
-            + " FROM pg_attribute a "
-            + "  JOIN pg_class t on a.attrelid = t.oid "
-            + "  JOIN pg_namespace s on t.relnamespace = s.oid "
-            + " WHERE a.attnum > 0 "
-            + "  AND NOT a.attisdropped "
-            + " AND t.relname = '"
-            + object_name_in
-            + "' "
-            + " AND s.nspname = '"
-            + connection.schema
-            + "' "
-            + " ORDER BY a.attnum; "
+                "SELECT a.attname as \"column_name\","
+                + " pg_catalog.format_type(a.atttypid, a.atttypmod) as \"data_type\" "
+                + " FROM pg_attribute a "
+                + "  JOIN pg_class t on a.attrelid = t.oid "
+                + "  JOIN pg_namespace s on t.relnamespace = s.oid "
+                + " WHERE a.attnum > 0 "
+                + "  AND NOT a.attisdropped "
+                + " AND t.relname = '" + object_name_in + "' "
+                + " AND s.nspname = '" + connection.schema + "' "
+                + " ORDER BY a.attnum; "
         )
         print(sql_stmt)
         pg_hook = PostgresHook(postgres_conn_id=source_conn_in, schema=source_conn_in)
@@ -131,7 +136,7 @@ def get_object(
 
 
 def save_object(
-    object_name_in, target_conn_in, data_in, column_list_in, type_list_in, source_db_in
+        object_name_in, target_conn_in, data_in, column_list_in, type_list_in, source_db_in
 ):
     """
 
@@ -159,16 +164,16 @@ def save_object(
 
     connection = BaseHook.get_connection(target_conn_in)
     connection_string = (
-        "postgresql://"
-        + str(connection.login)
-        + ":"
-        + str(connection.password)
-        + "@"
-        + str(connection.host)
-        + ":"
-        + str(connection.port)
-        + "/"
-        + target_conn_in
+            "postgresql://"
+            + str(connection.login)
+            + ":"
+            + str(connection.password)
+            + "@"
+            + str(connection.host)
+            + ":"
+            + str(connection.port)
+            + "/"
+            + target_conn_in
     )
 
     engine = create_engine(
@@ -185,7 +190,7 @@ def save_object(
     # This will also drop any DOT model views onto this data
     if MODE == "replace":
         with PostgresHook(
-            postgres_conn_id=target_conn_in, schema=target_conn_in
+                postgres_conn_id=target_conn_in, schema=target_conn_in
         ).get_conn() as conn:
             cur = conn.cursor()
             query = f"DROP TABLE IF EXISTS {schema}.{object_name_in} CASCADE;"
@@ -197,7 +202,7 @@ def save_object(
 
     # Test to see if schema exists, if not, create
     with PostgresHook(
-        postgres_conn_id=target_conn_in, schema=target_conn_in
+            postgres_conn_id=target_conn_in, schema=target_conn_in
     ).get_conn() as conn:
         cur = conn.cursor()
         query = f"CREATE SCHEMA IF NOT EXISTS {schema};"
@@ -223,12 +228,12 @@ def save_object(
 
 
 def sync_object(
-    object_name_in,
-    earliest_date_to_sync,
-    date_field,
-    source_conn_in,
-    target_conn_in,
-    columns_to_exclude,
+        object_name_in,
+        earliest_date_to_sync,
+        date_field,
+        source_conn_in,
+        target_conn_in,
+        columns_to_exclude,
 ):
     """
 
@@ -264,7 +269,6 @@ def sync_object(
         object_name_in, target_conn_in, data, column_list, type_list, source_conn_in
     )
 
-
 def drop_tables_in_dot_tests_schema(target_conn_in, schema_to_drop_from):
     """
     We are syncing new data where new columns and columns types might change.
@@ -285,24 +289,21 @@ def drop_tables_in_dot_tests_schema(target_conn_in, schema_to_drop_from):
     """
 
     with PostgresHook(
-        postgres_conn_id=target_conn_in, schema=target_conn_in
+            postgres_conn_id=target_conn_in, schema=target_conn_in
     ).get_conn() as conn:
         cur = conn.cursor()
         query1 = f"SET search_path TO {schema_to_drop_from}"
-        query2 = (
-            f"DO $$ DECLARE "
-            f"r RECORD; "
-            f"BEGIN "
-            f"FOR r IN (SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema()) "
-            f"LOOP	"
-            f"EXECUTE 'DROP TABLE IF EXISTS ' || QUOTE_IDENT(r.table_name) || ' CASCADE'; "
-            f"END LOOP; "
-            f"END $$; "
-        )
+        query2 = f"DO $$ DECLARE " \
+                 f"r RECORD; " \
+                 f"BEGIN " \
+                 f"FOR r IN (SELECT table_name FROM information_schema.tables WHERE table_schema = current_schema()) " \
+                 f"LOOP	" \
+                 f"EXECUTE 'DROP TABLE IF EXISTS ' || QUOTE_IDENT(r.table_name) || ' CASCADE'; " \
+                 f"END LOOP; " \
+                 f"END $$; "
         cur.execute(query1)
         cur.execute(query2)
 
-
 def run_dot_app(project_id_in):
     """
     Method to run the DOT.
@@ -335,10 +336,10 @@ def default_config():
 
 
 with DAG(
-    dag_id="run_dot_project",
-    schedule_interval="@weekly",
-    start_date=datetime(year=2022, month=3, day=1),
-    catchup=False,
+        dag_id="run_dot_project",
+        schedule_interval="@weekly",
+        start_date=datetime(year=2022, month=3, day=1),
+        catchup=False,
 ) as dag:
     config = json.loads(Variable.get("dot_config", default_var=default_config().read()))
 
@@ -374,7 +375,7 @@ def default_config():
                 python_callable=drop_tables_in_dot_tests_schema,
                 op_kwargs={
                     "target_conn_in": target_conn,
-                    "schema_to_drop_from": schema_to_drop_from,
+                    "schema_to_drop_from": schema_to_drop_from
                 },
                 dag=dag,
             )
@@ -384,10 +385,7 @@ def default_config():
         for i in range(len(objects_to_sync)):
 
             object_name = objects_to_sync[i]["object"]
-            if (
-                "date_field" in objects_to_sync[i]
-                and objects_to_sync[i]["date_field"] != ""
-            ):
+            if "date_field" in objects_to_sync[i] and objects_to_sync[i]["date_field"] != "":
                 date_field = objects_to_sync[i]["date_field"]
             else:
                 date_field = None