Skip to main content

Documentation Index

Fetch the complete documentation index at: https://sysg.dev/llms.txt

Use this file to discover all available pages before exploring further.

​
CRUD Application

A simple FastAPI CRUD service for testing systemg service management capabilities.

​
Features demonstrated

  • Service management with sysg start and sysg stop
  • Automatic recovery from failures
  • Zero-downtime rolling deployments
  • Modern Python with FastAPI and uvicorn
  • Package management with uv

​
Configuration

version: "1"
logs:
  sink: file
  max_bytes: 10485760
  max_files: 5
status:
  snapshot_mode: summary
  snapshot_interval_secs: 5

services:
  fastapi_server:
    command: "uv run uvicorn main:app --host 0.0.0.0 --port 8888"
    deployment_strategy: "rolling_start"
    restart_policy: "on_failure"
    retries: "10"
    backoff: "5s"
File logging is enabled here so sysg logs --service fastapi_server works out of the box. For production services whose output is already collected elsewhere, set logs.sink: none. Summary status snapshots keep sysg status and sysg inspect responsive by serving cached supervisor state instead of rebuilding live process details per command.

​
Application code

​
main.py

import random
from datetime import datetime, timezone
from typing import Dict, Optional

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="CRUD API", version="0.1.0")

class Todo(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str = Field(..., min_length=1, max_length=1000)
    id: Optional[int] = Field(default=None)
    timestamp: Optional[datetime] = Field(default=None)
    is_completed: bool = Field(default=False)

todos_db: Dict[int, Todo] = {}
next_id = 1

@app.get("/")
async def root():
    return {"status": "healthy", "timestamp": datetime.now(timezone.utc).isoformat()}

@app.post("/todos", response_model=Todo)
async def create_todo(todo: Todo) -> Todo:
    global next_id
    todo.id = next_id
    todo.timestamp = datetime.now(timezone.utc)
    next_id += 1
    todos_db[todo.id] = todo
    return todo

@app.get("/todos", response_model=list[Todo])
async def read_todos() -> list[Todo]:
    return list(todos_db.values())

@app.get("/todos/{todo_id}", response_model=Todo)
async def read_todo(todo_id: int) -> Todo:
    if todo_id not in todos_db:
        raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")
    return todos_db[todo_id]

@app.put("/todos/{todo_id}", response_model=Todo)
async def update_todo(todo_id: int, todo_update: Todo) -> Todo:
    if todo_id not in todos_db:
        raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")

    existing_todo = todos_db[todo_id]
    existing_todo.title = todo_update.title
    existing_todo.description = todo_update.description
    existing_todo.is_completed = todo_update.is_completed
    return existing_todo

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    if todo_id not in todos_db:
        raise HTTPException(status_code=404, detail=f"Todo with id {todo_id} not found")

    del todos_db[todo_id]
    return {"message": f"Todo {todo_id} deleted successfully"}

@app.get("/chaos")
async def chaos_endpoint():
    if random.random() < 0.7:
        raise HTTPException(
            status_code=500,
            detail="Chaos monkey struck! Random failure to test recovery."
        )
    return {"message": "Lucky you! The chaos monkey was sleeping."}

​
Dependencies

Create pyproject.toml: 
[project]
name = "crud-example"
version = "0.1.0"
description = "Simple CRUD API with FastAPI demonstrating systemg recovery"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.115.0",
    "uvicorn[standard]>=0.32.0",
    "pydantic>=2.10.0",
    "httpx>=0.27.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["."]

​
Run it

$ cd examples/crud
$ uv sync
$ sysg start --config crud.sysg.yaml

​
API endpoints

  • GET / - Health check
  • POST /todos - Create a new todo
  • GET /todos - List all todos
  • GET /todos/{id} - Get a specific todo
  • PUT /todos/{id} - Update a todo
  • DELETE /todos/{id} - Delete a todo
  • GET /chaos - Random failure endpoint (70% failure rate)

​
Testing

Run the test script to verify all endpoints: 
$ uv run python test_api.py
The test script will:
  1. Create a new todo
  2. Read all todos
  3. Update the todo
  4. Get a specific todo
  5. Test the chaos endpoint (demonstrates recovery)
  6. Delete the todo

​
Operations

​
Stop the service

$ sysg stop --config crud.sysg.yaml

​
View logs

$ sysg logs --service fastapi_server

​
Check status

$ sysg status

​
Example usage

​
Create a Todo

curl -X POST http://localhost:8888/todos \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Test systemg",
    "description": "Verify systemg service management",
    "is_completed": false
  }'

​
List Todos

curl http://localhost:8888/todos

​
Update a Todo

curl -X PUT http://localhost:8888/todos/1 \
  -H "Content-Type: application/json" \
  -d '{
    "title": "Test systemg",
    "description": "Successfully tested systemg!",
    "is_completed": true
  }'

​
Test chaos endpoint

curl http://localhost:8888/chaos

​
What happens

  1. Service starts with sysg start and serves the API on port 8888
  2. In-memory storage using Python dict for simplicity
  3. Chaos endpoint randomly fails to demonstrate recovery capabilities
  4. Automatic restarts when service crashes (up to 10 retries with 5s backoff)
  5. Rolling deployments ensure zero downtime during updates

​
Interactive API docs

FastAPI provides automatic interactive documentation at:
  • Swagger UI: http://localhost:8888/docs
  • ReDoc: http://localhost:8888/redoc

​
See also