File size: 2,819 Bytes
9ab0aad d4bd626 9ab0aad d4bd626 add59ac 9ab0aad d4bd626 9ab0aad c5cf00a 9ab0aad add59ac 9ab0aad 2a06aec 9ab0aad 2a06aec 9ab0aad d4bd626 9ab0aad d4bd626 9ab0aad d4bd626 9ab0aad a400f6e 9ab0aad dddb5bd 9ab0aad dddb5bd |
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 104 |
from sqlalchemy import (
create_engine,
MetaData,
Table,
Column,
String,
Integer,
Float,
insert,
text,
)
# Use a persistent SQLite database file
engine = create_engine("sqlite:///database.db")
metadata_obj = MetaData()
def create_dynamic_table(df):
"""
Creates a table dynamically based on DataFrame schema.
Args:
df: pandas DataFrame containing the data
Returns:
SQLAlchemy Table object
"""
# Drop existing table if it exists
if 'data_table' in metadata_obj.tables:
metadata_obj.remove(metadata_obj.tables['data_table'])
# Create columns based on DataFrame dtypes
columns = []
for col_name, dtype in df.dtypes.items():
if 'int' in str(dtype):
col_type = Integer
elif 'float' in str(dtype):
col_type = Float
else:
col_type = String(255) # Using a generous length for string columns
# First column becomes primary key
if len(columns) == 0:
columns.append(Column(col_name, col_type, primary_key=True))
else:
columns.append(Column(col_name, col_type))
# Create new table
table = Table('data_table', metadata_obj, *columns)
# Create table in database
metadata_obj.create_all(engine, tables=[table])
return table
def clear_database():
"""
Removes all tables from the database.
"""
metadata_obj.drop_all(engine)
metadata_obj.clear()
def insert_rows_into_table(rows, table):
"""
Inserts rows into the specified table.
Args:
rows: List of dictionaries containing the row data
table: SQLAlchemy Table object
"""
with engine.begin() as connection:
connection.execute(insert(table), rows)
def get_table_schema():
"""
Returns the current table schema as a string.
"""
try:
# Get list of tables
with engine.connect() as con:
tables = con.execute(text(
"SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
)).fetchall()
if not tables:
return None
# Use the first table found
table_name = tables[0][0]
# Get column information
with engine.connect() as con:
columns = con.execute(text(f"PRAGMA table_info({table_name})")).fetchall()
schema = []
for col in columns:
col_name = col[1]
col_type = col[2]
is_primary = "primary key" if col[5] == 1 else ""
schema.append(f"- {col_name} ({col_type}) {is_primary}".strip())
return "\n".join(schema)
except Exception as e:
return None |