import micropip
import pandas as pd
import numpy as np
import os
sz = 100
np.random.seed(42)
df = pd.DataFrame({"A": np.random.randn(sz), "B": [1] * sz})
df
def test_csv_write(df):
df.to_csv("data/test.csv")
def test_csv_read():
return pd.read_csv("data/test.csv")
test_csv_write(df)
test_csv_read()
def test_json_write(df):
df.to_json("test_json.json")
def test_json_read():
return pd.read_json("test_json.json")
test_json_write(df)
test_json_read()
await micropip.install("html5lib")
await micropip.install("lxml")
def test_html_write(df):
df.to_html("./test_html.html")
def test_html_read():
return pd.read_html("./test_html.html")[0]
test_html_write(df)
test_html_read()
def test_xml_write(df):
df.to_xml("test_xml.xml")
def test_xml_read():
return pd.read_xml("test_xml.xml")
test_xml_write(df)
test_xml_read()
await micropip.install("openpyxl")
def test_xlsx_write(df):
df.to_excel("test_xlsx.xlsx")
def test_xlsx_read():
return pd.read_excel("test_xlsx.xlsx")
test_xlsx_write(df)
test_xlsx_read()
import sqlite3
import os
def test_sql_write(df):
if os.path.exists("test.sql"):
os.remove("test.sql")
sql_db = sqlite3.connect("test.sql")
df.to_sql(name="test_table", con=sql_db)
sql_db.close()
def test_sql_read():
sql_db = sqlite3.connect("test.sql")
df = pd.read_sql_query("select * from test_table", sql_db)
sql_db.close()
return df
test_sql_write(df)
test_sql_read()
await micropip.install("tables")
def test_hdf_fixed_write(df):
df.to_hdf("test_fixed.hdf", "test", mode="w")
def test_hdf_fixed_read():
return pd.read_hdf("test_fixed.hdf", "test")
test_hdf_fixed_write(df)
test_hdf_fixed_read()
def test_hdf_fixed_write_compress(df):
df.to_hdf("test_fixed_compress.hdf", "test", mode="w", complib="blosc")
def test_hdf_fixed_read_compress():
return pd.read_hdf("test_fixed_compress.hdf", "test")
test_hdf_fixed_write_compress(df)
test_hdf_fixed_read_compress()
def test_hdf_table_write(df):
df.to_hdf("test_table.hdf", "test", mode="w", format="table")
def test_hdf_table_read():
return pd.read_hdf("test_table.hdf", "test")
test_hdf_table_write(df)
test_hdf_table_read()
def test_hdf_table_write_compress(df):
df.to_hdf(
"test_table_compress.hdf", "test", mode="w", complib="blosc", format="table"
)
def test_hdf_table_read_compress():
return pd.read_hdf("test_table_compress.hdf", "test")
test_hdf_table_write_compress(df)
test_hdf_table_read_compress()
await micropip.install("pyarrow")
def test_feather_write(df):
df.to_feather("test.feather")
def test_feather_read():
return pd.read_feather("test.feather")
test_feather_write(df)
test_feather_read()
def test_pickle_write(df):
df.to_pickle("test.pkl")
def test_pickle_read():
return pd.read_pickle("test.pkl")
test_pickle_write(df)
test_pickle_read()
def test_pickle_write_compress(df):
df.to_pickle("test.pkl.compress", compression="xz")
def test_pickle_read_compress():
return pd.read_pickle("test.pkl.compress", compression="xz")
test_pickle_write_compress(df)
test_pickle_read_compress()
def test_parquet_write(df):
df.to_parquet("test.parquet")
def test_parquet_read():
return pd.read_parquet("test.parquet")
test_parquet_write(df)
test_parquet_read()