ich versuche mit dem folgenden script ein Pandas DF mit 9000 rows und 13 spalten in eine MS SQL Tabelle zu übergeben.
Grundsätzlich tut es auch was es soll dauert aber bis zu 6 min, also gehe ich wohl irgendetwas grundsätzlich falsch an.
Ich hoffe jemand kann mir einen Tipp geben. Mit GTP bin ich schon gescheitert
Code: Alles auswählen
from dataclasses import dataclass
from typing import Dict, Any, Iterable
from pandas import DataFrame
from sqlalchemy import create_engine, inspect
import urllib
import pandas as pd
import pyodbc
@dataclass(frozen=True)
class ConnectionSettings:
"""Connection Settings."""
server: str
database: str
username: str
password: str
#driver: str = '{ODBC Driver 18 for SQL Server}'
#driver: str = '{ODBC Driver 13 for SQL Server}'
driver: str = '{ODBC Driver 18 for SQL Server}'
timeout: int = 30
class AzureDbConnection:
db = None
"""
Azure SQL database connection.
"""
def __init__(self, conn_settings: ConnectionSettings, echo: bool = False) -> None:
conn_params = urllib.parse.quote_plus(
'Driver=%s;' % conn_settings.driver +
'Server=tcp:%s.database.windows.net,1433;' % conn_settings.server +
'Database=%s;' % conn_settings.database +
'Uid=%s;' % conn_settings.username +
'Pwd=%s;' % conn_settings.password +
'Encrypt=yes;' +
'TrustServerCertificate=no;' +
'Connection Timeout=%s;' % conn_settings.timeout
)
conn_string = f'mssql+pyodbc:///?odbc_connect={conn_params}'
#self.db = create_engine(conn_string, echo=echo)
AzureDbConnection.db = create_engine(conn_string, echo=echo)
def connect(self) -> None:
"""Estimate connection."""
self.conn = self.db.connect()
def get_tables(self) -> Iterable[str]:
"""Get list of tables."""
inspector = inspect(self.db)
return [t for t in inspector.get_table_names()]
def dispose(self) -> None:
"""Dispose opened connections."""
self.conn.close()
self.db.dispose()
def execute(self, query: str) -> None:
"""Execute query."""
self.conn.execute(query)
class SQL_TabellenLadenBearbeiten:
'''Ermöglicht die Auswahl von Spalten aus bestimmten Tabellen
sowie Datumsbereich
und gibt diese als DataFrame zurück'''
def verbinder():
conn_settings = ConnectionSettings(
server = 'XXXXXXX',
database= 'XXXX',
username='XXX',
password='XXXX')
db_conn = AzureDbConnection(conn_settings)
return db_conn
def sql_test(tabellenName, df):
# truncate
# Einfügen der Werte
db_conn = SQL_TabellenLadenBearbeiten.verbinder()
db_conn.connect()
with AzureDbConnection.db.begin() as connection:
connection.execute(f"TRUNCATE TABLE [{tabellenName}]")
df.to_sql(tabellenName, connection, if_exists='append', index=False, chunksize=1000)
return print(f'Tabelle {tabellenName} wurde geleert und neu befüllt')