def pgimport(
    filename,
    database_uri,
    table_name,
    encoding="utf-8",
    dialect=None,
    create_table=True,
    schema=None,
    callback=None,
    timeout=0.1,
    chunk_size=8388608,
    max_samples=10000,
):
    """Import data from CSV into PostgreSQL using the fastest method

    Required: psql command
    """

    fobj = open_compressed(filename, mode="r", encoding=encoding)
    sample = fobj.read(chunk_size)

    if dialect is None:  # Detect dialect
        dialect = rows.plugins.csv.discover_dialect(
            sample.encode(encoding), encoding=encoding
        )
    elif isinstance(dialect, six.text_type):
        dialect = csv.get_dialect(dialect)

    if schema is None:
        # Detect field names
        reader = csv.reader(io.StringIO(sample), dialect=dialect)
        field_names = [slug(field_name) for field_name in next(reader)]

    else:
        field_names = list(schema.keys())

    if create_table:
        if schema is None:
            data = [
                dict(zip(field_names, row))
                for row in itertools.islice(reader, max_samples)
            ]
            table = rows.import_from_dicts(data)
            field_types = [table.fields[field_name] for field_name in field_names]
        else:
            field_types = list(schema.values())

        columns = [
            "{} {}".format(name, POSTGRESQL_TYPES.get(type_, DEFAULT_POSTGRESQL_TYPE))
            for name, type_ in zip(field_names, field_types)
        ]
        create_table = SQL_CREATE_TABLE.format(
            table_name=table_name, field_types=", ".join(columns)
        )
        execute_command(get_psql_command(create_table, database_uri=database_uri))

    # Prepare the `psql` command to be executed based on collected metadata
    command = get_psql_copy_command(
        database_uri=database_uri,
        dialect=dialect,
        direction="FROM",
        encoding=encoding,
        header=field_names,
        table_name=table_name,
    )
    rows_imported, error = 0, None
    fobj = open_compressed(filename, mode="rb")
    try:
        process = subprocess.Popen(
            shlex.split(command),
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        data = fobj.read(chunk_size)
        total_written = 0
        while data != b"":
            written = process.stdin.write(data)
            total_written += written
            if callback:
                callback(written, total_written)
            data = fobj.read(chunk_size)
        stdout, stderr = process.communicate()
        if stderr != b"":
            raise RuntimeError(stderr.decode("utf-8"))
        rows_imported = int(stdout.replace(b"COPY ", b"").strip())

    except FileNotFoundError:
        raise RuntimeError("Command `psql` not found")

    except BrokenPipeError:
        raise RuntimeError(process.stderr.read().decode("utf-8"))

    return {"bytes_written": total_written, "rows_imported": rows_imported}
def pgimport ( filename , database_uri , table_name , encoding = "utf-8" , dialect = None , create_table = True , schema = None , callback = None , timeout = 0.1 , chunk_size = 8388608 , max_samples = 10000 , ) : fobj = open_compressed ( filename , mode = "r" , encoding = encoding ) sample = fobj . read ( chunk_size ) if dialect is None : dialect = rows . plugins . csv . discover_dialect ( sample . encode ( encoding ) , encoding = encoding ) elif isinstance ( dialect , six . text_type ) : dialect = csv . get_dialect ( dialect ) if schema is None : reader = csv . reader ( io . StringIO ( sample ) , dialect = dialect ) field_names = [ slug ( field_name ) for field_name in next ( reader ) ] else : field_names = list ( schema . keys ( ) ) if create_table : if schema is None : data = [ dict ( zip ( field_names , row ) ) for row in itertools . islice ( reader , max_samples ) ] table = rows . import_from_dicts ( data ) field_types = [ table . fields [ field_name ] for field_name in field_names ] else : field_types = list ( schema . values ( ) ) columns = [ "{} {}" . format ( name , POSTGRESQL_TYPES . get ( type_ , DEFAULT_POSTGRESQL_TYPE ) ) for name , type_ in zip ( field_names , field_types ) ] create_table = SQL_CREATE_TABLE . format ( table_name = table_name , field_types = ", " . join ( columns ) ) execute_command ( get_psql_command ( create_table , database_uri = database_uri ) ) command = get_psql_copy_command ( database_uri = database_uri , dialect = dialect , direction = "FROM" , encoding = encoding , header = field_names , table_name = table_name , ) rows_imported , error = 0 , None fobj = open_compressed ( filename , mode = "rb" ) try : process = subprocess . Popen ( shlex . split ( command ) , stdin = subprocess . PIPE , stdout = subprocess . PIPE , stderr = subprocess . PIPE , ) data = fobj . read ( chunk_size ) total_written = 0 while data != b"" : written = process . stdin . write ( data ) total_written += written if callback : callback ( written , total_written ) data = fobj . read ( chunk_size ) stdout , stderr = process . communicate ( ) if stderr != b"" : raise RuntimeError ( stderr . decode ( "utf-8" ) ) rows_imported = int ( stdout . replace ( b"COPY " , b"" ) . strip ( ) ) except FileNotFoundError : raise RuntimeError ( "Command `psql` not found" ) except BrokenPipeError : raise RuntimeError ( process . stderr . read ( ) . decode ( "utf-8" ) ) return { "bytes_written" : total_written , "rows_imported" : rows_imported }
Import data from CSV into PostgreSQL using the fastest method

    Required: psql command