db_schema mit SQL-Anywhere

database/db_schema.py

@@ -1,6 +1,7 @@
-import pyodbc
 import json
 
+import pyodbc
+
 
 def connect():
     c = pyodbc.connect("DSN=Autoline_direkt64;UID=kcc;PWD=kcc123")
@@ -26,6 +27,73 @@ def convert_desc(col):
     return ", ".join(list(map(str, col)))
 
 
+def convert_desc_sql_anywhere(col):
+    nullable = "NULL" if col.nullable == 1 else "NOT NULL"
+
+    if col.type_name in (
+        "char",
+        "varchar",
+        "long varchar",
+        "long nvarchar",
+        "nvarchar",
+    ):
+        sizes = [20, 50, 100, 255]
+        length = "MAX"
+        for s in sizes:
+            if col.length < s:
+                length = s
+                break
+        return f"   [{col.column_name}] [varchar]({length}) {nullable}"
+
+    if col.type_name in (
+        "binary",
+        "long binary",
+        "varbinary",
+    ):
+        return f"   [{col.column_name}] [binary]({col.length}) {nullable}"
+
+    if col.type_name in (
+        "bit",
+        "tinyint",
+    ):
+        return f"   [{col.column_name}] [tinyint] {nullable}"
+
+    if col.type_name in (
+        "integer",
+        "smallint",
+        "unsigned smallint",
+        "unsigned int",
+    ):
+        return f"   [{col.column_name}] [int] {nullable}"
+
+    if col.type_name in (
+        "bigint",
+        "unsigned bigint",
+        "long varbit",
+    ):
+        return f"   [{col.column_name}] [bigint] {nullable}"
+
+    if col.type_name in (
+        "date",
+        "timestamp",
+        "timestamp with time zone",
+    ):
+        return f"   [{col.column_name}] [datetime] {nullable}"
+
+    if col.type_name in (
+        "decimal",
+        "float",
+        "double",
+        "numeric",
+    ):
+        return f"   [{col.column_name}] [decimal](28,8) {nullable}"
+
+    if col.type_name in ("time"):
+        return f"   [{col.column_name}] [time] {nullable}"
+
+    return ", ".join(list(map(str, col)))
+
+
 def convert_desc2(col):
     return ", ".join(list(map(str, col)))
 
@@ -62,9 +130,7 @@ def pkeys():
             if res.get(t) is None:
                 res[t] = []
             res[t].append(
-                f"   CONSTRAINT [{t}$0] PRIMARY KEY CLUSTERED (["
-                + "], [".join([c.column_name for c in cols])
-                + "])"
+                f"   CONSTRAINT [{t}$0] PRIMARY KEY CLUSTERED ([" + "], [".join([c.column_name for c in cols]) + "])"
             )
             crsr.cancel()
         except pyodbc.Error as e:

database/db_to_parquet.py

@@ -1,15 +1,16 @@
-from database import conn_string
-import pandas as pd
-import numpy as np
 import fastparquet
-from sqlalchemy import create_engine, inspect, schema, Table
+import numpy as np
+import pandas as pd
+from sqlalchemy import Table, create_engine, inspect, schema
+
+from database import conn_string
 
 # Copied from pandas with modifications
 
 
 def __get_dtype(column, sqltype):
     import sqlalchemy.dialects as sqld
-    from sqlalchemy.types import Integer, Float, Boolean, DateTime, Date, TIMESTAMP
+    from sqlalchemy.types import TIMESTAMP, Boolean, Date, DateTime, Float, Integer
 
     if isinstance(sqltype, Float):
         return float

database/model.py

@@ -29,55 +29,67 @@ class DsnConfig:
             ]
         )
 
-
-class DatabaseInspect:
-    _cursor: pyodbc.Cursor = None
-    _sqlalchemy_engine: Engine = None
-
-    def __init__(self, dsn: DsnConfig, source=False):
-        self.dsn = dsn
-        self.type = "SOURCE" if source else "DEST"
-
     @property
-    def conn_string(self) -> str:
-        if self.dsn.driver == "mssql":
+    def conn_string_pyodbc(self) -> str:
+        if self.driver == "mssql":
             return ";".join(
                 [
                     "Driver={SQL Server Native Client 11.0}",
-                    f"Server={self.dsn.server}",
-                    f"Database={self.dsn.database}",
-                    f"Uid={self.dsn.user}",
-                    f"Pwd={self.dsn.password}",
+                    f"Server={self.server}",
+                    f"Database={self.database}",
+                    f"Uid={self.user}",
+                    f"Pwd={self.password}",
                 ]
             )
-        if self.dsn.driver == "mysql":
-            return f"mysql+pymysql://{self.dsn.user}:{self.dsn.password}@{self.dsn.server}/{self.dsn.database}?charset=utf8mb4"
+        if self.driver == "mysql":
+            return f"mysql+pymysql://{self.user}:{self.password}@{self.server}/{self.database}?charset=utf8mb4"
         return ";".join(
             [
                 "Driver={PostgreSQL Unicode}",
-                f"Server={self.dsn.server}",
+                f"Server={self.server}",
                 "Port=5432",
-                f"Database={self.dsn.database}",
-                f"Uid={self.dsn.user}",
-                f"Pwd={self.dsn.password}",
+                f"Database={self.database}",
+                f"Uid={self.user}",
+                f"Pwd={self.password}",
             ]
         )
-        # f"DSN={self.dsn.server};UID={self.dsn.user};PWD={self.dsn.password}"
+        # f"DSN={self.server};UID={self.user};PWD={self.password}"
 
     @property
     def conn_string_sqlalchemy(self) -> str:
-        if self.dsn.driver == "mssql":
+        if self.driver == "mssql":
             return (
-                f"mssql+pyodbc://{self.dsn.user}:{self.dsn.password}@{self.dsn.server}/{self.dsn.database}?"
+                f"mssql+pyodbc://{self.user}:{self.password}@{self.server}/{self.database}?"
                 "driver=SQL+Server+Native+Client+11.0"
             )
-        if self.dsn.driver == "mysql":
-            return f"mysql+pymysql://{self.dsn.user}:{self.dsn.password}@{self.dsn.server}/{self.dsn.database}?charset=utf8mb4"
-        return f"pyodbc://{self.dsn.user}:{self.dsn.password}@{self.dsn.server}/{self.dsn.database}?driver={self.dsn.driver}"
+        if self.driver == "mysql":
+            return f"mysql+pymysql://{self.user}:{self.password}@{self.server}/{self.database}?charset=utf8mb4"
+        return f"pyodbc://{self.user}:{self.password}@{self.server}/{self.database}?driver={self.driver}"
+
+    @property
+    def bcp_conn_params(self) -> str:
+        return f"-S {self.server} -d {self.database} -U {self.user} -P {self.password}"
+
+
+class DatabaseInspect:
+    _cursor: pyodbc.Cursor = None
+    _sqlalchemy_engine: Engine = None
+
+    def __init__(self, dsn: DsnConfig, source=False):
+        self.dsn = dsn
+        self.type = "SOURCE" if source else "DEST"
+
+    @property
+    def conn_string(self) -> str:
+        return self.dsn.conn_string_pyodbc
+
+    @property
+    def conn_string_sqlalchemy(self) -> str:
+        return self.dsn.conn_string_sqlalchemy
 
     @property
     def bcp_conn_params(self) -> str:
-        return f"-S {self.dsn.server} -d {self.dsn.database} -U {self.dsn.user} -P {self.dsn.password}"
+        return self.dsn.bcp_conn_params
 
     @property
     def cursor(self) -> pyodbc.Cursor:

db.py

@@ -23,6 +23,17 @@ def compare(config_file: str):
 
 @app.command()
 def run(config_file: str, increment: str = "1", max: int = 5):
+    """
+    Executes a database operation using a specified configuration file.
+
+    Args:
+        config_file (str): The name of the configuration file (without extension) to use, located in the SQL config directory.
+        increment (str, optional): Determines whether to increment; treated as True if "1", otherwise False. Defaults to "1".
+        max (int, optional): The maximum number of operations to perform. Defaults to 5.
+
+    Returns:
+        None
+    """
     config_file = cfg.system_dir + f"\\SQL\\config\\{config_file}.json"
     database.run(config_file, increment == "1", max)
 
@@ -50,7 +61,9 @@ def schema():
 
 @app.command()
 def bcp_log():
-    logs_dir = cfg.system_dir + "\\SQL\\temp"
+    logs_dir = cfg.system_dir + "\\SQL\\logs"
+    if not Path(logs_dir).exists():
+        logs_dir = cfg.system_dir + "\\SQL\\temp"
     output_file = cfg.log_dir + "\\bcp.csv.log"
     database.bcp_log(logs_dir, output_file)