gctools/database/csv_export.py

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))