Create Test Dataframe
from pandas import util
df = util.testing.makeDataFrame()
df = util.testing.makeMissingDataframe()
df = util.testing.makeTimeDataFrame()
df = util.testing.makeMixedDataFrame()
df = util.testing.makePeriodFrame()
DataFrame To SQL
import inspect
from pandas import util
from sqlalchemy import create_engine
import psycopg2
def var_name_as_str(var):
"""
Gets the name of var. Does it from the out most frame inner-wards.
:param var: variable to get name from.
:return: string
"""
for fi in reversed(inspect.stack()):
names = [var_name for var_name, var_val in fi.frame.f_locals.items() if var_val is var]
if len(names) > 0:
return names[0]
def write_table(df, engine):
var_name = var_name_as_str(df)
df.to_sql(var_name, engine)
def drop_table(table, conn, commit=False):
curs = conn.cursor()
try:
curs.execute(f"DROP TABLE {table};")
except psycopg2.errors.UndefinedTable:
return
if commit: conn.commit()
pg_conn = lambda: psycopg2.connect(
user = "draftsharks",
password = "draftsharks",
host = "localhost",
port = "5432",
database = "sip"
)
def drop_write_and_close(df, engine):
conn = pg_conn()
drop_table(var_name_as_str(df), conn, True)
write_table(df, engine)
conn.commit()
conn.close()
if __name__ == "__main__":
engine = create_engine('postgresql://draftsharks:draftsharks@localhost:5432/sip')
my_df = util.testing.makeDataFrame()
drop_write_and_close(my_df, engine)
engine.dispose()