---
title: CRUD Application
---

# 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

```yaml
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

```python
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`:

```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

```bash
$ 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:

```bash
$ 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

```bash
$ sysg stop --config crud.sysg.yaml
```

### View logs

```bash
$ sysg logs --service fastapi_server
```

### Check status

```bash
$ sysg status
```

## Example usage

### Create a Todo
```bash
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
```bash
curl http://localhost:8888/todos
```

### Update a Todo
```bash
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
```bash
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

- [Configuration](../how-it-works/configuration) - Service definitions
- [Hello World](./hello-world) - Simple example to get started
- [Orchestrator](./orchestrator) - Multi-agent task execution
