Importar un archivo CSV en una tabla de base de datos sqlite3 usando Python

Tengo un archivo CSV y quiero importar este archivo de forma masiva a mi base de datos sqlite3 usando Python. el comando es “.import …..”. Pero parece que no puede funcionar así. ¿Puede alguien darme un ejemplo de cómo hacerlo en sqlite3? Estoy usando ventanas por si acaso. Gracias

import csv, sqlite3 con = sqlite3.connect(":memory:") cur = con.cursor() cur.execute("CREATE TABLE t (col1, col2);") # use your column names here with open('data.csv','rb') as fin: # `with` statement available in 2.5+ # csv.DictReader uses first line in file for column headings by default dr = csv.DictReader(fin) # comma is default delimiter to_db = [(i['col1'], i['col2']) for i in dr] cur.executemany("INSERT INTO t (col1, col2) VALUES (?, ?);", to_db) con.commit() con.close() 

La creación de una conexión sqlite a un archivo en el disco se deja como un ejercicio para el lector … pero ahora la biblioteca de pandas hace posible dos líneas.

 df = pandas.read_csv(csvfile) df.to_sql(table_name, conn, if_exists='append', index=False) 

El comando .import es una característica de la herramienta de línea de comandos sqlite3. Para hacerlo en Python, simplemente debe cargar los datos utilizando las facilidades que Python tiene, como el módulo csv , e insertar los datos como de costumbre.

De esta manera, también tiene control sobre los tipos que se insertan, en lugar de confiar en el comportamiento aparentemente no documentado de sqlite3.

Mis 2 centavos (más generics):

 import csv, sqlite3 import logging def _get_col_datatypes(fin): dr = csv.DictReader(fin) # comma is default delimiter fieldTypes = {} for entry in dr: feildslLeft = [f for f in dr.fieldnames if f not in fieldTypes.keys()] if not feildslLeft: break # We're done for field in feildslLeft: data = entry[field] # Need data to decide if len(data) == 0: continue if data.isdigit(): fieldTypes[field] = "INTEGER" else: fieldTypes[field] = "TEXT" # TODO: Currently there's no support for DATE in sqllite if len(feildslLeft) > 0: raise Exception("Failed to find all the columns data types - Maybe some are empty?") return fieldTypes def escapingGenerator(f): for line in f: yield line.encode("ascii", "xmlcharrefreplace").decode("ascii") def csvToDb(csvFile, outputToFile = False): # TODO: implement output to file with open(csvFile,mode='r', encoding="ISO-8859-1") as fin: dt = _get_col_datatypes(fin) fin.seek(0) reader = csv.DictReader(fin) # Keep the order of the columns name just as in the CSV fields = reader.fieldnames cols = [] # Set field and type for f in fields: cols.append("%s %s" % (f, dt[f])) # Generate create table statement: stmt = "CREATE TABLE ads (%s)" % ",".join(cols) con = sqlite3.connect(":memory:") cur = con.cursor() cur.execute(stmt) fin.seek(0) reader = csv.reader(escapingGenerator(fin)) # Generate insert statement: stmt = "INSERT INTO ads VALUES(%s);" % ','.join('?' * len(cols)) cur.executemany(stmt, reader) con.commit() return con 

Muchas gracias por la respuesta de bernie! Tuve que modificarlo un poco, esto es lo que funcionó para mí:

 import csv, sqlite3 conn = sqlite3.connect("pcfc.sl3") curs = conn.cursor() curs.execute("CREATE TABLE PCFC (id INTEGER PRIMARY KEY, type INTEGER, term TEXT, definition TEXT);") reader = csv.reader(open('PC.txt', 'r'), delimiter='|') for row in reader: to_db = [unicode(row[0], "utf8"), unicode(row[1], "utf8"), unicode(row[2], "utf8")] curs.execute("INSERT INTO PCFC (type, term, definition) VALUES (?, ?, ?);", to_db) conn.commit() 

Mi archivo de texto (PC.txt) se ve así:

 1 | Term 1 | Definition 1 2 | Term 2 | Definition 2 3 | Term 3 | Definition 3 
 #!/usr/bin/python # -*- coding: utf-8 -*- import sys, csv, sqlite3 def main(): con = sqlite3.connect(sys.argv[1]) # database file input cur = con.cursor() cur.executescript(""" DROP TABLE IF EXISTS t; CREATE TABLE t (COL1 TEXT, COL2 TEXT); """) # checks to see if table exists and makes a fresh table. with open(sys.argv[2], "rb") as f: # CSV file input reader = csv.reader(f, delimiter=',') # no header information with delimiter for row in reader: to_db = [unicode(row[0], "utf8"), unicode(row[1], "utf8")] # Appends data from CSV file representing and handling of text cur.execute("INSERT INTO neto (COL1, COL2) VALUES(?, ?);", to_db) con.commit() con.close() # closes connection to database if __name__=='__main__': main() 

Puedes hacer esto usando blaze & odo eficientemente

 import blaze as bz csv_path = 'data.csv' bz.odo(csv_path, 'sqlite:///data.db::data') 

Odo almacenará el archivo csv en data.db (base de datos sqlite) bajo los data esquema

O usas odo directamente, sin blaze . De cualquier manera está bien. Lee esta documentacion

Basado en la solución de Guy L (Love it) pero puede manejar campos escapados.

 import csv, sqlite3 def _get_col_datatypes(fin): dr = csv.DictReader(fin) # comma is default delimiter fieldTypes = {} for entry in dr: feildslLeft = [f for f in dr.fieldnames if f not in fieldTypes.keys()] if not feildslLeft: break # We're done for field in feildslLeft: data = entry[field] # Need data to decide if len(data) == 0: continue if data.isdigit(): fieldTypes[field] = "INTEGER" else: fieldTypes[field] = "TEXT" # TODO: Currently there's no support for DATE in sqllite if len(feildslLeft) > 0: raise Exception("Failed to find all the columns data types - Maybe some are empty?") return fieldTypes def escapingGenerator(f): for line in f: yield line.encode("ascii", "xmlcharrefreplace").decode("ascii") def csvToDb(csvFile,dbFile,tablename, outputToFile = False): # TODO: implement output to file with open(csvFile,mode='r', encoding="ISO-8859-1") as fin: dt = _get_col_datatypes(fin) fin.seek(0) reader = csv.DictReader(fin) # Keep the order of the columns name just as in the CSV fields = reader.fieldnames cols = [] # Set field and type for f in fields: cols.append("\"%s\" %s" % (f, dt[f])) # Generate create table statement: stmt = "create table if not exists \"" + tablename + "\" (%s)" % ",".join(cols) print(stmt) con = sqlite3.connect(dbFile) cur = con.cursor() cur.execute(stmt) fin.seek(0) reader = csv.reader(escapingGenerator(fin)) # Generate insert statement: stmt = "INSERT INTO \"" + tablename + "\" VALUES(%s);" % ','.join('?' * len(cols)) cur.executemany(stmt, reader) con.commit() con.close() 
 import csv, sqlite3 def _get_col_datatypes(fin): dr = csv.DictReader(fin) # comma is default delimiter fieldTypes = {} for entry in dr: feildslLeft = [f for f in dr.fieldnames if f not in fieldTypes.keys()] if not feildslLeft: break # We're done for field in feildslLeft: data = entry[field] # Need data to decide if len(data) == 0: continue if data.isdigit(): fieldTypes[field] = "INTEGER" else: fieldTypes[field] = "TEXT" # TODO: Currently there's no support for DATE in sqllite if len(feildslLeft) > 0: raise Exception("Failed to find all the columns data types - Maybe some are empty?") return fieldTypes def escapingGenerator(f): for line in f: yield line.encode("ascii", "xmlcharrefreplace").decode("ascii") def csvToDb(csvFile,dbFile,tablename, outputToFile = False): # TODO: implement output to file with open(csvFile,mode='r', encoding="ISO-8859-1") as fin: dt = _get_col_datatypes(fin) fin.seek(0) reader = csv.DictReader(fin) # Keep the order of the columns name just as in the CSV fields = reader.fieldnames cols = [] # Set field and type for f in fields: cols.append("\"%s\" %s" % (f, dt[f])) # Generate create table statement: stmt = "create table if not exists \"" + tablename + "\" (%s)" % ",".join(cols) print(stmt) con = sqlite3.connect(dbFile) cur = con.cursor() cur.execute(stmt) fin.seek(0) reader = csv.reader(escapingGenerator(fin)) # Generate insert statement: stmt = "INSERT INTO \"" + tablename + "\" VALUES(%s);" % ','.join('?' * len(cols)) cur.executemany(stmt, reader) con.commit() con.close() 

en aras de la simplicidad, puede utilizar la herramienta de línea de comandos sqlite3 del Makefile de su proyecto.

 %.sql3: %.csv rm -f $@ sqlite3 $@ -echo -cmd ".mode csv" ".import $< $*" %.dump: %.sql3 sqlite3 $< "select * from $*" 

make test.sql3 luego crea la base de datos sqlite desde un archivo test.csv existente, con una sola tabla "test". A continuación, puede make test.dump para verificar los contenidos.