import pandas as pd
import pyodbc
from sqlalchemy import create_engine # pip install SQLAlchemy
from sqlalchemy.engine import URL
import pypyodbc # pip install pypyodbc
# Database connection settings
server = 'serverName'
database = 'DB_Name'
username = 'un'
password = 'pw'
from sqlalchemy.types import (
Boolean,
Date,
DateTime,
Float,
Integer,
SmallInteger,
String,
Unicode,
BigInteger,
)
def get_sqlalchemy_types(df_types):
# dictionary mapping pandas dtypes to SQLAlchemy types
dtype_to_sqlalchemy_type = {
# 'int64': BigInteger,
# 'float64': Float,
'object': Unicode(length=255),
# 'datetime64[ns]': DateTime
}
# create an empty dictionary to store column names and SQLAlchemy types
col_dict = {}
# loop through columns and add name and type to dictionary
for col in df_types.columns:
col_dtype = str(df_types[col].dtype)
if col_dtype in dtype_to_sqlalchemy_type.keys():
sqlalchemy_type = dtype_to_sqlalchemy_type[col_dtype]
col_dict[col] = sqlalchemy_type
return col_dict
conn_str = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={server};
DATABASE={database};UID={username};PWD={password}'
conn = pyodbc.connect(conn_str)
connection_url = URL.create('mssql+pyodbc', query={'odbc_connect': conn_str})
enigne = create_engine(connection_url, module=pypyodbc)
result.to_sql('tablename', enigne, if_exists='append', index=False, schema='dbo',
dtype=get_sqlalchemy_types(result),
chunksize=100)
conn.close()
No comments:
Post a Comment