erster Testlauf csv-Export aus SQL-Datenbank im Impromptu-Format

database/csv_export.py

@@ -0,0 +1,50 @@
+import csv
+
+import pandas as pd
+from sqlalchemy import create_engine
+
+from database import conn_string
+
+
+def convert_data(element):
+    txt = str(element)
+    txt = txt.replace("None", "")
+    txt = txt.replace("False", "0").replace("True", "1")
+    txt = txt.replace("\t", "").replace("\r", "").replace("\n", "")
+    txt = txt.replace("\x81", "").replace("\x90", "")
+
+    txt = "" if txt in ["nan", "NaT"] else txt
+    return txt
+
+
+def table_to_csv(table_name: str, csv_file: str, dsn: str):
+    try:
+        conn = create_engine(dsn).connect().execution_options(stream_results=True)
+        df = pd.read_sql(f"SELECT * FROM {table_name}", conn, chunksize=1000)
+    except Exception as e:
+        print(e.args[1])
+
+    with open(csv_file, "w", encoding="latin-1", errors="replace", newline="") as fwh:
+        print("Kopiervorgang wird gestartet...")
+        i = 0
+        for chunk in df:
+            chunk.to_csv(fwh, ";", index=False, decimal=",", quoting=csv.QUOTE_NONNUMERIC)
+            i += chunk.shape[0]
+            if chunk.shape[0] == 1000:
+                print(f"1000 Zeilen zum SQL Server gesendet. Insgesamt gesendet: {i}")
+        print("")
+        print(f"{i} Zeilen kopiert.")
+
+
+if __name__ == "__main__":
+    dsn = {
+        "user": "sa",
+        "password": "Mffu3011#",
+        "server": "localhost\\GLOBALCUBE",
+        "database": "OPTIMA",
+        "driver": "mssql",
+        "schema": "import",
+    }
+    conn_str = conn_string(dsn)
+    table_to_csv("load.Aftersales_Rechnungen_ben_AW_final", "Aftersales_Rechnungen_ben_AW_final.csv", conn_str)
+    # print(timeit.timeit(s))