def csv_to_sqlite(
    input_filename,
    output_filename,
    samples=None,
    dialect=None,
    batch_size=10000,
    encoding="utf-8",
    callback=None,
    force_types=None,
    chunk_size=8388608,
    table_name="table1",
    schema=None,
):
    "Export a CSV file to SQLite, based on field type detection from samples"

    # TODO: automatically detect encoding if encoding == `None`
    # TODO: should be able to specify fields
    # TODO: if table_name is "2019" the final name will be "field_2019" - must
    #       be "table_2019"
    # TODO: if schema is provided and the names are in uppercase, this function
    #       will fail

    if dialect is None:  # Get a sample to detect dialect
        fobj = open_compressed(input_filename, mode="rb")
        sample = fobj.read(chunk_size)
        dialect = rows.plugins.csv.discover_dialect(sample, encoding=encoding)
    elif isinstance(dialect, six.text_type):
        dialect = csv.get_dialect(dialect)

    if schema is None:  # Identify data types
        fobj = open_compressed(input_filename, encoding=encoding)
        data = list(islice(csv.DictReader(fobj, dialect=dialect), samples))
        schema = rows.import_from_dicts(data).fields
        if force_types is not None:
            schema.update(force_types)

    # Create lazy table object to be converted
    # TODO: this lazyness feature will be incorported into the library soon so
    #       we can call here `rows.import_from_csv` instead of `csv.reader`.
    reader = csv.reader(
        open_compressed(input_filename, encoding=encoding), dialect=dialect
    )
    header = make_header(next(reader))  # skip header
    table = rows.Table(fields=OrderedDict([(field, schema[field]) for field in header]))
    table._rows = reader

    # Export to SQLite
    return rows.export_to_sqlite(
        table,
        output_filename,
        table_name=table_name,
        batch_size=batch_size,
        callback=callback,
    )
def csv_to_sqlite ( input_filename , output_filename , samples = None , dialect = None , batch_size = 10000 , encoding = "utf-8" , callback = None , force_types = None , chunk_size = 8388608 , table_name = "table1" , schema = None , ) : if dialect is None : fobj = open_compressed ( input_filename , mode = "rb" ) sample = fobj . read ( chunk_size ) dialect = rows . plugins . csv . discover_dialect ( sample , encoding = encoding ) elif isinstance ( dialect , six . text_type ) : dialect = csv . get_dialect ( dialect ) if schema is None : fobj = open_compressed ( input_filename , encoding = encoding ) data = list ( islice ( csv . DictReader ( fobj , dialect = dialect ) , samples ) ) schema = rows . import_from_dicts ( data ) . fields if force_types is not None : schema . update ( force_types ) reader = csv . reader ( open_compressed ( input_filename , encoding = encoding ) , dialect = dialect ) header = make_header ( next ( reader ) ) table = rows . Table ( fields = OrderedDict ( [ ( field , schema [ field ] ) for field in header ] ) ) table . _rows = reader return rows . export_to_sqlite ( table , output_filename , table_name = table_name , batch_size = batch_size , callback = callback , )
Export a CSV file to SQLite, based on field type detection from samples