in src/dma/collector/workflows/base.py [0:0]
def import_to_table(self, data: dict[str, list[dict]]) -> None:
"""Load a dictionary of result sets into duckdb.
The key of the dictionary becomes the table name in the database.
"""
for table_name, table_data in data.items():
if len(table_data) > 0:
column_names = table_data[0].keys()
self.local_db.register(
f"obj_{table_name}", pl.from_dicts(table_data, strict=False, infer_schema_length=10000)
)
self.local_db.execute(
f"insert into {table_name}({', '.join(column_name for column_name in column_names)}) select {', '.join(column_name for column_name in column_names)} from obj_{table_name}" # noqa: S608
)
self.local_db.execute(f"drop view obj_{table_name}")