Data Import Guide

import asyncio
import csv
from geode_client import Client

async def import_csv(filename: str):
    client = Client(host="localhost", port=3141, skip_verify=True)

    async with client.connection() as conn:
        # Begin transaction for bulk import
        await conn.begin()

        try:
            with open(filename, 'r') as file:
                reader = csv.DictReader(file)
                row_count = 0

                for row in reader:
                    await conn.execute(
                        """CREATE (:User {
                            id: $id,
                            name: $name,
                            email: $email,
                            age: $age,
                            city: $city
                        })""",
                        {
                            "id": int(row['id']),
                            "name": row['name'],
                            "email": row['email'],
                            "age": int(row['age']),
                            "city": row['city']
                        }
                    )
                    row_count += 1

                    # Commit in batches for large files
                    if row_count % 1000 == 0:
                        await conn.commit()
                        await conn.begin()
                        print(f"Imported {row_count} rows...")

            await conn.commit()
            print(f"Successfully imported {row_count} users")

        except Exception as e:
            await conn.rollback()
            raise e

asyncio.run(import_csv("users.csv"))

use geode_client::{Client, Value};
use std::collections::HashMap;
use std::error::Error;
use std::fs::File;
use csv::ReaderBuilder;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let client = Client::new("127.0.0.1", 3141).skip_verify(true);
    let mut conn = client.connect().await?;

    // Begin transaction
    conn.begin().await?;

    // Open CSV file
    let file = File::open("users.csv")?;
    let mut reader = ReaderBuilder::new()
        .has_headers(true)
        .from_reader(file);

    let mut row_count = 0;

    for result in reader.records() {
        let record = result?;

        let mut params = HashMap::new();
        params.insert("id".to_string(), Value::int(record[0].parse::<i64>()?));
        params.insert("name".to_string(), Value::string(&record[1]));
        params.insert("email".to_string(), Value::string(&record[2]));
        params.insert("age".to_string(), Value::int(record[3].parse::<i64>()?));
        params.insert("city".to_string(), Value::string(&record[4]));

        conn.query_with_params(
            r#"CREATE (:User {
                id: $id,
                name: $name,
                email: $email,
                age: $age,
                city: $city
            })"#,
            &params
        ).await?;

        row_count += 1;

        // Commit in batches
        if row_count % 1000 == 0 {
            conn.commit().await?;
            conn.begin().await?;
            println!("Imported {} rows...", row_count);
        }
    }

    conn.commit().await?;
    println!("Successfully imported {} users", row_count);

    Ok(())
}

import { createClient } from '@geodedb/client';
import { parse } from 'csv-parse/sync';
import { readFileSync } from 'fs';

async function importCSV(filename: string) {
    const client = await createClient('quic://localhost:3141');

    const fileContent = readFileSync(filename, 'utf-8');
    const records = parse(fileContent, {
        columns: true,
        skip_empty_lines: true
    });

    await client.beginTransaction();

    try {
        let rowCount = 0;

        for (const row of records) {
            await client.exec(
                `CREATE (:User {
                    id: $id,
                    name: $name,
                    email: $email,
                    age: $age,
                    city: $city
                })`,
                {
                    params: {
                        id: parseInt(row.id),
                        name: row.name,
                        email: row.email,
                        age: parseInt(row.age),
                        city: row.city
                    }
                }
            );
            rowCount++;

            // Commit in batches
            if (rowCount % 1000 === 0) {
                await client.commit();
                await client.beginTransaction();
                console.log(`Imported ${rowCount} rows...`);
            }
        }

        await client.commit();
        console.log(`Successfully imported ${rowCount} users`);

    } catch (error) {
        await client.rollback();
        throw error;
    } finally {
        await client.close();
    }
}

importCSV('users.csv');

const std = @import("std");
const geode = @import("geode_client");

pub fn main() !void {
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    var client = geode.GeodeClient.init(allocator, "localhost", 3141, true);
    defer client.deinit();

    try client.connect();
    try client.sendHello("csv-import", "1.0.0");
    _ = try client.receiveMessage(30000);

    // Begin transaction
    try client.sendBegin();
    _ = try client.receiveMessage(30000);

    // Read CSV file
    const file = try std.fs.cwd().openFile("users.csv", .{});
    defer file.close();

    var buf_reader = std.io.bufferedReader(file.reader());
    var reader = buf_reader.reader();

    // Skip header line
    _ = try reader.readUntilDelimiterAlloc(allocator, '\n', 4096);

    var row_count: usize = 0;
    var line_buf: [4096]u8 = undefined;

    while (try reader.readUntilDelimiterOrEof(&line_buf, '\n')) |line| {
        var iter = std.mem.splitScalar(u8, line, ',');

        const id = iter.next() orelse continue;
        const name = iter.next() orelse continue;
        const email = iter.next() orelse continue;
        const age = iter.next() orelse continue;
        const city = iter.next() orelse continue;

        var params = std.json.ObjectMap.init(allocator);
        defer params.deinit();

        try params.put("id", .{ .integer = try std.fmt.parseInt(i64, id, 10) });
        try params.put("name", .{ .string = name });
        try params.put("email", .{ .string = email });
        try params.put("age", .{ .integer = try std.fmt.parseInt(i64, age, 10) });
        try params.put("city", .{ .string = city });

        try client.sendRunGql(row_count + 1,
            \\CREATE (:User {
            \\  id: $id,
            \\  name: $name,
            \\  email: $email,
            \\  age: $age,
            \\  city: $city
            \\})
        , .{ .object = params });
        _ = try client.receiveMessage(30000);

        row_count += 1;
    }

    // Commit transaction
    try client.sendCommit();
    _ = try client.receiveMessage(30000);

    std.debug.print("Successfully imported {} users\n", .{row_count});
}

import asyncio
import json
from geode_client import Client

async def import_json(filename: str):
    client = Client(host="localhost", port=3141, skip_verify=True)

    async with client.connection() as conn:
        await conn.begin()

        try:
            with open(filename, 'r') as file:
                users = json.load(file)

            for user in users:
                await conn.execute(
                    """CREATE (:User {
                        id: $id,
                        name: $name,
                        email: $email,
                        tags: $tags
                    })""",
                    {
                        "id": user['id'],
                        "name": user['name'],
                        "email": user['email'],
                        "tags": user['tags']  # Lists are supported
                    }
                )

            await conn.commit()
            print(f"Successfully imported {len(users)} users")

        except Exception as e:
            await conn.rollback()
            raise e

asyncio.run(import_json("users.json"))

use geode_client::{Client, Value};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fs;

#[derive(Debug, Deserialize, Serialize)]
struct User {
    id: i64,
    name: String,
    email: String,
    tags: Vec<String>,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::new("127.0.0.1", 3141).skip_verify(true);
    let mut conn = client.connect().await?;

    // Read JSON file
    let content = fs::read_to_string("users.json")?;
    let users: Vec<User> = serde_json::from_str(&content)?;

    conn.begin().await?;

    for user in &users {
        let tags: Vec<Value> = user.tags.iter()
            .map(|t| Value::string(t))
            .collect();

        let mut params = HashMap::new();
        params.insert("id".to_string(), Value::int(user.id));
        params.insert("name".to_string(), Value::string(&user.name));
        params.insert("email".to_string(), Value::string(&user.email));
        params.insert("tags".to_string(), Value::list(tags));

        conn.query_with_params(
            r#"CREATE (:User {
                id: $id,
                name: $name,
                email: $email,
                tags: $tags
            })"#,
            &params
        ).await?;
    }

    conn.commit().await?;
    println!("Successfully imported {} users", users.len());

    Ok(())
}

import { createClient } from '@geodedb/client';
import { readFileSync } from 'fs';

interface User {
    id: number;
    name: string;
    email: string;
    tags: string[];
}

async function importJSON(filename: string) {
    const client = await createClient('quic://localhost:3141');

    const content = readFileSync(filename, 'utf-8');
    const users: User[] = JSON.parse(content);

    await client.beginTransaction();

    try {
        for (const user of users) {
            await client.exec(
                `CREATE (:User {
                    id: $id,
                    name: $name,
                    email: $email,
                    tags: $tags
                })`,
                {
                    params: {
                        id: user.id,
                        name: user.name,
                        email: user.email,
                        tags: user.tags
                    }
                }
            );
        }

        await client.commit();
        console.log(`Successfully imported ${users.length} users`);

    } catch (error) {
        await client.rollback();
        throw error;
    } finally {
        await client.close();
    }
}

importJSON('users.json');

package main

import (
    "bufio"
    "context"
    "database/sql"
    "encoding/json"
    "log"
    "os"

    _ "geodedb.com/geode"
)

type Event struct {
    Type      string                 `json:"type"`
    UserID    int                    `json:"user_id"`
    Timestamp string                 `json:"timestamp"`
    Data      map[string]interface{} `json:"data"`
}

func main() {
    db, err := sql.Open("geode", "localhost:3141")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    ctx := context.Background()

    file, err := os.Open("events.jsonl")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        log.Fatal(err)
    }

    scanner := bufio.NewScanner(file)
    rowCount := 0

    for scanner.Scan() {
        var event Event
        if err := json.Unmarshal(scanner.Bytes(), &event); err != nil {
            log.Printf("Skipping invalid JSON: %v", err)
            continue
        }

        dataJSON, _ := json.Marshal(event.Data)

        _, err = tx.ExecContext(ctx,
            `CREATE (:Event {type: ?, user_id: ?, timestamp: ?, data: ?})`,
            event.Type, event.UserID, event.Timestamp, string(dataJSON))
        if err != nil {
            log.Printf("Error importing event: %v", err)
            continue
        }

        rowCount++
        if rowCount%5000 == 0 {
            tx.Commit()
            tx, _ = db.BeginTx(ctx, nil)
            log.Printf("Imported %d events...", rowCount)
        }
    }

    tx.Commit()
    log.Printf("Successfully imported %d events", rowCount)
}

package main

import (
    "context"
    "database/sql"
    "log"

    _ "geodedb.com/geode"
    _ "github.com/go-sql-driver/mysql"
)

func main() {
    // Connect to MySQL
    mysqlDB, err := sql.Open("mysql", "root:password@tcp(localhost:3306)/myapp")
    if err != nil {
        log.Fatal(err)
    }
    defer mysqlDB.Close()

    // Connect to Geode
    geodeDB, err := sql.Open("geode", "localhost:3141")
    if err != nil {
        log.Fatal(err)
    }
    defer geodeDB.Close()

    ctx := context.Background()

    // Begin Geode transaction
    tx, err := geodeDB.BeginTx(ctx, nil)
    if err != nil {
        log.Fatal(err)
    }

    // Migrate users
    rows, err := mysqlDB.QueryContext(ctx, "SELECT id, name, email, created_at FROM users")
    if err != nil {
        log.Fatal(err)
    }
    defer rows.Close()

    userCount := 0
    for rows.Next() {
        var id int
        var name, email, createdAt string
        rows.Scan(&id, &name, &email, &createdAt)

        _, err = tx.ExecContext(ctx,
            `CREATE (:User {id: ?, name: ?, email: ?, created_at: ?})`,
            id, name, email, createdAt)
        if err != nil {
            tx.Rollback()
            log.Fatalf("Error migrating user %d: %v", id, err)
        }
        userCount++
    }

    log.Printf("Migrated %d users", userCount)

    // Migrate relationships
    rows, err = mysqlDB.QueryContext(ctx,
        "SELECT user_id, friend_id, created_at FROM friendships")
    if err != nil {
        tx.Rollback()
        log.Fatal(err)
    }
    defer rows.Close()

    relCount := 0
    for rows.Next() {
        var userID, friendID int
        var since string
        rows.Scan(&userID, &friendID, &since)

        _, err = tx.ExecContext(ctx, `
            MATCH (u1:User {id: ?})
            MATCH (u2:User {id: ?})
            CREATE (u1)-[:KNOWS {since: ?}]->(u2)
        `, userID, friendID, since)
        if err != nil {
            log.Printf("Warning: Failed to create relationship %d->%d: %v", userID, friendID, err)
            continue
        }
        relCount++
    }

    log.Printf("Migrated %d relationships", relCount)

    if err := tx.Commit(); err != nil {
        log.Fatal(err)
    }

    log.Println("Migration complete!")
}

package main

import (
    "context"
    "database/sql"
    "encoding/csv"
    "log"
    "os"
    "strconv"
    "sync"
    "sync/atomic"

    _ "geodedb.com/geode"
)

type Record struct {
    ID    int
    Name  string
    Email string
}

func importBatch(ctx context.Context, db *sql.DB, batch []Record, batchNum int, total *int64) {
    tx, err := db.BeginTx(ctx, nil)
    if err != nil {
        log.Printf("Batch %d: Failed to begin transaction: %v", batchNum, err)
        return
    }

    for _, r := range batch {
        _, err := tx.ExecContext(ctx,
            `CREATE (:User {id: ?, name: ?, email: ?})`,
            r.ID, r.Name, r.Email)
        if err != nil {
            tx.Rollback()
            log.Printf("Batch %d: Failed: %v", batchNum, err)
            return
        }
    }

    if err := tx.Commit(); err != nil {
        log.Printf("Batch %d: Commit failed: %v", batchNum, err)
        return
    }

    atomic.AddInt64(total, int64(len(batch)))
    log.Printf("Batch %d complete (%d records)", batchNum, len(batch))
}

func main() {
    db, err := sql.Open("geode", "localhost:3141")
    if err != nil {
        log.Fatal(err)
    }
    defer db.Close()

    // Increase connection pool for parallel imports
    db.SetMaxOpenConns(10)

    ctx := context.Background()

    // Read CSV
    file, _ := os.Open("large_users.csv")
    defer file.Close()
    reader := csv.NewReader(file)
    reader.Read() // Skip header

    var records []Record
    for {
        row, err := reader.Read()
        if err != nil {
            break
        }
        id, _ := strconv.Atoi(row[0])
        records = append(records, Record{ID: id, Name: row[1], Email: row[2]})
    }

    // Split into batches
    batchSize := 5000
    workers := 4
    var batches [][]Record

    for i := 0; i < len(records); i += batchSize {
        end := i + batchSize
        if end > len(records) {
            end = len(records)
        }
        batches = append(batches, records[i:end])
    }

    log.Printf("Processing %d records in %d batches with %d workers",
        len(records), len(batches), workers)

    // Process with worker pool
    var wg sync.WaitGroup
    sem := make(chan struct{}, workers)
    var total int64

    for i, batch := range batches {
        wg.Add(1)
        sem <- struct{}{}

        go func(b []Record, num int) {
            defer wg.Done()
            defer func() { <-sem }()
            importBatch(ctx, db, b, num, &total)
        }(batch, i)
    }

    wg.Wait()
    log.Printf("Successfully imported %d of %d records", total, len(records))
}