Tables Module

The tables module provides document-oriented data storage with JSONB support, similar to Dataverse or MongoDB. Tables can be scoped to organizations, applications, or globally.

Import

from bifrost import tables

Table Operations

tables.create()

Create a new table for storing documents.

async def create(
    name: str,
    description: str | None = None,
    table_schema: dict[str, Any] | None = None,
    scope: str | None = None,
    app: str | None = None,
) -> TableInfo

Parameters

Parameter	Type	Description
name	str	Table name (lowercase, underscores allowed, must start with letter)
description	str	Optional table description
table_schema	dict	Optional JSON Schema for validation hints
scope	str	None=context org, "global"=global, UUID=specific org
app	str	Application UUID to scope table to an app

Returns

TableInfo with table metadata including id, name, organization_id, application_id.

Example

# Create org-scoped table
customers = await tables.create("customers", description="Customer records")

# Create app-scoped table
app_data = await tables.create("app_settings", app="app-uuid-here")

# Create global table
shared = await tables.create("shared_lookups", scope="global")

tables.list()

List all tables in the current scope.

async def list(
    scope: str | None = None,
    app: str | None = None,
) -> list[TableInfo]

Parameters

Parameter	Type	Description
scope	str	None=context org (includes global), "global"=global only, UUID=specific org
app	str	Filter by application UUID

Example

# List all accessible tables
all_tables = await tables.list()

# List only app-specific tables
app_tables = await tables.list(app="app-uuid-here")

tables.delete()

Delete a table and all its documents.

async def delete(
    name: str,
    scope: str | None = None,
    app: str | None = None,
) -> bool

Returns

True if deleted, False if table not found.

Example

deleted = await tables.delete("old_customers")

Document Operations

tables.insert()

Insert a document into a table.

async def insert(
    table: str,
    data: dict[str, Any],
    scope: str | None = None,
    app: str | None = None,
) -> DocumentData

Parameters

Parameter	Type	Description
table	str	Table name
data	dict	Document data (any JSON-serializable dict)
scope	str	Organization scope
app	str	Application UUID

Returns

DocumentData with id, table_id, data, created_at, updated_at.

Example

doc = await tables.insert("customers", {
    "name": "Acme Corp",
    "email": "info@acme.com",
    "status": "active",
    "revenue": 50000
})
print(f"Created document: {doc.id}")

tables.get()

Get a document by ID.

async def get(
    table: str,
    doc_id: str,
    scope: str | None = None,
    app: str | None = None,
) -> DocumentData | None

Returns

DocumentData if found, None if not found.

Example

doc = await tables.get("customers", "uuid-here")
if doc:
    print(doc.data["name"])

tables.update()

Update a document (partial update, merges with existing data).

async def update(
    table: str,
    doc_id: str,
    data: dict[str, Any],
    scope: str | None = None,
    app: str | None = None,
) -> DocumentData | None

Parameters

Parameter	Type	Description
table	str	Table name
doc_id	str	Document UUID
data	dict	Fields to update (merged with existing data)

Example

# Only updates status field, other fields unchanged
updated = await tables.update("customers", doc_id, {"status": "inactive"})

tables.delete_document()

Delete a document.

async def delete_document(
    table: str,
    doc_id: str,
    scope: str | None = None,
    app: str | None = None,
) -> bool

Returns

True if deleted, False if not found.

tables.query()

Query documents with filtering and pagination.

async def query(
    table: str,
    where: dict[str, Any] | None = None,
    order_by: str | None = None,
    order_dir: str = "asc",
    limit: int = 100,
    offset: int = 0,
    scope: str | None = None,
    app: str | None = None,
) -> DocumentList

Parameters

Parameter	Type	Description
table	str	Table name
where	dict	Filter conditions (see Filter Operators below)
order_by	str	Field name to order by
order_dir	str	"asc" or "desc"
limit	int	Maximum documents to return (default 100, max 1000)
offset	int	Number of documents to skip

Returns

DocumentList with documents, total, limit, offset.

Example

# Simple equality filter
results = await tables.query("customers", where={"status": "active"})

# With pagination
results = await tables.query(
    "customers",
    where={"status": "active"},
    order_by="name",
    limit=20,
    offset=40
)

for doc in results.documents:
    print(doc.data["name"])

print(f"Showing {len(results.documents)} of {results.total}")

tables.count()

Count documents matching a filter.

async def count(
    table: str,
    where: dict[str, Any] | None = None,
    scope: str | None = None,
    app: str | None = None,
) -> int

Example

total = await tables.count("customers")
active = await tables.count("customers", where={"status": "active"})

Filter Operators

The where parameter supports advanced filter operators for flexible querying.

Simple Equality

where={"status": "active"}
# SQL: WHERE data->>'status' = 'active'

Comparison Operators

# Greater than
where={"amount": {"gt": 100}}

# Greater than or equal
where={"amount": {"gte": 100}}

# Less than
where={"amount": {"lt": 1000}}

# Less than or equal
where={"amount": {"lte": 1000}}

# Range query (multiple operators)
where={"amount": {"gte": 100, "lt": 1000}}

Not Equal

where={"status": {"ne": "deleted"}}

LIKE Patterns

# Case-sensitive LIKE
where={"name": {"like": "%acme%"}}

# Case-insensitive LIKE (recommended)
where={"name": {"ilike": "%acme%"}}

IN Lists

where={"category": {"in": ["electronics", "software", "services"]}}

NULL Checks

# Field is null
where={"deleted_at": {"is_null": True}}

# Field is not null
where={"deleted_at": {"is_null": False}}

Combined Filters

Multiple fields are combined with AND:

results = await tables.query("orders", where={
    "status": "active",
    "amount": {"gte": 100, "lt": 10000},
    "category": {"in": ["electronics", "software"]},
    "customer_name": {"ilike": "%corp%"}
})
# SQL: WHERE status = 'active'
#      AND amount >= 100 AND amount < 10000
#      AND category IN ('electronics', 'software')
#      AND customer_name ILIKE '%corp%'

Types

TableInfo

class TableInfo(BaseModel):
    id: str
    name: str
    description: str | None
    table_schema: dict[str, Any] | None
    organization_id: str | None
    application_id: str | None
    created_by: str | None

DocumentData

class DocumentData(BaseModel):
    id: str
    table_id: str
    data: dict[str, Any]
    created_at: str | None
    updated_at: str | None

DocumentList

class DocumentList(BaseModel):
    documents: list[DocumentData]
    total: int
    limit: int
    offset: int

Scoping

Tables support three levels of scoping:

Organization Scope (Default)

Tables are scoped to the current organization:

# Uses execution context org
table = await tables.create("customers")

Application Scope

Tables can be scoped to a specific application within an organization:

# Scoped to an app
table = await tables.create("app_data", app="app-uuid")
results = await tables.query("app_data", app="app-uuid")

Global Scope

Tables with scope="global" are platform-wide:

# Global table (no org_id)
table = await tables.create("shared_lookups", scope="global")

See Also

• Config Module - Key-value configuration
• Knowledge Module - Vector storage for AI