SQLite Litestream Clean Architecture — Replicate

SQLite Litestream

SQLite Litestream Clean Architecture WAL Replication S3 Backup Disaster Recovery Point-in-time Single Server Edge Computing Production

Litestream Setup

=== Litestream Installation and Config ===

อ่านเพิ่ม: MinIO S3 Compatible Storage self-hosted ทดแทน AWS S3 · อ่านเพิ่ม: LXC vs Docker เลือก Container Technology อะไรดี · อ่านเพิ่ม: Proxmox VE Cluster ทำ High Availability สำหรับ Home Lab

Install

curl -fsSL https://github.com/benbjohnson/litestream/releases/download/v0.3.13/litestream-v0.3.13-linux-amd64.tar.gz | tar xz

sudo mv litestream /usr/local/bin/

litestream.yml

dbs:

• path: /data/app.db

replicas:

• type: s3

bucket: my-backup-bucket

path: app.db

region: ap-southeast-1

retention: 720h # 30 days

sync-interval: 1s

snapshot-interval: 4h

Start replication

litestream replicate -config litestream.yml

Restore database

litestream restore -config litestream.yml -o /data/app.db

Point-in-time restore

litestream restore -config litestream.yml -o /data/app.db -timestamp 2024-01-15T10:30:00Z

Systemd service

[Unit]

Description=Litestream Replication

After=network.target

[Service]

Type=simple

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง uncore frequency ram คือ — คู่มือฉบับสมบูรณ์ 2026

ExecStart=/usr/local/bin/litestream replicate -config /etc/litestream.yml

Restart=always

RestartSec=5

Environment=AWS_ACCESS_KEY_ID=xxx

Environment=AWS_SECRET_ACCESS_KEY=xxx

[Install]

WantedBy=multi-user.target

แนะนำเพิ่มเติม — ระบบเทรดของ iCafeForex

from dataclasses import dataclass

@dataclass

class ReplicaConfig:

destination: str

type: str

retention: str

sync_interval: str

cost_gb_month: float

latency: str

replicas = [

ReplicaConfig("AWS S3", "s3", "30 days", "1s", 0.023, "~100ms"),

ReplicaConfig("Google Cloud Storage", "gcs", "30 days", "1s", 0.020, "~100ms"),

ReplicaConfig("Azure Blob Storage", "abs", "30 days", "1s", 0.018, "~100ms"),

ReplicaConfig("S3-compatible (MinIO)", "s3", "30 days", "1s", 0.0, "~10ms (local)"),

ReplicaConfig("Backblaze B2", "s3", "30 days", "1s", 0.005, "~150ms"),

]

print("=== Replica Destinations ===")

for r in replicas:

print(f" [{r.destination}] Type: {r.type}")

print(f" Retention: {r.retention} | Sync: {r.sync_interval}")

เนื้อหาเกี่ยวข้อง — อ่านต่อ: Redis Cluster Batch Processing Pipeline

print(f" Cost: /GB/month | Latency: {r.latency}")

Clean Architecture

=== Clean Architecture with SQLite ===

Project Structure:

app/

├── domain/ # Entity & Business Rules

│ ├── entity.py

│ └── repository.py # Interface

├── usecase/ # Application Logic

│ └── user_usecase.py

├── infrastructure/ # SQLite Implementation

│ └── sqlite_repo.py

├── delivery/ # HTTP / CLI

│ └── http_handler.py

└── main.py

แนะนำเพิ่มเติม — ดูสัญญาณเทรดที่ XM Signal

domain/entity.py

@dataclass

class User:

id: int

name: str

email: str

created_at: datetime

domain/repository.py (Interface)

from abc import ABC, abstractmethod

class UserRepository(ABC):

@abstractmethod

def get_by_id(self, user_id: int) -> User: ...

@abstractmethod

เนื้อหาเกี่ยวข้อง — บทความที่เกี่ยวข้อง: Midjourney Prompt Consensus Algorithm

def create(self, user: User) -> User: ...

@abstractmethod

def list_all(self) -> list[User]: ...

infrastructure/sqlite_repo.py

class SQLiteUserRepository(UserRepository):

def __init__(self, db_path: str):

self.conn = sqlite3.connect(db_path)

self._create_table()

def get_by_id(self, user_id: int) -> User:

row = self.conn.execute(

"SELECT id, name, email, created_at FROM users WHERE id = ?",

(user_id,)).fetchone()

return User(*row) if row else None

def create(self, user: User) -> User:

self.conn.execute(

"INSERT INTO users (name, email) VALUES (?, ?)",

(user.name, user.email))

self.conn.commit()

return user

usecase/user_usecase.py

class UserUseCase:

def __init__(self, repo: UserRepository):

self.repo = repo

def get_user(self, user_id: int) -> User:

return self.repo.get_by_id(user_id)

def create_user(self, name: str, email: str) -> User:

user = User(id=0, name=name, email=email, created_at=datetime.now())

return self.repo.create(user)

@dataclass

class ArchLayer:

layer: str

responsibility: str

depends_on: str

example: str

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง oVirt Virtualization Developer Experience DX สร้าง Self-Service VM Platform

layers = [

ArchLayer("Entity (Domain)", "Business objects and rules",

"Nothing (innermost layer)",

"User, Product, Order dataclass"),

ArchLayer("Repository Interface", "Data access contract",

"Entity only",

"UserRepository ABC with get, create, list methods"),

ArchLayer("Use Case", "Application business logic",

"Entity + Repository Interface",

"CreateUserUseCase, GetOrderUseCase"),

ArchLayer("Infrastructure", "Concrete implementations",

"Entity + Repository Interface",

"SQLiteUserRepo, S3FileStorage"),

ArchLayer("Delivery", "HTTP/CLI/gRPC handlers",

"Use Case",

"FastAPI routes, Click CLI commands"),

]

print("=== Clean Architecture Layers ===")

for l in layers:

print(f" [{l.layer}]")

print(f" Responsibility: {l.responsibility}")

print(f" Depends on: {l.depends_on}")

print(f" Example: {l.example}")

Production Deployment

# === Production Checklist ===

@dataclass
class CheckItem:
    category: str
    item: str
    config: str
    importance: str

checklist = [
    CheckItem("SQLite", "WAL mode enabled",
        "PRAGMA journal_mode=WAL;", "Critical"),
    CheckItem("SQLite", "Busy timeout set",
        "PRAGMA busy_timeout=5000;", "Critical"),
    CheckItem("SQLite", "Foreign keys enabled",
        "PRAGMA foreign_keys=ON;", "High"),
    CheckItem("SQLite", "Synchronous mode",
        "PRAGMA synchronous=NORMAL;", "High"),
    CheckItem("Litestream", "Replication running",
        "systemctl status litestream", "Critical"),
    CheckItem("Litestream", "Retention configured",
        "retention: 720h (30 days)", "High"),
    CheckItem("Litestream", "Restore tested",
        "litestream restore -o test.db monthly", "Critical"),
    CheckItem("Monitoring", "Replication lag alert",
        "Alert if no WAL segments for 5 minutes", "High"),
    CheckItem("Monitoring", "Database size tracking",
        "Track file size, warn at 80% disk", "Medium"),
    CheckItem("Backup", "S3 versioning enabled",
        "aws s3api put-bucket-versioning", "High"),
    CheckItem("Security", "Database file permissions",
        "chmod 600 app.db; chown app:app app.db", "High"),
    CheckItem("Performance", "Connection pooling",
        "Single writer, multiple readers via WAL", "High"),
]

print("=== Production Checklist ===")
for c in checklist:
    print(f"  [{c.importance}] {c.category}: {c.item}")
    print(f"    Config: {c.config}")

# Sizing guide
sizing = {
    "< 1 GB database": "Single SQLite + Litestream, any VPS",
    "1-10 GB database": "SQLite + Litestream, SSD VPS, 4GB+ RAM",
    "10-50 GB database": "Consider PostgreSQL migration",
    "> 50 GB database": "Definitely PostgreSQL or distributed DB",
    "< 100 writes/sec": "SQLite handles well",
    "100-1000 writes/sec": "SQLite with WAL, tune busy_timeout",
    "> 1000 writes/sec": "Need PostgreSQL or similar",
}

print(f"\n\nSizing Guide:")
for k, v in sizing.items():
    print(f"  [{k}]: {v}")

เคล็ดลับ

• WAL: เปิด WAL mode เสมอ ให้ Read ไม่ Block Write
• Restore Test: ทดสอบ Restore ทุกเดือน ไม่ใช่รอตอนเกิดปัญหา
• Single Writer: SQLite รองรับ Single Writer เท่านั้น ใช้ Mutex ป้องกัน
• S3 Lifecycle: ตั้ง S3 Lifecycle Rule ลบ Backup เก่าอัตโนมัติ
• Clean Arch: แยก Repository Interface ออกจาก Implementation เปลี่ยน DB ง่าย

Litestream คืออะไร

Open Source Replicate SQLite S3 Cloud Storage Real-time WAL Write-Ahead Log Point-in-time Recovery Embedded Single Server Edge Computing