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