feat: add workflow steps CRUD (Phase 4)

README.md

@@ -0,0 +1,138 @@
+# Distributed Workflow Engine
+
+A distributed workflow engine built with FastAPI, SQLAlchemy 2.0 (async), and Python 3.12+.
+
+## Features
+
+- **Async API** — FastAPI with app factory pattern and lifespan context manager
+- **Authentication** — JWT access/refresh tokens with Argon2 password hashing
+- **Workflow CRUD** — Create, read, update, and delete workflows scoped per user
+- **Async Database** — SQLAlchemy 2.0 async with PostgreSQL (asyncpg) in production, SQLite (aiosqlite) in tests
+- **Observability** — Structured logging with structlog
+- **Migrations** — Alembic for database schema management
+- **CI/CD** — GitHub Actions with ruff linting, mypy type checking, and pytest
+
+## Project Structure
+
+```
+backend/
+├── dwf/
+│   ├── main.py                  # App factory + lifespan
+│   ├── settings.py              # Pydantic Settings (env-based config)
+│   ├── api/
+│   │   ├── deps.py              # Shared dependencies (get_db, get_current_user)
+│   │   └── routers/
+│   │       ├── auth.py          # Register, login, refresh
+│   │       ├── health.py        # Health check endpoint
+│   │       └── workflows.py     # Workflow CRUD endpoints
+│   ├── domain/models/
+│   │   ├── user.py              # Pydantic schemas for auth
+│   │   └── workflow.py          # Pydantic schemas + WorkflowStatus enum
+│   ├── infrastructure/database/
+│   │   ├── engine.py            # Async engine + session factory
+│   │   └── models/
+│   │       ├── base.py          # SQLAlchemy DeclarativeBase
+│   │       ├── user.py          # User table
+│   │       └── workflow.py      # Workflow table
+│   └── services/
+│       ├── auth_service.py      # Auth logic (register, login, JWT)
+│       └── workflow_service.py  # Workflow CRUD logic
+├── tests/
+│   ├── conftest.py              # Async fixtures (engine, session, client)
+│   └── unit/
+│       ├── test_auth.py         # 8 auth tests
+│       └── test_workflow.py     # 8 workflow tests
+├── migration/                   # Alembic migrations
+├── Dockerfile
+├── requirements.txt
+└── pyproject.toml
+```
+
+## Getting Started
+
+### Prerequisites
+
+- Python 3.12+
+- PostgreSQL (or Neon for serverless)
+
+### Setup
+
+```bash
+cd backend
+
+# Create virtual environment
+python -m venv .venv
+source .venv/bin/activate
+
+# Install dependencies
+pip install -r requirements.txt
+
+# Configure environment
+cp .env.example .env
+# Edit .env with your DATABASE_URL and JWT_SECRET_KEY
+```
+
+### Environment Variables
+
+| Variable | Description | Example |
+|---|---|---|
+| `DATABASE_URL` | Async database connection string | `postgresql+asyncpg://user:pass@host/db` |
+| `JWT_SECRET_KEY` | Secret key for JWT signing | `openssl rand -hex 32` |
+| `LOG_LEVEL` | Logging level (default: `info`) | `debug` |
+
+### Run
+
+```bash
+uvicorn dwf.main:app --reload
+```
+
+API docs available at `http://localhost:8000/docs`.
+
+### Test
+
+```bash
+cd backend
+pytest tests/ -v
+```
+
+### Lint
+
+```bash
+ruff check backend/
+ruff format --check backend/
+```
+
+### Docker
+
+```bash
+docker compose build
+docker compose up
+```
+
+## API Endpoints
+
+| Method | Path | Description | Auth |
+|---|---|---|---|
+| `GET` | `/health` | Health check | No |
+| `POST` | `/auth/register` | Register new user | No |
+| `POST` | `/auth/login` | Login, get tokens | No |
+| `POST` | `/auth/refresh` | Refresh access token | No |
+| `POST` | `/workflows/` | Create workflow | Yes |
+| `GET` | `/workflows/` | List user's workflows | Yes |
+| `GET` | `/workflows/{id}` | Get workflow by ID | Yes |
+| `PUT` | `/workflows/{id}` | Update workflow | Yes |
+| `DELETE` | `/workflows/{id}` | Delete workflow | Yes |
+
+## Tech Stack
+
+- **Framework**: FastAPI
+- **ORM**: SQLAlchemy 2.0 (async)
+- **Database**: PostgreSQL + asyncpg
+- **Auth**: PyJWT + Argon2
+- **Validation**: Pydantic v2
+- **Config**: pydantic-settings
+- **Logging**: structlog
+- **Migrations**: Alembic
+- **Testing**: pytest + pytest-asyncio + httpx + aiosqlite
+- **Linting**: ruff + mypy
+- **CI**: GitHub Actions

backend/dwf/api/routers/step.py

@@ -0,0 +1,82 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from dwf.api.deps import get_current_active_user, get_db
+from dwf.domain.models.step import StepCreate, StepResponse, StepUpdate
+from dwf.infrastructure.database.models import User
+from dwf.services.step_service import StepService
+
+router = APIRouter(prefix="/workflows/{workflow_id}/steps", tags=["steps"])
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED, response_model=StepResponse)
+async def create_step(
+    workflow_id: str,
+    data: StepCreate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+):
+    service = StepService(db)
+    try:
+        step = await service.create_step(workflow_id, data)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
+    return step
+
+
+@router.get("/", response_model=list[StepResponse])
+async def list_steps(
+    workflow_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+):
+    service = StepService(db)
+    return await service.list_by_workflow(workflow_id)
+
+
+@router.get("/{step_id}", response_model=StepResponse)
+async def get_step(
+    workflow_id: str,
+    step_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+):
+    service = StepService(db)
+    step = await service.get_by_id(step_id, workflow_id)
+    if step is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Step not found"
+        )
+    return step
+
+
+@router.put("/{step_id}", response_model=StepResponse)
+async def update_step(
+    workflow_id: str,
+    step_id: str,
+    data: StepUpdate,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+):
+    service = StepService(db)
+    step = await service.update_step(step_id, workflow_id, data)
+    if step is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Step not found"
+        )
+    return step
+
+
+@router.delete("/{step_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_step(
+    workflow_id: str,
+    step_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+):
+    service = StepService(db)
+    deleted = await service.delete_step(step_id, workflow_id)
+    if not deleted:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Step not found"
+        )

backend/dwf/domain/models/step.py

@@ -0,0 +1,40 @@
+from datetime import datetime
+from enum import Enum
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class StepType(str, Enum):
+    HTTP = "http"
+    SCRIPT = "script"
+    EMAIL = "email"
+
+
+class StepCreate(BaseModel):
+    name: str
+    type: StepType
+    config: dict = Field(default_factory=dict)
+    position: int
+    depends_on: list[str] = Field(default_factory=list)
+
+
+class StepUpdate(BaseModel):
+    name: str | None = None
+    type: StepType | None = None
+    config: dict | None = None
+    position: int | None = None
+    depends_on: list[str] | None = None
+
+
+class StepResponse(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: UUID
+    workflow_id: UUID
+    name: str
+    type: StepType
+    config: dict
+    position: int
+    depends_on: list[str]
+    created_at: datetime
+    updated_at: datetime

backend/dwf/infrastructure/database/models/__init__.py

@@ -1,5 +1,6 @@
 from .base import Base
+from .steps import Step
 from .user import User
 from .workflow import Workflow
 
-__all__ = ["Base", "User", "Workflow"]
+__all__ = ["Base", "Step", "User", "Workflow"]

backend/dwf/infrastructure/database/models/steps.py

@@ -0,0 +1,64 @@
+from datetime import datetime
+from uuid import uuid4
+
+from sqlalchemy import JSON, DateTime, ForeignKey, String, Uuid
+from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.sql import func
+
+from .base import Base
+
+
+class Step(Base):
+    __tablename__ = "steps"
+
+    id: Mapped[str] = mapped_column(
+        Uuid,
+        primary_key=True,
+        default=uuid4,
+    )
+
+    workflow_id: Mapped[str] = mapped_column(
+        Uuid,
+        ForeignKey("workflows.id"),
+        nullable=False,
+    )
+
+    name: Mapped[str] = mapped_column(
+        String(255),
+        index=True,
+        nullable=False,
+    )
+
+    type: Mapped[str] = mapped_column(
+        String(50),
+        nullable=False,
+    )
+
+    config: Mapped[dict] = mapped_column(
+        JSON,
+        nullable=False,
+        default=dict,
+    )
+
+    position: Mapped[int] = mapped_column(
+        nullable=False,
+    )
+
+    depends_on: Mapped[list[str]] = mapped_column(
+        JSON,
+        nullable=False,
+        default=list,
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    )
+
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )

backend/dwf/main.py

@@ -39,11 +39,13 @@ def create_app(settings: Settings | None = None) -> FastAPI:
 
     from dwf.api.routers.auth import router as auth_router
     from dwf.api.routers.health import router as health_router
+    from dwf.api.routers.step import router as steps_router
     from dwf.api.routers.workflows import router as workflows_router
 
     app.include_router(health_router)
     app.include_router(auth_router)
     app.include_router(workflows_router)
+    app.include_router(steps_router)
 
     return app