18. Examples

Example 1 — Ingest JSON from an API, query, export

A common pattern: receive a JSON payload from a server, normalise it into tables, run a relational query, and export results.

import json
from relpy import RelPy, AutoNumber, col, count, sum_, avg

# --- Schema ---
db = RelPy()

db.create_table("sessions")
db.add_column("sessions", "id",          AutoNumber, is_primary_key=True)
db.add_column("sessions", "game_system", str, nullable=False)
db.add_column("sessions", "city",        str, nullable=False)

db.create_table("registrations")
db.add_column("registrations", "id",         AutoNumber, is_primary_key=True)
db.add_column("registrations", "session_id", int, nullable=False, references="sessions.id")
db.add_column("registrations", "status",     str, nullable=False)
db.add_column("registrations", "price",      float, nullable=False)
db.add_column("registrations", "checked_in", bool, nullable=False, default=False)

# --- Simulate JSON payload ---
payload = json.loads('''[
    {"session": {"game_system": "D&D 5e", "city": "Tel Aviv"},
     "registrations": [
         {"status": "confirmed", "price": 55.0, "checked_in": true},
         {"status": "confirmed", "price": 55.0, "checked_in": false}
     ]},
    {"session": {"game_system": "Pathfinder", "city": "Haifa"},
     "registrations": [
         {"status": "confirmed", "price": 50.0, "checked_in": true},
         {"status": "waitlist",  "price": 0.0,  "checked_in": false}
     ]}
]''')

# --- Normalise and insert ---
for item in payload:
    sess = db.insert("sessions", item["session"])
    for reg in item["registrations"]:
        db.insert("registrations", {**reg, "session_id": sess["id"]})

# --- Query: confirmed + checked-in revenue by game system ---
result = (
    db.query("registrations")
      .join("sessions")
      .where(
          (col("registrations.status")     == "confirmed") &
          (col("registrations.checked_in") == True)
      )
      .group_by("sessions.game_system")
      .aggregate(
          players=count(),
          revenue=sum_("registrations.price"),
          avg_price=avg("registrations.price"),
      )
      .order_by("revenue", descending=True)
      .to_list()
)

for row in result:
    print(row)

Example 2 — Save, load, and continue inserting

from relpy import RelPy, AutoNumber, col

db = RelPy()
db.create_table("tasks")
db.add_column("tasks", "id",     AutoNumber, is_primary_key=True)
db.add_column("tasks", "title",  str, nullable=False)
db.add_column("tasks", "done",   bool, nullable=False, default=False)

db.insert("tasks", {"title": "Write docs"})
db.insert("tasks", {"title": "Write tests"})

db.save("tasks.relpy.json")

# ---  Later in a different script ---
db2 = RelPy.load("tasks.relpy.json")

# AutoNumber continues from where it left off
db2.insert("tasks", {"title": "Ship it", "done": False})

print(db2.query("tasks").pluck("id"))   # [1, 2, 3]
print(db2.query("tasks").count())       # 3

[1, 2, 3]
3

Example 3 — Encrypted email column

from relpy import RelPy, AutoNumber, col

key = RelPy.generate_encryption_key()

db = RelPy(encryption_key=key)
db.create_table("users")
db.add_column("users", "id",    AutoNumber, is_primary_key=True)
db.add_column("users", "name",  str, nullable=False)
db.add_column("users", "email", str, nullable=False, is_encrypted=True)

db.insert("users", {"name": "Alice", "email": "alice@example.com"})

# Default export hides the value
print(db.to_list("users"))
# [{"id": 1, "name": "Alice", "email": "[ENCRYPTED]"}]

# Decrypted export requires the key
print(db.to_list("users", decrypt=True))
# [{"id": 1, "name": "Alice", "email": "alice@example.com"}]

# Query on an encrypted column works via blind index
result = db.query("users").where(col("email") == "alice@example.com").first()
print(result["name"])  # Alice

Example 4 — Mixing RelPyDB results with plain Python

from relpy import RelPy, AutoNumber, col

db = RelPy()
# ... (insert orders as above)

# Pull rows into Python and continue with plain logic
paid_rows = (
    db.query("orders")
      .where(col("status") == "paid")
      .to_list()
)

# Standard Python comprehension
totals_by_user = {}
for row in paid_rows:
    uid = row["user_id"]
    totals_by_user[uid] = totals_by_user.get(uid, 0) + row["amount"]

top_user_id = max(totals_by_user, key=totals_by_user.get)
print(f"Top buyer: user {top_user_id}, total: {totals_by_user[top_user_id]}")

RelPyDB rows are plain Python dicts — pass them to any standard library function, third-party tool, or your own logic without conversion.

Example 5 — Export to pandas and NumPy

import pandas as pd
import numpy as np
from relpy import RelPy, AutoNumber

db = RelPy()
# ... setup and inserts

# to pandas
df = db.to_pandas("orders")
print(df.describe())

# to numpy
arr = db.to_numpy("orders")
print(arr.shape)

# export only one column
amounts = db.to_numpy("orders", column_name="amount")
print(amounts.mean())

Example 6 — Unique index prevents duplicates

from relpy import RelPy, AutoNumber, ConstraintError

db = RelPy()
db.create_table("users")
db.add_column("users", "id",    AutoNumber, is_primary_key=True)
db.add_column("users", "email", str, nullable=False)

db.create_index("users", "email", unique=True)

db.insert("users", {"email": "alice@example.com"})  # OK

try:
    db.insert("users", {"email": "alice@example.com"})  # duplicate
except ConstraintError as e:
    print(f"Blocked: {e}")

Blocked: Unique index violation on users.email: alice@example.com