Database

Overview

The Ziet Database provides persistent storage for your applications. Unlike Memory (which is scoped to a single agent run), Database stores data permanently across all runs.

from ziet import db

# Insert data
user_id = db.insert("users", {
    "email": "alice@example.com",
    "name": "Alice",
    "created_at": "now()"
})

# Query data
user = db.query("users").where("email", "alice@example.com").first()

# Update data
db.update("users", {"name": "Alice Smith"}).where("id", user_id).execute()

# Delete data
db.delete("users").where("id", user_id).execute()

Database stores data forever - perfect for user accounts, orders, products, and application state.

Database vs Memory

Feature	Database	Memory
Scope	Global (all runs, agents, endpoints)	Per agent run
Persistence	Permanent	Temporary (cleared after run)
Use case	User data, products, orders	Intermediate results, agent context
Access	Actions, Agents, Endpoints	Actions, Agents only
Query	SQL-like queries	Key-value lookup

When to use Database:

User accounts and profiles
Products, orders, transactions
Application state
Data that needs to persist forever

When to use Memory:

Intermediate action results
Agent workflow context
Temporary data during execution

Basic Operations

Insert

Add new records:

from ziet import db

# Insert single record
user_id = db.insert("users", {
    "email": "alice@example.com",
    "name": "Alice",
    "role": "admin",
    "created_at": "now()"
})

print(user_id)  # "usr_abc123xyz"

Returns: The ID of the inserted record

Query

Fetch records:

# Get all records
users = db.query("users").all()

# Get first record
user = db.query("users").first()

# Get specific record
user = db.query("users").where("id", user_id).first()

# Get with filters
admins = db.query("users").where("role", "admin").all()

Update

Modify existing records:

# Update single field
db.update("users", {"name": "Alice Smith"}) \
  .where("id", user_id) \
  .execute()

# Update multiple fields
db.update("users", {
    "name": "Alice Smith",
    "role": "super_admin",
    "updated_at": "now()"
}).where("id", user_id).execute()

Delete

Remove records:

# Delete single record
db.delete("users").where("id", user_id).execute()

# Delete multiple records
db.delete("users").where("role", "guest").execute()

Querying Data

Where Clauses

Filter records:

# Equals
users = db.query("users").where("role", "admin").all()

# Not equals
users = db.query("users").where("role", "!=", "guest").all()

# Greater than
orders = db.query("orders").where("total", ">", 100).all()

# Less than
orders = db.query("orders").where("total", "<", 50).all()

# LIKE (pattern matching)
users = db.query("users").where("email", "LIKE", "%@gmail.com").all()

# IN (multiple values)
users = db.query("users").where("role", "IN", ["admin", "moderator"]).all()

Multiple Conditions

Chain where clauses:

# AND conditions
orders = db.query("orders") \
    .where("status", "pending") \
    .where("total", ">", 100) \
    .all()

# OR conditions
from ziet.db import or_

orders = db.query("orders").where(
    or_(
        ("status", "pending"),
        ("status", "processing")
    )
).all()

Sorting

Order results:

# Ascending (default)
users = db.query("users").order_by("created_at").all()

# Descending
users = db.query("users").order_by("created_at", "DESC").all()

# Multiple columns
users = db.query("users") \
    .order_by("role", "ASC") \
    .order_by("name", "ASC") \
    .all()

Limiting and Pagination

# Limit results
users = db.query("users").limit(10).all()

# Offset (skip first N)
users = db.query("users").offset(10).limit(10).all()

# Pagination helper
page = 2
per_page = 10
users = db.query("users") \
    .offset((page - 1) * per_page) \
    .limit(per_page) \
    .all()

Counting

# Count all
total = db.query("users").count()

# Count with filter
admins_count = db.query("users").where("role", "admin").count()

Using in Actions

from ziet import Action, db

@Action(
    id="create_user",
    name="Create User",
    description="Create user account"
)
def create_user(email: str, name: str) -> None:
    """Create user in database."""
    user_id = db.insert("users", {
        "email": email,
        "name": name,
        "role": "user",
        "created_at": "now()"
    })
    
    print(f"Created user: {user_id}")

@Action(
    id="get_user_orders",
    name="Get User Orders",
    description="Fetch all orders for a user"
)
def get_user_orders(user_id: str) -> None:
    """Get user's orders from database."""
    orders = db.query("orders") \
        .where("user_id", user_id) \
        .order_by("created_at", "DESC") \
        .all()
    
    print(f"Found {len(orders)} orders")

Using in Agents

from ziet import Agent, Action, db, memory

@Action(
    id="save_research",
    name="Save Research",
    description="Save research to database"
)
def save_research(topic: str) -> None:
    """Save research results to database."""
    # Get results from memory
    results = memory.get("research_results")
    
    # Save to database
    research_id = db.insert("research", {
        "topic": topic,
        "results": results,
        "created_at": "now()"
    })
    
    memory.add(key="research_id", value=research_id)

@Agent(
    id="research_agent",
    name="ResearchAgent",
    instructions="""
    Research the topic and save results to database.
    This allows us to access research across different runs.
    """,
    actions=["search_topic", "save_research"]
)
class ResearchAgent:
    pass

Using in Endpoints

from ziet import Endpoint, db

@Endpoint(
    id="get_users",
    method="GET",
    path="/users"
)
def get_users(role: str = None, limit: int = 10) -> dict:
    """List users from database."""
    query = db.query("users")
    
    if role:
        query = query.where("role", role)
    
    users = query.limit(limit).all()
    
    return {
        "status": 200,
        "body": {
            "users": users,
            "count": len(users)
        }
    }

@Endpoint(
    id="create_order",
    method="POST",
    path="/orders"
)
def create_order(user_id: str, items: list, total: float) -> dict:
    """Create order in database."""
    order_id = db.insert("orders", {
        "user_id": user_id,
        "total": total,
        "status": "pending",
        "created_at": "now()"
    })
    
    # Insert order items
    for item in items:
        db.insert("order_items", {
            "order_id": order_id,
            "product_id": item["product_id"],
            "quantity": item["quantity"],
            "price": item["price"]
        })
    
    return {
        "status": 201,
        "body": {"order_id": order_id}
    }

Complete Example: E-commerce

from ziet import Endpoint, Action, Agent, db, memory

# ===== Endpoints (REST API) =====

@Endpoint(id="list_products", method="GET", path="/products")
def list_products(category: str = None) -> dict:
    """List products."""
    query = db.query("products")
    if category:
        query = query.where("category", category)
    
    products = query.all()
    return {"status": 200, "body": {"products": products}}

@Endpoint(id="create_order", method="POST", path="/orders")
def create_order(user_id: str, product_ids: list) -> dict:
    """Create order."""
    # Get products
    products = db.query("products") \
        .where("id", "IN", product_ids) \
        .all()
    
    total = sum(p["price"] for p in products)
    
    # Create order
    order_id = db.insert("orders", {
        "user_id": user_id,
        "total": total,
        "status": "pending",
        "created_at": "now()"
    })
    
    # Add items
    for product in products:
        db.insert("order_items", {
            "order_id": order_id,
            "product_id": product["id"],
            "price": product["price"]
        })
    
    return {
        "status": 201,
        "body": {"order_id": order_id, "total": total}
    }

# ===== Actions =====

@Action(
    id="charge_payment",
    name="Charge Payment",
    description="Process payment for order"
)
def charge_payment(order_id: str) -> None:
    """Charge payment using Stripe."""
    from ziet.integrations import stripe
    
    # Get order from database
    order = db.query("orders").where("id", order_id).first()
    
    # Charge payment
    payment = stripe.create_payment_intent(
        amount=int(order["total"] * 100),
        currency="usd"
    )
    
    # Update order status
    db.update("orders", {
        "status": "paid",
        "payment_id": payment["id"]
    }).where("id", order_id).execute()
    
    memory.add(key="payment_id", value=payment["id"])

@Action(
    id="send_confirmation",
    name="Send Confirmation",
    description="Send order confirmation email"
)
def send_confirmation(order_id: str) -> None:
    """Send confirmation email."""
    from ziet.integrations import sendgrid
    
    # Get order with items
    order = db.query("orders").where("id", order_id).first()
    items = db.query("order_items").where("order_id", order_id).all()
    
    # Get user
    user = db.query("users").where("id", order["user_id"]).first()
    
    # Send email
    sendgrid.send(
        to=user["email"],
        subject="Order Confirmed",
        body=f"<h1>Order {order_id}</h1><p>Total: ${order['total']}</p>"
    )

# ===== Agent =====

@Agent(
    id="order_fulfillment_agent",
    name="OrderFulfillmentAgent",
    instructions="""
    Process order fulfillment:
    1. Charge payment using Stripe
    2. Send confirmation email to customer
    
    All order data is stored in the database and persists across runs.
    """,
    actions=["charge_payment", "send_confirmation"]
)
class OrderFulfillmentAgent:
    pass

Usage:

# Create order via API
curl -X POST https://api.ziet.ai/orders \
  -d '{"user_id": "usr_123", "product_ids": ["prod_1", "prod_2"]}'

# Returns: {"order_id": "ord_abc123", "total": 99.99}

# Agent automatically fulfills order
ziet run order_fulfillment_agent --data '{"order_id": "ord_abc123"}'

Schema Management

Tables are Auto-Created

No need to define schemas - tables are created automatically on first insert:

# First insert creates the "users" table
db.insert("users", {
    "email": "alice@example.com",
    "name": "Alice"
})

Built-in Fields

Every table automatically gets:

id - Unique identifier (auto-generated)
created_at - Timestamp of creation
updated_at - Timestamp of last update

Field Types

Ziet automatically infers field types:

db.insert("products", {
    "name": "Laptop",           # string
    "price": 999.99,            # float
    "in_stock": True,           # boolean
    "quantity": 10,             # integer
    "tags": ["electronics"],    # list
    "metadata": {"color": "silver"}  # dict
})

Relationships

One-to-Many

# Create user
user_id = db.insert("users", {"email": "alice@example.com"})

# Create orders for user
order1 = db.insert("orders", {"user_id": user_id, "total": 100})
order2 = db.insert("orders", {"user_id": user_id, "total": 200})

# Query user's orders
orders = db.query("orders").where("user_id", user_id).all()

Many-to-Many

Use a join table:

# Create product
product_id = db.insert("products", {"name": "Laptop"})

# Create order
order_id = db.insert("orders", {"total": 999})

# Link product to order (join table)
db.insert("order_products", {
    "order_id": order_id,
    "product_id": product_id,
    "quantity": 1
})

# Query products in order
product_links = db.query("order_products") \
    .where("order_id", order_id) \
    .all()

product_ids = [link["product_id"] for link in product_links]
products = db.query("products").where("id", "IN", product_ids).all()

Best Practices

Use Database for permanent data

# ✅ Good - User data in database
user_id = db.insert("users", {"email": email})

# ❌ Bad - User data in memory (gets cleared)
memory.add(key="user", value={"email": email})

Use Memory for temporary data

# ✅ Good - Intermediate results in memory
memory.add(key="search_results", value=results)

# ❌ Bad - Temporary data in database
db.insert("temp_results", {"data": results})

Index frequently queried fields

Query by common fields:

# Frequently query by user_id
orders = db.query("orders").where("user_id", user_id).all()

# Frequently query by email
user = db.query("users").where("email", email).first()

Coming soon: Database transactions

# Future feature
with db.transaction():
    order_id = db.insert("orders", {...})
    db.insert("order_items", {...})
    db.insert("order_items", {...})
# All succeed or all fail

Store relationships explicitly

# Create user
user_id = db.insert("users", {"email": "alice@example.com"})

# Create order WITH user_id relationship
order_id = db.insert("orders", {
    "user_id": user_id,  # Explicit relationship
    "total": 100
})

Limitations

Current limitations:

No JOIN queries (fetch related data separately)
No transactions yet (coming soon)
No migrations (tables auto-created)
Max 10,000 records per query

For complex queries: Consider using dedicated databases (PostgreSQL, MongoDB) via integrations.

Next Steps

Endpoints

Build REST APIs with endpoints

Memory

Use memory for temporary data

Actions

Access database from actions

Agents

Use database in agent workflows

Getting Started

Agentic

Application

Integrations

Deployment

Overview

Database vs Memory

Basic Operations

Insert

Query

Update

Delete

Querying Data

Where Clauses

Multiple Conditions

Sorting

Counting

Using in Actions

Using in Agents

Using in Endpoints

Complete Example: E-commerce

Schema Management

Tables are Auto-Created

Built-in Fields

Field Types

Relationships

One-to-Many

Many-to-Many

Best Practices

Limitations

Next Steps

Endpoints

Memory

Actions

Agents

​Overview

​Database vs Memory

​Basic Operations

​Insert

​Query

​Update

​Delete

​Querying Data

​Where Clauses

​Multiple Conditions

​Sorting

​Limiting and Pagination

​Counting

​Using in Actions

​Using in Agents

​Using in Endpoints

​Complete Example: E-commerce

​Schema Management

​Tables are Auto-Created

​Built-in Fields

​Field Types

​Relationships

​One-to-Many

​Many-to-Many

​Best Practices

​Limitations

​Next Steps

Endpoints

Memory

Actions

Agents

Overview

Database vs Memory

Basic Operations

Insert

Query

Update

Delete

Querying Data

Where Clauses

Multiple Conditions

Sorting

Limiting and Pagination

Counting

Using in Actions

Using in Agents

Using in Endpoints

Complete Example: E-commerce

Schema Management

Tables are Auto-Created

Built-in Fields

Field Types

Relationships

One-to-Many

Many-to-Many

Best Practices

Limitations

Next Steps