Method fetchSingle

api/datasets.go:200–318

(hash string, dest *sql.DB, dataSource *config.DataSource)

}

func (api *API) fetchSingle(hash string, dest *sql.DB, dataSource *config.DataSource) error {
	dataConnection, err := api.ReadDataConnection(hash, dataSource.DataConnection)
	if err != nil {
		return err
	}
	switch dataConnection.Type {
	case "csv":
		f, err := api.readFile(dataConnection.Path)
		if err != nil {
			return err
		}
		r := csv.NewReader(f)
		firstLine, err := r.Read()
		if err != nil {
			return err
		}
		columns := firstLine
		for i := range columns {
			columns[i] = strconv.Quote(columns[i])
		}

		_, err = dest.Exec(fmt.Sprintf("CREATE TABLE %s (%s)", dataSource.ID, strings.Join(columns, ",")))
		if err != nil {
			return err
		}

		params := []string{}
		for i := range columns {
			params = append(params, fmt.Sprintf("$%d", i+1))
		}
		stmt, err := dest.Prepare(fmt.Sprintf("INSERT INTO %s VALUES (%s)", dataSource.ID, strings.Join(params, ",")))
		if err != nil {
			return err
		}
		defer stmt.Close()

		for {
			record, err := r.Read()
			if err != nil {
				if err == io.EOF {
					return nil
				}
				return err
			}
			if len(record) != len(columns) {
				return errors.New("inconsistent number of fields")
			}
			values := make([]interface{}, len(record))
			for i := range record {
				values[i] = record[i]
			}
			_, err = stmt.Exec(values...)
			if err != nil {
				return err
			}
		}
	case "postgres":
		db, err := sql.Open(dataConnection.Type, dataConnection.ConnectionString)