n6k-duckdb 0.3.0

In-process bridge between DuckDB connections with permission-based access control

n6k-duckdb

In-process bridge between DuckDB connections with permission-based access control.

Install

pip install n6k-duckdb

Usage

import duckdb
from n6k_duckdb.bridge import bridge

cfg = {"allow_unsigned_extensions": "true"}
source = duckdb.connect(config=cfg)
source.sql("CREATE TABLE users(id INTEGER, name VARCHAR)")
source.sql("INSERT INTO users VALUES (1, 'alice'), (2, 'bob')")

target = duckdb.connect(config=cfg)
bridge(source, target, "app", permissions={"users": "readwrite"})

# Query through the bridge
target.sql("SELECT * FROM app.users").show()

# Insert through the bridge (requires 'readwrite')
target.sql("INSERT INTO app.users VALUES (3, 'charlie')")

# Update and delete also work with 'readwrite'
target.sql("UPDATE app.users SET name = 'Alice' WHERE id = 1")
target.sql("DELETE FROM app.users WHERE id = 2")

Permissions

Permission	Allows
'read'	SELECT only
'readwrite'	SELECT, INSERT, UPDATE, DELETE

Tables not listed in permissions are inaccessible.

How it works

bridge() automatically installs and loads the n6k_bridge DuckDB extension from the n6k extension repository. The extension creates an in-process bridge between two DuckDB database instances using a token-based handshake.

No network server is required — data flows directly between connections in the same process.

Server framework (optional extra)

This package also ships a reusable FastAPI framework for serving the n6k protocol — install with pip install "n6k-duckdb[test-server]".

Minimal server with a DuckDB backend:

import duckdb
from fastapi import FastAPI, WebSocket
from n6k_duckdb.duckdb_handler import DuckDBHandler
from n6k_duckdb.server_fastapi.register import register
from n6k_duckdb.server_asgi.ws import WsReject

app = FastAPI()


def session_factory(ws: WebSocket, **_):
    # The client sends ?catalog=<name> matching its ATTACH name.
    catalog = ws.query_params.get("catalog")
    if not catalog:
        raise WsReject(code=4400, reason="missing ?catalog= query param")
    con = duckdb.connect()
    con.execute(f"ATTACH ':memory:' AS \"{catalog}\"")
    # ... populate tables in `con` ...
    return DuckDBHandler(con, catalog_name=catalog)


register(app, prefix="", session=session_factory)

# Run: `uvicorn your_module:app --port 8099`

A worked example — auth, per-WS observability, read-only sessionless — lives in src/n6k_duckdb/test_server/app.py. Run it with python -m n6k_duckdb.test_server --port 8099.

The framework is backend-agnostic: N6KSessionHandler / N6KSessionlessHandler can wrap any data source. For a non-duckdb example see src/n6k_duckdb/server/__tests__/test_memory_handler.py (an in-memory dict[str, pa.Table] backend).