Umstellung Typer

database/db_create.py

@@ -1,7 +1,7 @@
+import pandas as pd
 import json
-from collections import namedtuple
 from pathlib import Path
-import pandas as pd
+from collections import namedtuple
 import pyodbc
 
 # from re import escape
@@ -11,7 +11,7 @@ import pyodbc
 
 DbCreateConfig = namedtuple(
     "DbCreateConfig",
-    "name csv_file clients filter source_dsn target_dsn stage_dir batch_dir",
+    "name csv_file clients filter source_dsn target_dsn stage_dir batch_dir logs_dir",
 )
 DsnConfig = namedtuple("DsnConfig", "user password server database driver schema")
 
@@ -113,6 +113,19 @@ class database_inspect:
             source_insp_cols = [col[0] for col in q.fetchall()]
         return source_insp_cols
 
+    def get_columns_is_typeof_str(self, table):
+        source_insp_cols = [
+            col.data_type in [pyodbc.SQL_CHAR, pyodbc.SQL_VARCHAR]
+            for col in self.cursor.columns(table=table)
+        ]
+        if len(source_insp_cols) == 0:
+            q = self.cursor.execute(
+                "SELECT COLLATION_NAME as column_collation FROM information_schema.columns "
+                + f"WHERE TABLE_NAME = '{self.convert_table(table)}'"
+            )
+            source_insp_cols = [len(col[0]) > 0 for col in q.fetchall()]
+        return source_insp_cols
+
     def convert_table(self, table):
         if "." in table:
             table = table.split(".")[-1]
@@ -184,6 +197,10 @@ def create(config_file="dbtools/OPTIMA.json"):  #
             )
 
             target_columns_list = target_db.get_columns(current_table["target"])
+            target_column_types = target_db.get_columns_is_typeof_str(
+                current_table["target"]
+            )
+
             if "CLIENT_DB" in target_columns_list:
                 target_columns_list.remove("CLIENT_DB")
                 target_columns_list.append("Client_DB")
@@ -235,11 +252,14 @@ def create(config_file="dbtools/OPTIMA.json"):  #
                     )
                 # select_columns = "T1.[" + "], T1.[".join(intersect) + "],"
                 select_columns = ""
-                for col in target_columns_list:
+                for col, col_type in zip(target_columns_list, target_column_types):
                     if col in intersect:
-                        select_columns += f"T1.[{col}], "
+                        if col_type:
+                            select_columns += f"dbo.cln(T1.[{col}]), "
+                        else:
+                            select_columns += f"T1.[{col}], "
                     elif col == "Client_DB":
-                        select_columns += "'" + client_db + '\' as \\"Client_DB\\", '
+                        select_columns += f"'{client_db}' as \\\"Client_DB\\\", "
                     else:
                         select_columns += "'' as \\\"" + col + '\\", '
 

db.py

@@ -1,25 +1,28 @@
 import config
 import database
-import plac
+import typer
 
 
-class DatabaseTools:
-    commands = ["create", "run", "schema"]
+app = typer.Typer()
+cfg = config.Config()
 
-    def __init__(self):
-        self.cfg = config.Config()
 
-    def create(self, config_file: str):
-        config_file = self.cfg.system_dir + f"\\SQL\\config\\{config_file}.json"
-        database.create(config_file)
+@app.command()
+def create(config_file: str):
+    config_file = cfg.system_dir + f"\\SQL\\config\\{config_file}.json"
+    database.create(config_file)
 
-    def run(self):
-        batch_dir = self.cfg.system_dir + "\\SQL\\batch"
-        database.run(batch_dir)
 
-    def schema(self):
-        database.schema()
+@app.command()
+def run():
+    batch_dir = cfg.system_dir + "\\SQL\\batch"
+    database.run(batch_dir)
+
+
+@app.command()
+def schema():
+    database.schema()
 
 
 if __name__ == "__main__":
-    plac.Interpreter.call(DatabaseTools)
+    app()