File size: 3,017 Bytes
5875608 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 |
import pandas as pd
import sqlite3
def csv_to_sqlite(csv_file, db_name, table_name):
# Read the CSV file into a pandas DataFrame
df = pd.read_csv(csv_file)
# Connect to the SQLite database (it will create the database file if it doesn't exist)
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
# Infer the schema based on the DataFrame columns and data types
def create_table_from_df(df, table_name):
# Get column names and types
col_types = []
for col in df.columns:
dtype = df[col].dtype
if dtype == "int64":
col_type = "INTEGER"
elif dtype == "float64":
col_type = "REAL"
else:
col_type = "TEXT"
col_types.append(f'"{col}" {col_type}')
# Create the table schema
col_definitions = ", ".join(col_types)
create_table_query = (
f"CREATE TABLE IF NOT EXISTS {table_name} ({col_definitions});"
)
# print(create_table_query)
# Execute the table creation query
cursor.execute(create_table_query)
print(f"Table '{table_name}' created with schema: {col_definitions}")
# Create table schema
create_table_from_df(df, table_name)
# Insert CSV data into the SQLite table
df.to_sql(table_name, conn, if_exists="replace", index=False)
# Commit and close the connection
conn.commit()
conn.close()
print(f"Data loaded into '{table_name}' table in '{db_name}' SQLite database.")
def run_sql_query(db_name, query):
"""
Executes a SQL query on a SQLite database and returns the results.
Args:
db_name (str): The name of the SQLite database file.
query (str): The SQL query to run.
Returns:
list: Query result as a list of tuples, or an empty list if no results or error occurred.
"""
try:
# Connect to the SQLite database
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
# Execute the SQL query
cursor.execute(query)
# Fetch all results
results = cursor.fetchall()
# Close the connection
conn.close()
# Return results or an empty list if no results were found
return results if results else []
except sqlite3.Error as e:
print(f"An error occurred while executing the query: {e}")
return []
def get_table_schema(db_name, table_name):
"""
Retrieves the schema (columns and data types) for a given table in the SQLite database.
Args:
db_name (str): The name of the SQLite database file.
table_name (str): The name of the table.
Returns:
list: A list of tuples with column name, data type, and other info.
"""
conn = sqlite3.connect(db_name)
cursor = conn.cursor()
# Use PRAGMA to get the table schema
cursor.execute(f"PRAGMA table_info({table_name});")
schema = cursor.fetchall()
conn.close()
return schema
|