Multi-tenant Authentication with API Keys - SDK Usage

This notebook demonstrates how to use TimeDB’s authentication system using the Python SDK directly.

What we’ll cover:

1. Setting up the database schema with users table using SDK

2. Creating tenants and users using Python functions

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
import psycopg
from datetime import datetime, timezone, timedelta
import timedb as td
from timedb.db import users as db_users

# 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 the SDK.

# Delete existing schema (for a clean start)
td.delete(dsn=DATABASE_URL)
print("Deleted existing tables")

# Create schema with users table using SDK
td.create(dsn=DATABASE_URL, create_users_table=True)
print("Created schema with users table")

Part 2: Create Tenants and Users

Now we’ll create two separate tenants using the timedb.db.users module. Each user gets a unique API key that is tied to their tenant_id.

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

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

# Create User A for Tenant A using SDK
with psycopg.connect(DATABASE_URL) as conn:
    user_a = db_users.create_user(
        conn,
        tenant_id=tenant_a_id,
        email="alice@tenant-a.com"
    )
    conn.commit()

api_key_a = user_a['api_key']
print(f"Created User A:")
print(f"  User ID: {user_a['user_id']}")
print(f"  Tenant ID: {user_a['tenant_id']}")
print(f"  Email: {user_a['email']}")
print(f"  API Key: {api_key_a}")
print(f"  Is Active: {user_a['is_active']}")

# Create User B for Tenant B using SDK
with psycopg.connect(DATABASE_URL) as conn:
    user_b = db_users.create_user(
        conn,
        tenant_id=tenant_b_id,
        email="bob@tenant-b.com"
    )
    conn.commit()

api_key_b = user_b['api_key']
print(f"Created User B:")
print(f"  User ID: {user_b['user_id']}")
print(f"  Tenant ID: {user_b['tenant_id']}")
print(f"  Email: {user_b['email']}")
print(f"  API Key: {api_key_b}")
print(f"  Is Active: {user_b['is_active']}")

# List all users to verify
with psycopg.connect(DATABASE_URL) as conn:
    all_users = db_users.list_users(conn, include_inactive=True)

print(f"\nTotal users: {len(all_users)}\n")
for user in all_users:
    status = "Active" if user['is_active'] else "Inactive"
    print(f"Email: {user['email']:30} | Tenant ID: {user['tenant_id']} | Status: {status}")

Part 3: Start the API Server

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

# Kill any existing API server first
import subprocess
result = subprocess.run(["pkill", "-f", "uvicorn.*timedb"], capture_output=True)
if result.returncode == 0:
    print("Killed existing API server")
    time.sleep(2)  # Wait for port to be released
else:
    print("No existing API server found")

# Start API server in background using SDK
td.start_api_background(host="127.0.0.1", port=8000)
print("API server started in background")

# 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']}")
    print(f"Version: {response.json().get('version', 'unknown')}")
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 the SDK.

# Deactivate User B using SDK
with psycopg.connect(DATABASE_URL) as conn:
    success = db_users.deactivate_user(conn, email="bob@tenant-b.com")
    conn.commit()

if success:
    print("Successfully deactivated user: bob@tenant-b.com")
else:
    print("User not found")

# 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 using SDK
with psycopg.connect(DATABASE_URL) as conn:
    success = db_users.activate_user(conn, email="bob@tenant-b.com")
    conn.commit()

if success:
    print("Successfully reactivated user: 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}")
if response.status_code == 200:
    print(f"  SUCCESS: Access restored for User B")

Part 7: Additional SDK Functions

Let’s explore a few more user management functions available in the SDK.

# Get user by email
with psycopg.connect(DATABASE_URL) as conn:
    user = db_users.get_user_by_email(conn, email="alice@tenant-a.com")

if user:
    print("User found:")
    print(f"  Email: {user['email']}")
    print(f"  Tenant ID: {user['tenant_id']}")
    print(f"  User ID: {user['user_id']}")
    print(f"  Is Active: {user['is_active']}")

# Regenerate API key for User B
with psycopg.connect(DATABASE_URL) as conn:
    new_api_key = db_users.regenerate_api_key(conn, email="bob@tenant-b.com")
    conn.commit()

if new_api_key:
    print(f"New API key generated for bob@tenant-b.com:")
    print(f"  {new_api_key}")
    print(f"\nOld API key: {api_key_b[:20]}...")
    print(f"New API key: {new_api_key[:20]}...")

    # Update the variable
    api_key_b_old = api_key_b
    api_key_b = new_api_key

# Verify old key no longer works
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": api_key_b_old}
)
print(f"Request with old API key:")
print(f"  Status: {response.status_code}")

# Verify new key works
response = requests.get(
    f"{API_BASE_URL}/values",
    headers={"X-API-Key": api_key_b}
)
print(f"\nRequest with new API key:")
print(f"  Status: {response.status_code}")

# List users for a specific tenant
with psycopg.connect(DATABASE_URL) as conn:
    tenant_a_users = db_users.list_users(conn, tenant_id=tenant_a_id)

print(f"Users in Tenant A ({tenant_a_id}):")
for user in tenant_a_users:
    print(f"  {user['email']} - Active: {user['is_active']}")

Summary

This notebook demonstrated TimeDB’s multi-tenant authentication system using the Python SDK.

Key SDK Functions Used:

1. Schema Management (timedb module):

   • td.create(dsn=url, create_users_table=True) - Create schema with users table

   • td.delete(dsn=url) - Delete schema

   • td.start_api_background(host=..., port=...) - Start API server

2. User Management (timedb.db.users module):

   • db_users.create_user(conn, tenant_id=..., email=...) - Create user with API key

   • db_users.list_users(conn, tenant_id=None, include_inactive=False) - List users

   • db_users.get_user_by_email(conn, email=...) - Get user by email

   • db_users.deactivate_user(conn, email=...) - Revoke API access

   • db_users.activate_user(conn, email=...) - Restore API access

   • db_users.regenerate_api_key(conn, email=...) - Generate new API key

   • db_users.get_user_by_api_key(conn, api_key=...) - Validate API key

Key Differences from CLI:

• No need for subprocess calls or parsing command output

• Direct Python function calls with return values

• Full access to user dictionaries with all fields

• More programmatic control over database operations

• Requires managing database connections with psycopg.connect()

Authentication Concepts:

• API keys are generated securely using secrets.token_urlsafe(32)

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

• When users_table exists, authentication is required for all endpoints

• Use X-API-Key header with all API requests

• Invalid or missing API keys return 401 Unauthorized