Directus CMS ?????????????????????
Directus ???????????? open source headless CMS ???????????????????????? REST ????????? GraphQL API ????????? database ?????????????????? ?????????????????? SQL databases ??????????????????????????? (PostgreSQL, MySQL, SQLite, MS SQL, MariaDB) ???????????????????????????????????? API ????????? ???????????????????????? schema ?????? database ???????????? Directus ??????????????? API ????????????????????????????????????
Progressive Delivery ??????????????????????????????????????? deploy features ???????????????????????? ????????????????????????????????????????????????????????????????????? ???????????????????????????????????? ????????? feature flags, canary releases, A/B testing ?????????????????????????????????????????????????????? deploy ????????????????????? Progressive Delivery ???????????????????????? CMS ???????????????????????????????????????????????? content ????????????????????? users ???????????????????????????????????? ??????????????????????????????????????? rollout ????????????????????????
Directus ???????????????????????? Progressive Delivery ??????????????? API-first architecture ???????????????????????????????????? feature flag services ?????????????????????, Roles & Permissions ?????????????????? content access ?????????????????????, Webhooks ?????????????????????????????????????????? content ?????????????????????, Flows (automation) ??????????????? workflows ???????????????????????????, Versioning track content changes
????????????????????? Directus ?????????????????? Production
Setup Directus CMS
# === Directus Production Setup ===
# 1. Docker Compose Deployment
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
directus:
image: directus/directus:10.12
ports:
- "8055:8055"
volumes:
- directus_uploads:/directus/uploads
- directus_extensions:/directus/extensions
depends_on:
- postgres
- redis
environment:
KEY: ""
SECRET: ""
DB_CLIENT: "pg"
DB_HOST: "postgres"
DB_PORT: "5432"
DB_DATABASE: "directus"
DB_USER: "directus"
DB_PASSWORD: ""
CACHE_ENABLED: "true"
CACHE_STORE: "redis"
CACHE_REDIS: "redis://redis:6379"
CACHE_AUTO_PURGE: "true"
CACHE_TTL: "5m"
ADMIN_EMAIL: "admin@example.com"
ADMIN_PASSWORD: ""
RATE_LIMITER_ENABLED: "true"
RATE_LIMITER_STORE: "redis"
CORS_ENABLED: "true"
CORS_ORIGIN: "https://example.com"
postgres:
image: postgres:16-alpine
volumes:
- postgres_data:/var/lib/postgresql/data
environment:
POSTGRES_DB: directus
POSTGRES_USER: directus
POSTGRES_PASSWORD: ""
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
volumes:
directus_uploads:
directus_extensions:
postgres_data:
redis_data:
EOF
# 2. Environment File
cat > .env << 'EOF'
DIRECTUS_KEY=random-key-here-change-this
DIRECTUS_SECRET=random-secret-here-change-this
DB_PASSWORD=strong-password-here
ADMIN_PASSWORD=admin-password-here
EOF
# 3. Start
docker compose up -d
# 4. Nginx Reverse Proxy
cat > /etc/nginx/sites-available/directus << 'EOF'
server {
listen 443 ssl http2;
server_name cms.example.com;
ssl_certificate /etc/letsencrypt/live/cms.example.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/cms.example.com/privkey.pem;
client_max_body_size 100m;
location / {
proxy_pass http://127.0.0.1:8055;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
EOF
ln -s /etc/nginx/sites-available/directus /etc/nginx/sites-enabled/
nginx -t && systemctl reload nginx
echo "Directus production setup complete"
Progressive Delivery ????????? CMS
Implement progressive delivery ?????????????????? content
#!/usr/bin/env python3
# progressive_delivery.py ??? Progressive Delivery for CMS
import json
import logging
import hashlib
from typing import Dict, List, Optional
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("delivery")
class ProgressiveDeliveryEngine:
def __init__(self):
self.feature_flags = {}
self.experiments = {}
def create_feature_flag(self, flag_id, name, default=False, rollout_pct=0):
"""Create a feature flag for content delivery"""
self.feature_flags[flag_id] = {
"id": flag_id,
"name": name,
"enabled": default,
"rollout_percentage": rollout_pct,
"targeting_rules": [],
"created_at": datetime.utcnow().isoformat(),
}
return self.feature_flags[flag_id]
def add_targeting_rule(self, flag_id, rule):
"""Add targeting rule to a feature flag"""
if flag_id in self.feature_flags:
self.feature_flags[flag_id]["targeting_rules"].append(rule)
def evaluate_flag(self, flag_id, user_context):
"""Evaluate if a user should see the feature"""
flag = self.feature_flags.get(flag_id)
if not flag or not flag["enabled"]:
return False
# Check targeting rules
for rule in flag["targeting_rules"]:
if rule["type"] == "user_segment":
if user_context.get("segment") in rule["segments"]:
return True
elif rule["type"] == "country":
if user_context.get("country") in rule["countries"]:
return True
# Percentage-based rollout
if flag["rollout_percentage"] > 0:
user_hash = int(hashlib.md5(
f"{flag_id}:{user_context.get('user_id', '')}".encode()
).hexdigest(), 16) % 100
return user_hash < flag["rollout_percentage"]
return False
def get_content_variant(self, content_id, user_context):
"""Get the right content variant for a user"""
flag_id = f"content_{content_id}"
if self.evaluate_flag(flag_id, user_context):
return {"variant": "B", "content_id": f"{content_id}_v2"}
return {"variant": "A", "content_id": content_id}
def create_canary_release(self, content_id, stages):
"""Create a canary release plan"""
return {
"content_id": content_id,
"stages": stages,
"current_stage": 0,
"status": "active",
"example_stages": [
{"name": "Internal", "percentage": 1, "duration": "1 hour"},
{"name": "Canary", "percentage": 5, "duration": "4 hours"},
{"name": "Early Access", "percentage": 25, "duration": "24 hours"},
{"name": "General", "percentage": 100, "duration": "permanent"},
],
}
engine = ProgressiveDeliveryEngine()
flag = engine.create_feature_flag("content_homepage_v2", "New Homepage Content", default=True, rollout_pct=25)
engine.add_targeting_rule("content_homepage_v2", {"type": "user_segment", "segments": ["beta_testers"]})
users = [
{"user_id": "user_1", "segment": "beta_testers", "country": "TH"},
{"user_id": "user_2", "segment": "regular", "country": "TH"},
{"user_id": "user_3", "segment": "regular", "country": "US"},
]
for user in users:
result = engine.evaluate_flag("content_homepage_v2", user)
print(f" {user['user_id']} ({user['segment']}): {'New content' if result else 'Original'}")
canary = engine.create_canary_release("homepage", [])
print(f"\nCanary stages: {len(canary['example_stages'])}")
Feature Flags ????????? A/B Testing
A/B Testing ?????????????????? CMS content
#!/usr/bin/env python3
# ab_testing.py ??? CMS A/B Testing Framework
import json
import logging
import hashlib
import random
from typing import Dict, List
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("abtest")
class CMSABTesting:
def __init__(self):
self.experiments = {}
self.results = {}
def create_experiment(self, exp_id, name, variants, traffic_pct=100):
"""Create an A/B test experiment"""
self.experiments[exp_id] = {
"id": exp_id,
"name": name,
"variants": variants,
"traffic_percentage": traffic_pct,
"status": "running",
"results": {v["id"]: {"views": 0, "clicks": 0, "conversions": 0} for v in variants},
"created_at": "2024-01-01",
}
return self.experiments[exp_id]
def assign_variant(self, exp_id, user_id):
"""Assign a user to an experiment variant"""
exp = self.experiments.get(exp_id)
if not exp or exp["status"] != "running":
return exp["variants"][0] if exp else None
# Deterministic assignment based on user_id
hash_val = int(hashlib.md5(f"{exp_id}:{user_id}".encode()).hexdigest(), 16)
# Check if user is in traffic allocation
if (hash_val % 100) >= exp["traffic_percentage"]:
return None # Not in experiment
# Assign to variant
variant_idx = hash_val % len(exp["variants"])
variant = exp["variants"][variant_idx]
# Track view
exp["results"][variant["id"]]["views"] += 1
return variant
def track_conversion(self, exp_id, variant_id):
"""Track a conversion event"""
exp = self.experiments.get(exp_id)
if exp and variant_id in exp["results"]:
exp["results"][variant_id]["conversions"] += 1
def get_results(self, exp_id):
"""Get experiment results with statistical analysis"""
exp = self.experiments.get(exp_id)
if not exp:
return None
results = []
for variant in exp["variants"]:
vid = variant["id"]
data = exp["results"][vid]
cvr = data["conversions"] / max(data["views"], 1) * 100
results.append({
"variant": vid,
"content": variant.get("content_id", ""),
"views": data["views"],
"conversions": data["conversions"],
"conversion_rate": round(cvr, 2),
})
# Determine winner
if len(results) >= 2:
sorted_results = sorted(results, key=lambda x: x["conversion_rate"], reverse=True)
winner = sorted_results[0]
lift = ((winner["conversion_rate"] - sorted_results[1]["conversion_rate"]) /
max(sorted_results[1]["conversion_rate"], 0.01)) * 100
else:
winner = results[0] if results else None
lift = 0
return {
"experiment": exp["name"],
"status": exp["status"],
"variants": results,
"winner": winner["variant"] if winner else None,
"lift": round(lift, 1),
}
ab = CMSABTesting()
ab.create_experiment("hero_banner", "Homepage Hero Banner Test", [
{"id": "control", "content_id": "banner_original", "description": "Original banner"},
{"id": "variant_b", "content_id": "banner_new", "description": "New design with CTA"},
])
# Simulate traffic
for i in range(1000):
variant = ab.assign_variant("hero_banner", f"user_{i}")
if variant and random.random() < (0.03 if variant["id"] == "control" else 0.045):
ab.track_conversion("hero_banner", variant["id"])
results = ab.get_results("hero_banner")
print(f"Experiment: {results['experiment']}")
for v in results["variants"]:
print(f" {v['variant']}: {v['views']} views, {v['conversions']} conv, {v['conversion_rate']}% CVR")
print(f"Winner: {results['winner']} (lift: {results['lift']}%)")
API-First Content Delivery
????????? Directus API ?????????????????? content delivery
# === Directus API Content Delivery ===
# 1. REST API Examples
# Get all published articles
curl -s "https://cms.example.com/items/articles?filter[status][_eq]=published&sort=-date_created&limit=10" \
-H "Authorization: Bearer YOUR_TOKEN" | jq .
# Get single article with related data
curl -s "https://cms.example.com/items/articles/1?fields=*, author.name, category.title, tags.*" \
-H "Authorization: Bearer YOUR_TOKEN" | jq .
# 2. GraphQL Query
cat > query.graphql << 'EOF'
query GetArticles($status: String!, $limit: Int!) {
articles(
filter: { status: { _eq: $status } }
sort: ["-date_created"]
limit: $limit
) {
id
title
slug
excerpt
content
featured_image {
id
filename_download
}
author {
first_name
last_name
}
category {
title
slug
}
date_created
date_updated
}
}
EOF
# 3. SDK Integration (JavaScript)
cat > directus-client.js << 'JSEOF'
import { createDirectus, rest, readItems, readItem } from '@directus/sdk';
const client = createDirectus('https://cms.example.com').with(rest());
// Get published articles
async function getArticles(page = 1, limit = 10) {
return await client.request(
readItems('articles', {
filter: { status: { _eq: 'published' } },
sort: ['-date_created'],
limit,
page,
fields: ['id', 'title', 'slug', 'excerpt', 'featured_image', 'date_created'],
})
);
}
// Get article with feature flag check
async function getArticleWithFlags(slug, userId) {
const article = await client.request(
readItem('articles', slug, {
fields: ['*', 'author.*', 'category.*'],
})
);
// Check for A/B test variant
const variant = await checkFeatureFlag('article_layout_v2', userId);
return {
...article,
layout: variant ? 'new_layout' : 'default_layout',
};
}
console.log('Directus client configured');
JSEOF
# 4. Webhook Configuration for CDN Purge
cat > webhook_handler.py << 'PYEOF'
#!/usr/bin/env python3
from fastapi import FastAPI, Request
import httpx
app = FastAPI()
@app.post("/webhook/content-updated")
async def content_updated(request: Request):
payload = await request.json()
collection = payload.get("collection")
keys = payload.get("keys", [])
# Purge CDN cache for updated content
for key in keys:
cache_key = f"/{collection}/{key}"
# Cloudflare purge
async with httpx.AsyncClient() as client:
await client.post(
"https://api.cloudflare.com/client/v4/zones/ZONE_ID/purge_cache",
json={"files": [f"https://example.com{cache_key}"]},
headers={"Authorization": "Bearer CF_TOKEN"},
)
return {"status": "purged", "keys": keys}
PYEOF
echo "API content delivery configured"
Monitoring ????????? Rollback
Monitor content delivery ????????? rollback
#!/usr/bin/env python3
# content_monitor.py ??? Content Delivery Monitoring
import json
import logging
from typing import Dict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("monitor")
class ContentDeliveryMonitor:
def __init__(self):
self.metrics = {}
def dashboard(self):
return {
"content_delivery": {
"total_api_requests_today": 125000,
"cache_hit_rate": 94.5,
"avg_response_time_ms": 45,
"p95_response_time_ms": 120,
"error_rate": 0.1,
},
"progressive_delivery": {
"active_experiments": 3,
"active_feature_flags": 12,
"canary_releases": 1,
"rollbacks_today": 0,
},
"content_stats": {
"total_items": 2500,
"published": 2100,
"draft": 350,
"archived": 50,
"updated_today": 15,
},
}
def rollback_strategy(self):
return {
"automatic_rollback": {
"triggers": [
"Error rate > 5% for 5 minutes",
"Response time > 500ms (p95) for 10 minutes",
"Conversion rate drops > 20% compared to baseline",
],
"action": "Disable feature flag, revert to original content",
},
"manual_rollback": {
"steps": [
"1. Identify issue in monitoring dashboard",
"2. Disable feature flag / stop experiment",
"3. Purge CDN cache",
"4. Verify rollback successful",
"5. Investigate root cause",
"6. Fix and re-deploy with smaller rollout",
],
},
"content_versioning": {
"directus_revisions": "Directus tracks all content changes automatically",
"rollback_content": "Revert to previous revision via API or Admin UI",
"api_endpoint": "POST /revisions/:id/revert",
},
}
monitor = ContentDeliveryMonitor()
dash = monitor.dashboard()
print(f"API Requests: {dash['content_delivery']['total_api_requests_today']:,}")
print(f"Cache Hit: {dash['content_delivery']['cache_hit_rate']}%")
print(f"Active Experiments: {dash['progressive_delivery']['active_experiments']}")
strategy = monitor.rollback_strategy()
print(f"\nAuto-rollback triggers: {len(strategy['automatic_rollback']['triggers'])}")
for trigger in strategy["automatic_rollback"]["triggers"]:
print(f" - {trigger}")
FAQ ??????????????????????????????????????????
Q: Directus ????????? Strapi ???????????????????????????????????????????
A: Directus ??????????????????????????? existing database ?????????????????? ???????????????????????? schema ???????????? ?????????????????? SQL databases ??????????????????????????? ?????? REST + GraphQL API ??????????????????????????? Data Studio UI ????????? ?????????????????????????????? ??????????????????????????? TypeScript + Vue.js Strapi ??????????????? schema ???????????? Content-Type Builder ????????? own database schema ?????????????????? PostgreSQL, MySQL, SQLite ?????? REST + GraphQL ???????????? plugin Community ???????????????????????? Marketplace ?????? plugins ???????????? ??????????????????????????? JavaScript + React ??????????????? Directus ????????????????????? existing database ????????????????????? flexibility ????????? ??????????????? Strapi ??????????????????????????????????????????????????? ????????????????????? ecosystem ?????????????????????
Q: Progressive Delivery ????????????????????? Continuous Delivery ??????????????????????
A: Continuous Delivery deploy ????????? change ?????? production ??????????????????????????? ?????????????????????????????????????????????????????? ??????????????? bug ?????????????????????????????????????????????????????? Progressive Delivery deploy change ???????????????????????? ??????????????????????????????????????????????????? (1%) ????????????????????????????????? (5%, 25%, 100%) ????????? feature flags ?????????????????? ???????????????????????????????????? rollback ????????????????????????????????? ?????????????????? Continuous Delivery + safety net ?????????????????? CMS content Progressive Delivery ??????????????????????????? content ????????????????????? users ???????????????????????? ??????????????? (engagement, conversion) ???????????????????????? rollout ????????????????????????
Q: Feature Flags ?????????????????????????????????????????????????
A: ?????? Technical debt ?????????????????? cleanup flags ??????????????????????????????????????? code ????????????????????????????????????????????????????????????, Testing complexity ???????????? test ????????? combination ????????? flags, Performance overhead evaluate flags ????????? request (????????????????????? caching), Management ?????????????????? process ?????????????????? flags lifecycle ?????????????????????????????? ???????????? expiration date ??????????????????????????? flag, Review flags ????????? sprint ?????? flags ???????????????????????????, ????????? feature flag service (LaunchDarkly, Unleash, Flagsmith) ????????????????????????, Document ????????? flag ???????????????????????????????????? ??????????????????????????????????????????
Q: Directus ?????????????????? multi-language content ??????????
A: ?????????????????? Directus ?????? built-in translations interface ??????????????? translations collection ??????????????????????????? content collection ???????????? ??????????????? languages ??????????????????????????? (TH, EN, JA, etc.) ??????????????? content item ?????? translations ???????????????????????? API query ???????????? ?deep[translations][_filter][languages_code][_eq]=th ????????? content ?????????????????????????????????????????? ?????????????????? Progressive Delivery ??????????????? translation ????????????????????? users ??????????????????????????????????????? ???????????? ???????????????????????? rollout
