Surrealdb Python

SurrealDB Python SDK

Running SurrealDB (Server Mode)

Persist data with RocksDB:

surreal start -u root -p root rocksdb:database

In-memory:

surreal start -u root -p root

Client/Server Mode

Connect via WebSocket using the Surreal context manager:

from surrealdb import Surreal

with Surreal("ws://localhost:8000/rpc") as db:
    db.signin({"username": "root", "password": "root"})
    db.use("namespace_test", "database_test")

    db.create(
        "person",
        {
            "user": "me",
            "password": "safe",
            "marketing": True,
            "tags": ["python", "documentation"],
        },
    )

    print(db.select("person"))

    print(db.update("person", {
        "user": "you",
        "password": "very_safe",
        "marketing": False,
        "tags": ["Awesome"],
    }))

    print(db.delete("person"))

    db.query("""
    insert into person {
        user: 'me',
        password: 'very_safe',
        tags: ['python', 'documentation']
    };
    """)

    print(db.query("select * from person"))

    print(db.query("""
    update person content {
        user: 'you',
        password: 'more_safe',
        tags: ['awesome']
    };
    """))

    print(db.query("delete person"))

Embedded Mode

Run SurrealDB directly inside your Python process — no server required.

• In-memory: Surreal("mem://") / AsyncSurreal("mem://")
• File-based persistence: Surreal(f"file://{db_path}") / AsyncSurreal(f"file://{db_path}")

See references/embedded.md for a complete async example with file-based persistence.

File-Based Persistent Embedded Example

import asyncio
import tempfile
from pathlib import Path

from surrealdb import AsyncSurreal


async def main() -> None:
    with tempfile.TemporaryDirectory() as tmpdir:
        db_path = Path(tmpdir) / "mydb"
        db_url = f"file://{db_path}"

        # First connection: create data
        async with AsyncSurreal(db_url) as db:
            await db.use("test", "test")

            await db.create(
                "company",
                {"name": "Acme Corp", "founded": 2020, "employees": 100},
            )

            await db.create(
                "company",
                {"name": "TechStart Inc", "founded": 2021, "employees": 50},
            )

            companies = await db.select("company")
            print(f"Created companies: {companies}")

        # Second connection: verify persistence
        async with AsyncSurreal(db_url) as db:
            await db.use("test", "test")

            companies = await db.select("company")
            print(f"Loaded companies from disk: {companies}")

            updated = await db.query("""
                UPDATE company SET employees = employees + 10
                WHERE name = "Acme Corp"
            """)
            print(f"Updated company: {updated}")

            all_companies = await db.select("company")
            print(f"All companies after update: {all_companies}")


if __name__ == "__main__":
    asyncio.run(main())