SQLite Litestream Agile Scrum Kanban

SQLite Litestream Agile

SQLite Litestream Agile Scrum Kanban Streaming Replication Backup S3 WAL Sprint Planning Kanban Board WIP Limit Project Management Production

Litestream Setup

# === Litestream Configuration ===

# Installation
# curl -L https://github.com/benbjohnson/litestream/releases/latest/download/litestream-linux-amd64.tar.gz | tar xz
# sudo mv litestream /usr/local/bin/

# litestream.yml — Configuration
# dbs:
#   - path: /data/app.db
#     replicas:
#       - type: s3
#         bucket: my-backup-bucket
#         path: backups/app.db
#         region: ap-southeast-1
#         retention: 720h          # 30 days
#         retention-check-interval: 1h
#         sync-interval: 1s        # Near real-time
#
#       - type: sftp
#         host: backup-server:22
#         user: backup
#         path: /backups/app.db
#         key-path: /root/.ssh/id_rsa

# Run Litestream as sidecar
# litestream replicate -config /etc/litestream.yml

# Restore from backup
# litestream restore -config /etc/litestream.yml /data/app.db
# litestream restore -o /data/app.db s3://my-backup-bucket/backups/app.db

# Docker Compose — App + Litestream
# version: '3.8'
# services:
#   app:
#     image: my-app:latest
#     volumes:
#       - db_data:/data
#     ports:
#       - "8080:8080"
#   litestream:
#     image: litestream/litestream:latest
#     volumes:
#       - db_data:/data
#       - ./litestream.yml:/etc/litestream.yml
#     command: replicate -config /etc/litestream.yml
#     environment:
#       - AWS_ACCESS_KEY_ID=
#       - AWS_SECRET_ACCESS_KEY=

# SQLite WAL Mode (required)
# PRAGMA journal_mode=WAL;
# PRAGMA busy_timeout=5000;
# PRAGMA synchronous=NORMAL;
# PRAGMA cache_size=-20000;  -- 20MB cache

from dataclasses import dataclass

@dataclass
class BackupConfig:
    replica_type: str
    destination: str
    sync_interval: str
    retention: str
    cost: str
    use_case: str

configs = [
    BackupConfig("S3", "AWS S3 / R2 / MinIO", "1s", "30 days", "$0.023/GB/mo", "Cloud Backup"),
    BackupConfig("GCS", "Google Cloud Storage", "1s", "30 days", "$0.020/GB/mo", "GCP Users"),
    BackupConfig("Azure Blob", "Azure Blob Storage", "1s", "30 days", "$0.018/GB/mo", "Azure Users"),
    BackupConfig("SFTP", "Remote Server", "10s", "Custom", "Server cost", "On-premise"),
    BackupConfig("ABS", "Backblaze B2", "1s", "30 days", "$0.005/GB/mo", "Budget"),
]

print("=== Litestream Replica Options ===")
for c in configs:
    print(f"  [{c.replica_type}] {c.destination}")
    print(f"    Sync: {c.sync_interval} | Retention: {c.retention} | Cost: {c.cost}")

Agile Project Management

# === Agile Project Tracking with SQLite ===

# SQLite Schema for Agile Board
# CREATE TABLE projects (
#   id INTEGER PRIMARY KEY,
#   name TEXT NOT NULL,
#   methodology TEXT CHECK(methodology IN ('scrum', 'kanban', 'scrumban')),
#   created_at DATETIME DEFAULT CURRENT_TIMESTAMP
# );
#
# CREATE TABLE sprints (
#   id INTEGER PRIMARY KEY,
#   project_id INTEGER REFERENCES projects(id),
#   name TEXT NOT NULL,
#   start_date DATE,
#   end_date DATE,
#   goal TEXT,
#   status TEXT DEFAULT 'planning'
# );
#
# CREATE TABLE tasks (
#   id INTEGER PRIMARY KEY,
#   project_id INTEGER REFERENCES projects(id),
#   sprint_id INTEGER REFERENCES sprints(id),
#   title TEXT NOT NULL,
#   description TEXT,
#   status TEXT DEFAULT 'backlog',
#   priority TEXT DEFAULT 'medium',
#   story_points INTEGER DEFAULT 0,
#   assignee TEXT,
#   created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
#   updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
# );
#
# CREATE INDEX idx_tasks_status ON tasks(status);
# CREATE INDEX idx_tasks_sprint ON tasks(sprint_id);

# Sprint Velocity Query
# SELECT s.name as sprint,
#   COUNT(t.id) as tasks_completed,
#   SUM(t.story_points) as points_completed,
#   julianday(s.end_date) - julianday(s.start_date) as sprint_days
# FROM sprints s
# JOIN tasks t ON t.sprint_id = s.id
# WHERE t.status = 'done' AND s.status = 'completed'
# GROUP BY s.id
# ORDER BY s.start_date DESC LIMIT 5;

@dataclass
class SprintMetric:
    sprint: str
    planned_points: int
    completed_points: int
    tasks_done: int
    velocity: float
    burndown: str

sprints = [
    SprintMetric("Sprint 10", 34, 31, 12, 31.0, "On track"),
    SprintMetric("Sprint 11", 32, 35, 14, 35.0, "Ahead"),
    SprintMetric("Sprint 12", 36, 28, 10, 28.0, "Behind — blocked by API"),
    SprintMetric("Sprint 13", 30, 32, 13, 32.0, "On track"),
    SprintMetric("Sprint 14", 33, 30, 11, 30.0, "Slight delay"),
]

avg_velocity = sum(s.velocity for s in sprints) / len(sprints)
print("=== Sprint Metrics ===")
for s in sprints:
    completion = s.completed_points / s.planned_points * 100
    print(f"  [{s.sprint}] Planned: {s.planned_points} | Done: {s.completed_points} ({completion:.0f}%)")
    print(f"    Tasks: {s.tasks_done} | Status: {s.burndown}")
print(f"\n  Average Velocity: {avg_velocity:.1f} points/sprint")

Kanban Board

# === Kanban Board Implementation ===

# Kanban Columns with WIP Limits
# Backlog → To Do (WIP: 5) → In Progress (WIP: 3) → Review (WIP: 2) → Done

@dataclass
class KanbanColumn:
    column: str
    wip_limit: int
    current: int
    tasks: list

board = [
    KanbanColumn("Backlog", 999, 8, ["Setup CI/CD", "Write API docs", "DB migration", "Add caching", "Auth refactor", "Error handling", "Logging", "Tests"]),
    KanbanColumn("To Do", 5, 3, ["Implement search API", "User dashboard UI", "Payment integration"]),
    KanbanColumn("In Progress", 3, 2, ["Deploy Litestream backup", "Sprint review presentation"]),
    KanbanColumn("Review", 2, 1, ["Fix login bug #234"]),
    KanbanColumn("Done", 999, 15, ["..."]),
]

print("=== Kanban Board ===")
for col in board:
    status = "OK" if col.current <= col.wip_limit else "OVER WIP!"
    print(f"  [{col.column}] {col.current}/{col.wip_limit} [{status}]")
    if col.column != "Done":
        for t in col.tasks[:3]:
            print(f"    - {t}")
        if len(col.tasks) > 3:
            print(f"    ... and {len(col.tasks)-3} more")

# Cycle Time & Lead Time
cycle_times = [2.5, 1.0, 3.0, 1.5, 4.0, 2.0, 1.0, 3.5, 2.0, 1.5]
lead_times = [5.0, 3.0, 7.0, 4.0, 8.0, 5.0, 3.0, 6.0, 4.0, 3.5]

avg_cycle = sum(cycle_times) / len(cycle_times)
avg_lead = sum(lead_times) / len(lead_times)
throughput = len(cycle_times) / 2  # tasks per week

print(f"\n  Flow Metrics:")
print(f"    Avg Cycle Time: {avg_cycle:.1f} days")
print(f"    Avg Lead Time: {avg_lead:.1f} days")
print(f"    Throughput: {throughput:.0f} tasks/week")

best_practices = {
    "WIP Limits": "จำกัดงาน In Progress ป้องกัน Context Switching",
    "Pull System": "ดึงงานเมื่อพร้อม ไม่ Push งานเข้า",
    "Visualize": "Board แสดงสถานะงานทั้งหมด",
    "Measure Flow": "วัด Cycle Time Lead Time Throughput",
    "Continuous Improvement": "ปรับ WIP Limits Process ตลอด",
    "Litestream Backup": "Backup SQLite Board ทุกวินาที",
}

print(f"\n  Best Practices:")
for k, v in best_practices.items():
    print(f"    [{k}]: {v}")

เคล็ดลับ

• WAL Mode: ใช้ WAL Mode เสมอกับ SQLite + Litestream
• S3 Backup: Litestream → S3 ราคาถูก Recovery เร็ว
• WIP Limit: จำกัด WIP ลด Context Switching เพิ่ม Focus
• Velocity: ใช้ Average Velocity วางแผน Sprint ถัดไป
• Retrospective: ทำ Retrospective ทุก Sprint ปรับปรุงกระบวนการ

Litestream คืออะไร

Streaming Replication SQLite Continuous Backup S3 Azure GCS SFTP WAL Near Real-time Recovery Sidecar Process ไม่ต้องหยุด Application

สรุป

SQLite Litestream Agile Scrum Kanban Streaming Replication S3 Backup WAL Sprint Velocity Kanban Board WIP Limit Cycle Time Production Workflow