Generate a CTE from a local csv file#
import pandas as pd
replace_nan = lambda x: x.replace('nan', 'null')
def df_to_values(df, columns=None, replace_nans=True):
if columns is None:
columns = df.columns.tolist()
newdata = str(list(df[columns].to_records(
index=False))
)[1:-1]
if replace_nans:
newdata = replace_nan(newdata)
return newdata
def cte_from_csv(localfile, colgroups, cte_names, head=False):
df = pd.read_csv(localfile)
if head:
df = df.head()
return 'with ' + ', '.join([
f'''
{cte_names[i]}({', '.join(colgroups[i])}) as (
VALUES {df_to_values(df, columns=colgroups[i], replace_nans=True)}
)
'''
for i, _ in enumerate(colgroups)
])
temp.csv#
one,two,three
1,2.,3.3
,2.3,3.5
11,.22,.003
Example#
loc = "temp.csv"
print( cte_from_csv(loc, [['one', 'two', 'three']], ['foo'],))
with
foo(one, two, three) as (
VALUES (1., 2., 3.3), (null, 2.3, 3.5), (11., 0.22, 0.003)
)
Dollar encode#
def doll_df_to_values(df, cols,
cols_to_dollar_encode=None):
data = [tuple(x) for x in df[cols].to_records(index=False)]
vec = ['(' + ', '.join([enc(x[i], cols[i] in cols_to_dollar_encode)
for (i, _) in enumerate(x)]) + ')' for x in data]
return ', '.join(vec)
def enc(x, dollar_enc):
if dollar_enc:
return f'$${x}$$'
elif isinstance(x, str):
return f"'{x}'"
elif np.isnan(x):
return 'null'
else:
return f"{x}"