Musingly Platform Integration Guide¶
A comprehensive guide for engineers integrating content generation systems with the Musingly micro-learning platform. This document covers the platform architecture, content model, Python SDK usage, and ETL workflows for batch importing content.
Table of Contents¶
- Introduction
- Platform Architecture
- Content Model
- SDK Setup
- Creating Import Bundles
- Validation
- Import Process
- Snapshot Lifecycle
- API Reference
- Best Practices
- Troubleshooting
Introduction¶
The Musingly platform delivers technical education through micro-learning—atomic 5-10 minute lessons that can be consumed independently or as part of structured learning paths. Unlike traditional LMS systems, Musingly models knowledge as a Directed Acyclic Graph (DAG) where lessons have explicit prerequisite relationships.
Who is this guide for?¶
- ETL Engineers building content pipelines that produce import bundles
- AI/ML Engineers developing content generation systems
- Backend Developers integrating content workflows with the platform
- Data Engineers managing content versioning and deployment
Key Capabilities¶
| Feature | Description |
|---|---|
| Pydantic Models | Type-safe Python SDK with IDE autocompletion |
| JSON Schema | Language-agnostic contract for multi-language support |
| Snapshot Versioning | Atomic updates with instant rollback capability |
| Slug-Based References | Human-readable, portable import bundles |
| DAG Prerequisites | Knowledge graph with prerequisite relationships |
Platform Architecture¶
System Overview¶
┌─────────────────────────────────────────────────────────────────────┐
│ CONTENT GENERATION PIPELINE │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ AI/LLM │───▶│ Python │───▶│ Import │ │
│ │ Generator │ │ Service │ │ Bundle │ │
│ └──────────────┘ └──────────────┘ └──────┬───────┘ │
│ │ │
│ Uses Python SDK │ │
│ (Pydantic Models) │ │
└──────────────────────────────────────────────────│───────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ VALIDATION LAYER │
│ │
│ • Schema validation (Pydantic/JSON Schema) │
│ • Referential integrity (slug resolution) │
│ • Business rules (no orphaned entities) │
│ • DAG cycle detection │
└──────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ MUSINGLY PLATFORM │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ SNAPSHOT SYSTEM │ │
│ ├───────────────────────────────────────────────────────────────┤ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ v1.0.0 │───▶│ v1.1.0 │───▶│ v2.0.0 │ ◀── CURRENT │ │
│ │ │ archived │ │ archived │ │ current │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │ │
│ │ ▼ │ │
│ │ ┌──────────┐ │ │
│ │ │ v2.1.0 │ ◀── DRAFT │ │
│ │ │ draft │ (importing) │ │
│ │ └──────────┘ │ │
│ └───────────────────────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────────────────────┐ │
│ │ REST API │ │
│ │ Deno + Hono → Supabase Edge Functions │ │
│ └───────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────┘
SDK Generation Pipeline¶
The platform maintains type-safe contracts across TypeScript and Python:
src/schemas/import.ts (Zod Schemas)
│
▼
sdk/dist/import-schemas.json (JSON Schema)
│
▼
python-sdk/musingly_core/models.py (Pydantic)
Build commands:
# Full build: Zod → JSON Schema → Python SDK
deno task sdk:all
# Individual steps
deno task sdk:json-schema # Generate JSON Schema from Zod
deno task python-sdk:generate # Generate Python SDK from JSON Schema
Content Model¶
Entity Hierarchy¶
The platform organizes content in a four-level hierarchy:
┌─────────────────────────────────────────────────────────────────┐
│ DOMAIN │
│ (e.g., "Machine Learning", "Software Engineering") │
└─────────────────────────────────────────────────────────────────┘
│
│ contains many
▼
┌─────────────────────────────────────────────────────────────────┐
│ TRAIL │
│ (e.g., "Transformer Architecture Mastery" - ~10-20 hours) │
└─────────────────────────────────────────────────────────────────┘
│
│ composed of ordered
▼
┌─────────────────────────────────────────────────────────────────┐
│ CONCEPT │
│ (e.g., "Attention Mechanisms" - ~30-60 mins) │
└─────────────────────────────────────────────────────────────────┘
│
│ groups related
▼
┌─────────────────────────────────────────────────────────────────┐
│ SPARK │
│ (e.g., "Self-Attention Explained" - 5-10 mins) │
└─────────────────────────────────────────────────────────────────┘
Entity Definitions¶
Domain (Knowledge Field)¶
The broadest category representing a field of study.
Import Schema Fields:
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| name | string (2-100) | Yes | Display name |
| slug | string | Yes | URL-safe identifier (lowercase, hyphens) |
| description | string (max 500) | No | Brief description |
| icon | string (max 50) | No | Icon identifier (emoji or icon name) |
| color | string | No | Hex color code (e.g., #6366F1) |
| is_published | boolean | No | Visibility flag (default: false) |
Trail (Learning Journey)¶
A curated sequence of Concepts forming a complete learning path.
Import Schema Fields:
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| domain_slug | string | Yes | Reference to parent domain |
| title | string (2-200) | Yes | Display title |
| slug | string | Yes | URL-safe identifier |
| description | string (max 1000) | No | Learning path description |
| difficulty_level | enum | Yes | beginner, intermediate, advanced, expert |
| is_published | boolean | No | Visibility flag (default: false) |
Concept (Knowledge Cluster)¶
Groups 3-8 related Sparks into a coherent sub-topic.
Import Schema Fields:
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| trail_slug | string | No | Reference to parent trail (optional for standalone) |
| title | string (2-200) | Yes | Display title |
| slug | string | Yes | URL-safe identifier |
| description | string (max 1000) | No | Topic description |
| order_index | integer | No | Order within trail (0-indexed, default: 0) |
| is_published | boolean | No | Visibility flag (default: false) |
Spark (Atomic Learning Unit)¶
The fundamental unit—a focused micro-lesson consumable in 5-10 minutes.
Import Schema Fields:
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| title | string (2-200) | Yes | Display title |
| slug | string | Yes | URL-safe identifier |
| summary | string (max 500) | No | Brief summary for listings |
| content_md | string | Yes | Full lesson content in Markdown |
| estimated_mins | integer (1-60) | Yes | Estimated completion time |
| difficulty | enum | Yes | beginner, intermediate, advanced, expert |
| is_published | boolean | No | Visibility flag (default: false) |
| ai_metadata | object | No | AI generation provenance |
AI Metadata Fields:
| Field | Type | Description |
|-------|------|-------------|
| generated_by | string | AI model identifier |
| generation_date | datetime | ISO 8601 timestamp |
| source_materials | string[] | Reference documents used |
| extraction_confidence | float (0-1) | Confidence score |
| review_status | enum | pending, approved, rejected |
Beacon (Discovery Tag)¶
Tags for cross-cutting discovery that transcend the hierarchy.
Import Schema Fields:
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| name | string (2-100) | Yes | Display name |
| slug | string | Yes | URL-safe identifier |
| description | string (max 500) | No | Brief description |
| category | enum | No | technology, concept, application, level, format |
Relationship Entities¶
ConceptSparkImport (Concept-Spark Link)¶
Associates sparks with concepts and defines order.
| Field | Type | Required | Description |
|---|---|---|---|
concept_slug |
string | Yes | Parent concept slug |
spark_slug |
string | Yes | Spark to associate |
order_index |
integer | No | Order within concept (0-indexed) |
SparkBeaconImport (Spark-Beacon Tag)¶
Tags sparks with beacons for discovery.
| Field | Type | Required | Description |
|---|---|---|---|
spark_slug |
string | Yes | Spark to tag |
beacon_slug |
string | Yes | Beacon to apply |
SparkLinkImport (Prerequisite Edge)¶
Defines prerequisite relationships between sparks (DAG edges).
| Field | Type | Required | Description |
|---|---|---|---|
source_slug |
string | Yes | Prerequisite spark (learn first) |
target_slug |
string | Yes | Dependent spark (learn after) |
link_type |
enum | Yes | prerequisite, recommended, related, deepens |
weight |
float (0-1) | No | Dependency strength |
description |
string (max 500) | No | Why this link exists |
Link Types:
- prerequisite: Required knowledge (weight: 0.8-1.0)
- recommended: Helpful but not required (weight: 0.5-0.7)
- related: Tangentially connected (weight: 0.3-0.5)
- deepens: Advanced extension (weight: 0.6-0.8)
ConceptLinkImport (Concept Prerequisite)¶
Defines prerequisite relationships between concepts.
| Field | Type | Required | Description |
|---|---|---|---|
source_slug |
string | Yes | Prerequisite concept |
target_slug |
string | Yes | Dependent concept |
link_type |
enum | Yes | prerequisite, recommended, related, deepens |
weight |
float (0-1) | No | Dependency strength |
Knowledge Graph¶
The prerequisite links form a DAG enabling:
- Prerequisite visualization: Show learners what they need first
- Learning path calculation: Find optimal routes to any topic
- Gap analysis: Identify missing knowledge
- Recommendations: Suggest next steps based on current knowledge
┌─────────────┐
│ Linear │
│ Algebra │
└──────┬──────┘
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Matrix │ │ Vector │ │ Tensor │
│Operations│ │ Spaces │ │ Basics │
└────┬─────┘ └────┬─────┘ └────┬─────┘
│ │ │
└────────────┼────────────┘
│
▼
┌──────────────┐
│ Word │
│ Embeddings │
└──────┬───────┘
│
┌────────────┴────────────┐
│ │
▼ ▼
┌─────────────┐ ┌─────────────┐
│ Self- │ │ Cosine │
│ Attention │ │ Similarity │
└──────┬──────┘ └─────────────┘
│
▼
┌─────────────┐
│ Multi-Head │
│ Attention │
└─────────────┘
SDK Setup¶
Building the SDKs¶
Generate both TypeScript and Python SDKs:
# Full build pipeline
deno task sdk:all
# This runs:
# 1. deno task sdk:build:full - TypeScript SDK from OpenAPI
# 2. deno task sdk:json-schema - JSON Schema from Zod schemas
# 3. deno task python-sdk:generate - Python SDK from JSON Schema
SDK Outputs¶
| Output | Location | Description |
|---|---|---|
| TypeScript SDK | sdk/dist/ |
npm-compatible package |
| JSON Schema | sdk/dist/import-schemas.json |
Language-agnostic contract |
| Python SDK | python-sdk/ |
Pydantic models package |
Installing the Python SDK¶
Option 1: Local Development¶
# Install in editable mode
pip install -e /path/to/core/python-sdk
# Or in requirements.txt
-e /path/to/core/python-sdk
Option 2: pyproject.toml Reference¶
Option 3: Published Package¶
Verifying Installation¶
from musingly_core.models import ImportBundle, DomainImport, SparkImport
# Check installation
import musingly_core
print(f"SDK Version: {musingly_core.__version__}")
# Verify models are available
bundle = ImportBundle(
schema_version="1.0.0",
version_label="test",
metadata={"generated_at": "2026-01-19T00:00:00Z", "source_system": "test"},
)
print(f"Bundle created: {bundle.version_label}")
Creating Import Bundles¶
Bundle Structure¶
An ImportBundle contains all entities for a complete content update:
from musingly_core.models import (
ImportBundle,
ImportMetadata,
DomainImport,
TrailImport,
ConceptImport,
SparkImport,
BeaconImport,
ConceptSparkImport,
SparkBeaconImport,
SparkLinkImport,
ConceptLinkImport,
)
from datetime import datetime, timezone
bundle = ImportBundle(
schema_version="1.0.0",
version_label="v2.1.0",
name="January 2026 Content Update",
metadata=ImportMetadata(
generated_at=datetime.now(timezone.utc).isoformat(),
source_system="content-generator",
source_version="1.0.0",
description="AI-generated ML curriculum expansion",
),
domains=[...],
trails=[...],
concepts=[...],
sparks=[...],
beacons=[...],
concept_sparks=[...],
spark_beacons=[...],
spark_links=[...],
concept_links=[...],
)
Entity Examples¶
Domains¶
domains = [
DomainImport(
name="Machine Learning",
slug="machine-learning",
description="Fundamentals and advanced topics in machine learning",
icon="brain",
color="#6366F1",
is_published=True,
),
DomainImport(
name="Software Engineering",
slug="software-engineering",
description="Best practices for building robust software systems",
icon="code",
color="#10B981",
is_published=True,
),
]
Trails¶
trails = [
TrailImport(
domain_slug="machine-learning", # References domain by slug
title="Transformer Architecture Mastery",
slug="transformer-mastery",
description="Deep dive into attention mechanisms and transformer models",
difficulty_level="intermediate",
is_published=False, # Draft until reviewed
),
]
Concepts¶
concepts = [
ConceptImport(
trail_slug="transformer-mastery",
title="Attention Mechanisms",
slug="attention-mechanisms",
description="Understanding self-attention and its variants",
order_index=0,
is_published=False,
),
ConceptImport(
trail_slug="transformer-mastery",
title="Positional Encoding",
slug="positional-encoding",
description="How transformers understand sequence order",
order_index=1,
is_published=False,
),
]
Sparks¶
sparks = [
SparkImport(
title="Self-Attention Explained",
slug="self-attention-explained",
summary="Learn how tokens communicate through attention weights",
content_md="""
# Self-Attention Explained
Self-attention allows each token in a sequence to attend to all other tokens...
## Key Concepts
1. **Query, Key, Value**: The three projections...
2. **Attention Weights**: Computed via softmax...
3. **Scaled Dot-Product**: Why we divide by √d_k...
## Example
```python
import torch
attention = torch.softmax(Q @ K.T / math.sqrt(d_k), dim=-1) @ V
#### Beacons
```python
beacons = [
BeaconImport(
name="Transformers",
slug="transformers",
description="Content related to transformer architectures",
category="technology",
),
BeaconImport(
name="NLP",
slug="nlp",
description="Natural Language Processing topics",
category="application",
),
]
Relationships¶
# Link sparks to concepts with ordering
concept_sparks = [
ConceptSparkImport(
concept_slug="attention-mechanisms",
spark_slug="self-attention-explained",
order_index=0,
),
ConceptSparkImport(
concept_slug="attention-mechanisms",
spark_slug="multi-head-attention",
order_index=1,
),
]
# Tag sparks with beacons
spark_beacons = [
SparkBeaconImport(
spark_slug="self-attention-explained",
beacon_slug="transformers",
),
SparkBeaconImport(
spark_slug="self-attention-explained",
beacon_slug="nlp",
),
]
# Prerequisite graph (DAG edges)
spark_links = [
SparkLinkImport(
source_slug="linear-algebra-basics", # Learn first
target_slug="self-attention-explained", # Then this
link_type="prerequisite",
weight=0.9,
description="Linear algebra concepts needed for attention",
),
SparkLinkImport(
source_slug="self-attention-explained",
target_slug="multi-head-attention",
link_type="prerequisite",
weight=0.95,
),
]
# Concept-level prerequisites
concept_links = [
ConceptLinkImport(
source_slug="attention-mechanisms",
target_slug="transformer-encoder",
link_type="prerequisite",
weight=0.85,
),
]
Complete Generator Example¶
from musingly_core.models import ImportBundle, ImportMetadata
from datetime import datetime, timezone
import json
def create_content_bundle() -> ImportBundle:
"""Create a complete import bundle for new ML content."""
return ImportBundle(
schema_version="1.0.0",
version_label="v2.1.0",
name="ML Curriculum Expansion Q1 2026",
metadata=ImportMetadata(
generated_at=datetime.now(timezone.utc).isoformat(),
source_system="ml-content-generator",
source_version="2.0.0",
description="Expanded coverage of transformer architectures",
),
domains=[
# ... domain definitions
],
trails=[
# ... trail definitions
],
concepts=[
# ... concept definitions
],
sparks=[
# ... spark definitions
],
beacons=[
# ... beacon definitions
],
concept_sparks=[
# ... concept-spark links
],
spark_beacons=[
# ... spark-beacon tags
],
spark_links=[
# ... spark prerequisite graph
],
concept_links=[
# ... concept prerequisite graph
],
)
def export_bundle(bundle: ImportBundle, output_path: str) -> None:
"""Export bundle to JSON file for import."""
with open(output_path, "w") as f:
json.dump(bundle.model_dump(), f, indent=2)
print(f"Bundle exported to {output_path}")
if __name__ == "__main__":
bundle = create_content_bundle()
export_bundle(bundle, "import-bundle-v2.1.0.json")
Validation¶
Pre-Flight Validation¶
Pydantic provides automatic schema validation. Add custom rules for referential integrity:
from pydantic import ValidationError
from musingly_core.models import ImportBundle
import json
def validate_bundle(bundle_path: str) -> tuple[bool, list[str]]:
"""Validate a bundle file before import."""
errors = []
try:
with open(bundle_path) as f:
data = json.load(f)
# Pydantic validates schema, types, and constraints
bundle = ImportBundle.model_validate(data)
# Custom validation rules
errors.extend(validate_referential_integrity(bundle))
errors.extend(validate_dag_acyclic(bundle))
errors.extend(validate_slug_uniqueness(bundle))
except ValidationError as e:
for error in e.errors():
loc = ".".join(str(x) for x in error["loc"])
errors.append(f"Schema error at {loc}: {error['msg']}")
except json.JSONDecodeError as e:
errors.append(f"Invalid JSON: {e}")
return len(errors) == 0, errors
def validate_referential_integrity(bundle: ImportBundle) -> list[str]:
"""Check all slug references point to existing entities."""
errors = []
# Collect all defined slugs
domain_slugs = {d.slug for d in bundle.domains}
trail_slugs = {t.slug for t in bundle.trails}
concept_slugs = {c.slug for c in bundle.concepts}
spark_slugs = {s.slug for s in bundle.sparks}
beacon_slugs = {b.slug for b in bundle.beacons}
# Validate trail -> domain references
for trail in bundle.trails:
if trail.domain_slug not in domain_slugs:
errors.append(
f"Trail '{trail.slug}' references unknown domain '{trail.domain_slug}'"
)
# Validate concept -> trail references
for concept in bundle.concepts:
if concept.trail_slug and concept.trail_slug not in trail_slugs:
errors.append(
f"Concept '{concept.slug}' references unknown trail '{concept.trail_slug}'"
)
# Validate concept_sparks
for cs in bundle.concept_sparks:
if cs.concept_slug not in concept_slugs:
errors.append(f"concept_sparks references unknown concept '{cs.concept_slug}'")
if cs.spark_slug not in spark_slugs:
errors.append(f"concept_sparks references unknown spark '{cs.spark_slug}'")
# Validate spark_beacons
for sb in bundle.spark_beacons:
if sb.spark_slug not in spark_slugs:
errors.append(f"spark_beacons references unknown spark '{sb.spark_slug}'")
if sb.beacon_slug not in beacon_slugs:
errors.append(f"spark_beacons references unknown beacon '{sb.beacon_slug}'")
# Validate spark_links
for link in bundle.spark_links:
if link.source_slug not in spark_slugs:
errors.append(f"spark_links references unknown source '{link.source_slug}'")
if link.target_slug not in spark_slugs:
errors.append(f"spark_links references unknown target '{link.target_slug}'")
return errors
def validate_dag_acyclic(bundle: ImportBundle) -> list[str]:
"""Ensure spark and concept links form a DAG (no cycles)."""
errors = []
# Build adjacency list for spark graph
spark_graph: dict[str, set[str]] = {}
for link in bundle.spark_links:
if link.source_slug not in spark_graph:
spark_graph[link.source_slug] = set()
spark_graph[link.source_slug].add(link.target_slug)
# Detect cycles using DFS
def has_cycle(graph: dict[str, set[str]]) -> list[str] | None:
visited = set()
path = set()
path_list = []
def dfs(node: str) -> list[str] | None:
visited.add(node)
path.add(node)
path_list.append(node)
for neighbor in graph.get(node, []):
if neighbor in path:
cycle_start = path_list.index(neighbor)
return path_list[cycle_start:] + [neighbor]
if neighbor not in visited:
result = dfs(neighbor)
if result:
return result
path.remove(node)
path_list.pop()
return None
for node in graph:
if node not in visited:
cycle = dfs(node)
if cycle:
return cycle
return None
cycle = has_cycle(spark_graph)
if cycle:
errors.append(f"Cycle detected in spark_links: {' -> '.join(cycle)}")
return errors
def validate_slug_uniqueness(bundle: ImportBundle) -> list[str]:
"""Ensure slugs are unique within each entity type."""
errors = []
def check_duplicates(items: list, entity_type: str) -> list[str]:
slugs = [item.slug for item in items]
seen = set()
duplicates = set()
for slug in slugs:
if slug in seen:
duplicates.add(slug)
seen.add(slug)
return [f"Duplicate {entity_type} slug: '{s}'" for s in duplicates]
errors.extend(check_duplicates(bundle.domains, "domain"))
errors.extend(check_duplicates(bundle.trails, "trail"))
errors.extend(check_duplicates(bundle.concepts, "concept"))
errors.extend(check_duplicates(bundle.sparks, "spark"))
errors.extend(check_duplicates(bundle.beacons, "beacon"))
return errors
# Usage
if __name__ == "__main__":
valid, errors = validate_bundle("import-bundle-v2.1.0.json")
if valid:
print("✓ Bundle is valid")
else:
print("✗ Validation failed:")
for error in errors:
print(f" - {error}")
Import Process¶
CLI Commands¶
# Validate without making database changes
deno task content:import content/bundle.json --dry-run
# Import as draft snapshot (requires manual promotion)
deno task content:import content/bundle.json
# Import and automatically promote to current
deno task content:import content/bundle.json --promote
Import Workflow¶
┌─────────────────────────────────────────────────────────────────────┐
│ 1. PRE-FLIGHT │
│ • Validate bundle JSON against Pydantic models │
│ • Check referential integrity │
│ • Detect DAG cycles │
│ • Verify slug uniqueness │
└─────────────────────────────────────────────────────────────────────┘
│
▼ (validation passed)
┌─────────────────────────────────────────────────────────────────────┐
│ 2. CREATE DRAFT SNAPSHOT │
│ • Call: create_import_snapshot(name, version_label) │
│ • Returns: snapshot_id (UUID) │
│ • Status: 'draft' │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 3. START IMPORT LOG │
│ • Call: start_import(version_label, source_system) │
│ • Returns: import_id (UUID) │
│ • Status: 'pending' │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 4. BULK INSERT (Order Matters!) │
│ 1. Domains (no dependencies) │
│ 2. Trails (depends on domains) │
│ 3. Concepts (depends on trails) │
│ 4. Sparks (no direct dependencies) │
│ 5. Beacons (no dependencies) │
│ 6. Concept_sparks (depends on concepts + sparks) │
│ 7. Spark_beacons (depends on sparks + beacons) │
│ 8. Spark_links (depends on sparks) │
│ 9. Concept_links (depends on concepts) │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 5. FINALIZE IMPORT │
│ • Call: finalize_import(import_id, snapshot_id, entity_counts) │
│ • Status: 'ready' │
│ • Snapshot ready for promotion │
└─────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────┐
│ 6. VERIFY (Optional but Recommended) │
│ • Run queries against draft snapshot │
│ • Verify entity counts match expected │
│ • Check sample content renders correctly │
└─────────────────────────────────────────────────────────────────────┘
│
▼ (verification passed)
┌─────────────────────────────────────────────────────────────────────┐
│ 7. PROMOTE │
│ • Call: promote_import(import_id, snapshot_id) │
│ • Snapshot becomes 'current' │
│ • Previous snapshot becomes 'archived' │
│ • Import status: 'promoted' │
└─────────────────────────────────────────────────────────────────────┘
Typical ETL Workflow¶
# 1. Build/update the SDK (when schemas change)
deno task sdk:all
# 2. Generate content in Python
cd /path/to/content-generator
pip install -e /path/to/core/python-sdk
python generate_content.py --output bundle.json
# 3. Validate the bundle
deno task content:import bundle.json --dry-run
# 4. Import to staging (as draft)
deno task content:import bundle.json
# 5. Verify in database, then promote
deno task content:promote --snapshot-id=<uuid>
Environment Setup¶
# Required environment variables
export SUPABASE_URL="https://your-project.supabase.co"
export SUPABASE_SECRET_KEY="your-service-role-key"
# Or use a .env file in the project root
Snapshot Lifecycle¶
Snapshot States¶
┌───────────────────────────────────────────────────────────────────┐
│ SNAPSHOT LIFECYCLE │
├───────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────┐ │
│ │ DRAFT │ ─── Import in progress, not visible to users │
│ └────┬────┘ │
│ │ │
│ │ finalize_import() │
│ ▼ │
│ ┌─────────┐ │
│ │ READY │ ─── Import complete, awaiting promotion │
│ └────┬────┘ │
│ │ │
│ │ promote_import() │
│ ▼ │
│ ┌──────────┐ │
│ │ CURRENT │ ─── Active snapshot, served to all users │
│ └────┬─────┘ │
│ │ │
│ │ (new snapshot promoted) │
│ ▼ │
│ ┌──────────┐ │
│ │ ARCHIVED │ ─── Historical, can be used for rollback │
│ └──────────┘ │
│ │
└───────────────────────────────────────────────────────────────────┘
Rollback Procedure¶
If issues are discovered after promotion:
-- Option 1: Rollback to parent snapshot
SELECT rollback_import('<import_id>');
-- Option 2: Promote a specific archived snapshot
SELECT promote_snapshot('<archived_snapshot_id>');
Cleanup Old Snapshots¶
-- Keep only the 5 most recent published snapshots
SELECT cleanup_old_snapshots(5);
-- Check import statistics
SELECT * FROM get_import_stats(30); -- Last 30 days
API Reference¶
Database Functions¶
| Function | Description |
|---|---|
create_import_snapshot(name, version_label, description, parent_id) |
Creates a new draft snapshot |
start_import(version_label, source_system, source_version, created_by) |
Starts an import log entry |
update_import_status(import_id, status, ...) |
Updates import status and details |
finalize_import(import_id, snapshot_id, entity_counts) |
Marks import as ready |
promote_import(import_id, snapshot_id) |
Promotes snapshot to current |
rollback_import(import_id) |
Rolls back to previous snapshot |
delete_draft_snapshot(snapshot_id) |
Deletes a draft and its data |
cleanup_old_snapshots(keep_count) |
Archives old snapshots |
get_import_stats(days) |
Returns import statistics |
Python SDK Models¶
| Model | Description |
|---|---|
ImportBundle |
Complete import payload |
ImportMetadata |
Source and timing metadata |
DomainImport |
Top-level category |
TrailImport |
Learning path |
ConceptImport |
Topic cluster |
SparkImport |
Micro-lesson content |
BeaconImport |
Discovery tag |
ConceptSparkImport |
Concept-Spark relationship |
SparkBeaconImport |
Spark-Beacon tag |
SparkLinkImport |
Spark prerequisite edge |
ConceptLinkImport |
Concept prerequisite edge |
BundleValidationError |
Validation error details |
BundleValidationResult |
Validation result with stats |
Deno Task Commands¶
| Command | Description |
|---|---|
deno task sdk:all |
Full SDK build (TS + JSON Schema + Python) |
deno task sdk:json-schema |
Generate JSON Schema from Zod |
deno task python-sdk:generate |
Generate Python SDK from JSON Schema |
deno task python-sdk:test |
Run Python SDK tests |
deno task content:import <file> |
Import bundle as draft |
deno task content:import <file> --promote |
Import and promote |
deno task content:import <file> --dry-run |
Validate only |
deno task content:verify |
Verify snapshot integrity |
deno task content:promote |
Promote a ready snapshot |
Best Practices¶
1. Version Labeling¶
Use consistent versioning:
# Semantic versioning
version_label = "v2.1.0"
# Date-based
version_label = "2026-01-Q1"
# Content-based
version_label = "ml-expansion-jan2026"
2. Incremental Updates¶
For partial updates, only include changed entities:
bundle = ImportBundle(
schema_version="1.0.0",
version_label="v2.1.1-hotfix",
metadata=ImportMetadata(
generated_at=datetime.now(timezone.utc).isoformat(),
source_system="content-editor",
description="Fixed typos in attention-mechanisms sparks",
),
# Empty arrays for unchanged entity types
domains=[],
trails=[],
concepts=[],
# Only include modified sparks
sparks=[
SparkImport(
slug="self-attention-explained",
# ... updated content
),
],
beacons=[],
concept_sparks=[],
spark_beacons=[],
spark_links=[],
concept_links=[],
)
3. CI/CD Integration¶
# .github/workflows/content-deploy.yml
name: Deploy Content
on:
push:
branches: [main]
paths:
- 'content/**'
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: '3.10'
- name: Install SDK
run: pip install ./python-sdk
- name: Validate Bundle
run: python scripts/validate_bundle.py content/bundle.json
- name: Import to Staging
env:
DATABASE_URL: ${{ secrets.STAGING_DATABASE_URL }}
run: python scripts/import_bundle.py content/bundle.json --no-promote
- name: Run Tests
run: pytest tests/content/
- name: Promote if Tests Pass
env:
DATABASE_URL: ${{ secrets.STAGING_DATABASE_URL }}
run: python scripts/promote_snapshot.py
4. Error Handling¶
from musingly_core.models import ImportBundle
import logging
logger = logging.getLogger(__name__)
def safe_import(bundle_path: str, importer) -> dict:
"""Import with comprehensive error handling."""
# Load and validate
try:
with open(bundle_path) as f:
bundle = ImportBundle.model_validate(json.load(f))
except ValidationError as e:
logger.error(f"Schema validation failed: {e}")
raise
# Pre-flight checks
valid, errors = validate_bundle(bundle_path)
if not valid:
logger.error(f"Pre-flight validation failed: {errors}")
raise ValueError(f"Validation errors: {errors}")
# Import
snapshot_id = None
try:
result = importer.import_bundle(bundle, auto_promote=False)
snapshot_id = result["snapshot_id"]
logger.info(f"Import successful: {result}")
return result
except Exception as e:
logger.error(f"Import failed: {e}")
# Cleanup draft snapshot if created
if snapshot_id:
try:
importer.delete_draft(snapshot_id)
logger.info(f"Cleaned up draft snapshot: {snapshot_id}")
except Exception as cleanup_error:
logger.warning(f"Failed to cleanup draft: {cleanup_error}")
raise
5. Slug Conventions¶
Follow these patterns for URL-safe slugs:
import re
def create_slug(text: str) -> str:
"""Convert text to a valid slug."""
slug = text.lower()
slug = re.sub(r'[^\w\s-]', '', slug) # Remove special chars
slug = re.sub(r'[\s_]+', '-', slug) # Replace spaces/underscores
slug = re.sub(r'-+', '-', slug) # Collapse multiple hyphens
slug = slug.strip('-') # Remove leading/trailing
return slug
# Examples
create_slug("Self-Attention Explained") # "self-attention-explained"
create_slug("What is AI?") # "what-is-ai"
create_slug("Python 3.10 Features") # "python-310-features"
Troubleshooting¶
Common Issues¶
1. "Duplicate slug" Error¶
Cause: Same slug used twice in the bundle.
Fix: Ensure each entity has a unique slug within its type.
slugs = [s.slug for s in bundle.sparks]
duplicates = [s for s in slugs if slugs.count(s) > 1]
if duplicates:
raise ValueError(f"Duplicate spark slugs: {set(duplicates)}")
2. Foreign Key Violation¶
Cause: Trail references a domain_slug that doesn't exist.
Fix: Ensure all referenced entities are included in the bundle.
3. DAG Cycle Detected¶
Cause: Prerequisite chain forms a loop.
Fix: Review spark_links and remove the edge creating the cycle.
4. Schema Version Mismatch¶
Cause: Bundle uses a schema version the SDK doesn't support.
Fix: Rebuild the Python SDK to get the latest schema:
Viewing Import Logs¶
-- Recent imports
SELECT
version_label,
status,
started_at,
completed_at,
entity_counts,
error_message
FROM import_logs
ORDER BY started_at DESC
LIMIT 10;
-- Failed imports with details
SELECT
version_label,
error_message,
error_details,
started_at
FROM import_logs
WHERE status = 'failed'
ORDER BY started_at DESC;
-- Import statistics
SELECT * FROM get_import_stats(30);
Next Steps¶
- Set up your Python environment with the SDK installed
- Create a sample bundle using the examples above
- Run validation to ensure correctness
- Test import on a staging database
- Integrate into your content generation pipeline
For API consumption (not content production), see the full API documentation at /doc on the running server.