Multi-tenant Authentication with API Keys - CLI Usage

This notebook demonstrates how to use TimeDB’s authentication system using the CLI commands with shell syntax.

What we’ll cover:

1. Setting up the database schema with users table

2. Creating tenants and users via CLI

3. Using API keys to authenticate requests

4. Demonstrating tenant isolation (each user only sees their own data)

5. Testing authentication enforcement

import os
import uuid
import requests
import pandas as pd
import time
from datetime import datetime, timezone, timedelta

# Use TEST_DATABASE_URL for testing (falls back to DATABASE_URL if not set)
DATABASE_URL = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL")
if not DATABASE_URL:
    raise ValueError("Please set TEST_DATABASE_URL or DATABASE_URL environment variable")

# IMPORTANT: Override both TIMEDB_DSN and DATABASE_URL so the API server uses the same database
os.environ["TIMEDB_DSN"] = DATABASE_URL
os.environ["DATABASE_URL"] = DATABASE_URL

# API base URL
API_BASE_URL = "http://127.0.0.1:8000"

print(f"Database URL configured: {DATABASE_URL[:50]}...")
print(f"Environment variables set for API server")
print(f"API URL: {API_BASE_URL}")

Part 1: Setup Database Schema with Users Table

First, we’ll create the database schema including the users table for authentication using CLI commands.

# Delete existing schema (for a clean start)
!timedb delete tables --dsn "{DATABASE_URL}" -y

# Create schema with users table
!timedb create tables --dsn "{DATABASE_URL}" --with-users -y

Part 2: Create Tenants and Users

Now we’ll create two separate tenants, each with their own user. Each user gets a unique API key that is tied to their tenant_id.

# Generate two tenant IDs
tenant_a_id = str(uuid.uuid4())
tenant_b_id = str(uuid.uuid4())

print(f"Tenant A ID: {tenant_a_id}")
print(f"Tenant B ID: {tenant_b_id}")

# Create User A for Tenant A
!timedb users create --dsn "{DATABASE_URL}" --tenant-id {tenant_a_id} --email alice@tenant-a.com

# Extract API key from the output above
# You'll need to copy the API key manually from the output above
api_key_a = input("Enter API Key for User A from output above: ")
print(f"Stored API Key A: {api_key_a}")

# Create User B for Tenant B
!timedb users create --dsn "{DATABASE_URL}" --tenant-id {tenant_b_id} --email bob@tenant-b.com

# Extract API key from the output above
api_key_b = input("Enter API Key for User B from output above: ")
print(f"Stored API Key B: {api_key_b}")

# List all users to verify
!timedb users list --dsn "{DATABASE_URL}"

Part 3: Start the API Server

Now we start the API server. With the users_table present, authentication will be required for all endpoints.

# Kill any existing API server first
!pkill -f "uvicorn.*timedb" || true
time.sleep(2)
print("Killed any existing API server")

# Start API server in background using CLI
!timedb api --host 127.0.0.1 --port 8000 &

# Give the server time to start
time.sleep(3)

# Verify API is running
try:
    response = requests.get(f"{API_BASE_URL}/")
    print(f"API is running: {response.json()['name']}")
except Exception as e:
    print(f"API check failed: {e}")

Part 4: Test Authentication Enforcement

Let’s verify that authentication is required when the users_table exists.

# Test request WITHOUT API key - should return 401
response = requests.get(f"{API_BASE_URL}/values")
print(f"Request without API key:")
print(f"  Status: {response.status_code}")
print(f"  Response: {response.json()}")

# Test request with INVALID API key - should return 401
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": "invalid-key-12345"}
)
print(f"Request with invalid API key:")
print(f"  Status: {response.status_code}")
print(f"  Response: {response.json()}")

# Test request with VALID API key - should return 200
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": api_key_a}
)
print(f"Request with valid API key (User A):")
print(f"  Status: {response.status_code}")
print(f"  Response: {response.json()}")

Part 5: Demonstrate Tenant Isolation

Now let’s upload data from both users and verify that each user can only see their own tenant’s data.

# User A uploads temperature data
base_time = datetime(2025, 1, 1, 0, 0, tzinfo=timezone.utc)

value_rows_a = [
    {"valid_time": (base_time + timedelta(hours=i)).isoformat(),
     "value_key": "temperature_a",
     "value": 20.0 + i}
    for i in range(6)
]

response = requests.post(
    f"{API_BASE_URL}/upload",
    json={
        "run_start_time": datetime.now(timezone.utc).isoformat(),
        "value_rows": value_rows_a
    },
    headers={"X-API-Key": api_key_a, "Content-Type": "application/json"}
)
print(f"User A upload status: {response.status_code}")
print(f"Response: {response.json()}")

# User B uploads humidity data
value_rows_b = [
    {"valid_time": (base_time + timedelta(hours=i)).isoformat(),
     "value_key": "humidity_b",
     "value": 50.0 + i}
    for i in range(6)
]

response = requests.post(
    f"{API_BASE_URL}/upload",
    json={
        "run_start_time": datetime.now(timezone.utc).isoformat(),
        "value_rows": value_rows_b
    },
    headers={"X-API-Key": api_key_b, "Content-Type": "application/json"}
)
print(f"User B upload status: {response.status_code}")
print(f"Response: {response.json()}")

# User A reads data - should ONLY see their own data (temperature_a)
response = requests.get(
    f"{API_BASE_URL}/values",
    params={
        "start_valid": base_time.isoformat(),
        "end_valid": (base_time + timedelta(hours=6)).isoformat(),
        "mode": "flat"
    },
    headers={"X-API-Key": api_key_a}
)
data_a = response.json()
print(f"User A sees {data_a['count']} records:")
if data_a['count'] > 0:
    df_a = pd.DataFrame(data_a['data'])
    print(df_a[['valid_time', 'series_key', 'value']])
    print(f"\nSeries keys visible to User A: {df_a['series_key'].unique().tolist()}")

# User B reads data - should ONLY see their own data (humidity_b)
response = requests.get(
    f"{API_BASE_URL}/values",
    params={
        "start_valid": base_time.isoformat(),
        "end_valid": (base_time + timedelta(hours=6)).isoformat(),
        "mode": "flat"
    },
    headers={"X-API-Key": api_key_b}
)
data_b = response.json()
print(f"User B sees {data_b['count']} records:")
if data_b['count'] > 0:
    df_b = pd.DataFrame(data_b['data'])
    print(df_b[['valid_time', 'series_key', 'value']])
    print(f"\nSeries keys visible to User B: {df_b['series_key'].unique().tolist()}")

# Verify tenant isolation
print("Tenant Isolation Verification:")
print(f"  User A (Tenant A) sees: {data_a['count']} records")
print(f"  User B (Tenant B) sees: {data_b['count']} records")
print("")

# Check that they see different data
if data_a['count'] > 0 and data_b['count'] > 0:
    series_a = set(pd.DataFrame(data_a['data'])['series_key'].unique())
    series_b = set(pd.DataFrame(data_b['data'])['series_key'].unique())
    overlap = series_a & series_b

    if len(overlap) == 0:
        print("SUCCESS: No overlapping series between tenants!")
        print(f"  Tenant A series: {series_a}")
        print(f"  Tenant B series: {series_b}")
    else:
        print(f"WARNING: Found overlapping series: {overlap}")

Part 6: User Deactivation

Let’s demonstrate how deactivating a user revokes their API access using CLI commands.

# Deactivate User B
!timedb users deactivate --dsn "{DATABASE_URL}" --email bob@tenant-b.com -y

# Try to access API with deactivated user's key - should return 401
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": api_key_b}
)
print(f"Request with deactivated user's API key:")
print(f"  Status: {response.status_code}")
print(f"  Response: {response.json()}")

# Reactivate User B
!timedb users activate --dsn "{DATABASE_URL}" --email bob@tenant-b.com

# Verify access is restored
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": api_key_b}
)
print(f"Request after reactivation:")
print(f"  Status: {response.status_code}")

Summary

This notebook demonstrated TimeDB’s multi-tenant authentication system using CLI commands.

Key CLI Commands Used:

1. Schema Management:

   • timedb create tables --dsn <url> --with-users -y - Create schema with users table

   • timedb delete tables --dsn <url> -y - Delete schema

2. User Management:

   • timedb users create --dsn <url> --tenant-id <uuid> --email <email> - Create user with API key

   • timedb users list --dsn <url> - List all users

   • timedb users deactivate --dsn <url> --email <email> -y - Revoke API access

   • timedb users activate --dsn <url> --email <email> - Restore API access

   • timedb users regenerate-key --dsn <url> --email <email> - Generate new API key

3. API Server:

   • timedb api --host <host> --port <port> - Start API server

Python Variable Injection in Shell Commands:

Notice how we used {variable} syntax to inject Python variables into shell commands. Important: When the variable contains special characters (like URLs with ? or &), you must wrap it in quotes:

DATABASE_URL = "postgresql://user:pass@host/db?sslmode=require"

# CORRECT - with quotes around the variable
!timedb create tables --dsn "{DATABASE_URL}"

# WRONG - without quotes, shell will interpret ? and & as special characters
!timedb create tables --dsn {DATABASE_URL}

Authentication Concepts:

• When users_table exists, authentication is required for all endpoints

• Use X-API-Key header with all requests

• Invalid or missing API keys return 401 Unauthorized

• Each user’s API key is tied to a tenant_id for data isolation