Advanced: Multi-Step Code Generation Workflows
Large systems—web applications, microservices, data pipelines—rarely fit in a single code generation prompt. Instead, you break the project into logical layers or components, generate each with focused prompts, then integrate them. This article teaches advanced workflow patterns for generating complex codebases layer by layer.
The Layered Generation Approach
Instead of one massive prompt, organize generation into distinct layers:
Layer 1: Data Models & Schemas
→ Define dataclasses, database models, API schemas
Layer 2: Database Access & Repositories
→ Generate CRUD operations, queries, migrations
Layer 3: Business Logic & Services
→ Generate rule engines, calculations, workflows
Layer 4: API Endpoints & Controllers
→ Generate request/response handlers
Layer 5: Integration & Tests
→ Generate client code, integration tests, documentation
Each layer depends on the previous one. Layer 2 needs Layer 1's data models. Layer 3 needs Layer 2's repositories. And so on.
Multi-Step Workflow Example: Building a Todo API
Let's generate a complete REST API for a todo application:
Step 1: Define Data Models
Language: Python 3.11
Framework: FastAPI + SQLAlchemy 2.0
Database: PostgreSQL
Generate: Data models (dataclasses) and SQLAlchemy ORM classes
Requirements:
1. Todo model with fields: id (UUID), title (str), description (str), completed (bool), due_date (datetime)
2. User model with fields: id (UUID), email (str), password_hash (str), created_at (datetime)
3. TodoItem (association) linking Todo to User
4. Use SQLAlchemy declarative syntax (from sqlalchemy.orm import declarative_base)
5. Include validation: title required and non-empty, email valid format
6. Include timestamps: created_at, updated_at on all models
7. Use proper relationships: User.todos (one-to-many)
Output: models.py with all classes and relationships
Generated output: models.py
from sqlalchemy import Column, String, Boolean, DateTime, ForeignKey, UUID
from sqlalchemy.orm import declarative_base, relationship
from datetime import datetime
import uuid
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(UUID, primary_key=True, default=uuid.uuid4)
email = Column(String, unique=True, nullable=False)
password_hash = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
todos = relationship("Todo", back_populates="owner")
class Todo(Base):
__tablename__ = "todos"
id = Column(UUID, primary_key=True, default=uuid.uuid4)
title = Column(String, nullable=False)
description = Column(String)
completed = Column(Boolean, default=False)
due_date = Column(DateTime)
owner_id = Column(UUID, ForeignKey("users.id"), nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
owner = relationship("User", back_populates="todos")
Step 2: Generate Database Repository Layer
Language: Python 3.11
Framework: FastAPI + SQLAlchemy 2.0
Context: Use the User and Todo models from models.py (provided above)
Generate: Repository classes for data access
Requirements:
1. TodoRepository class with methods:
- create(db: Session, user_id: UUID, title: str, description: str, due_date: datetime | None) -> Todo
- get_by_id(db: Session, todo_id: UUID) -> Todo | None
- list_for_user(db: Session, user_id: UUID) -> list[Todo]
- update(db: Session, todo_id: UUID, title: str | None, completed: bool | None) -> Todo
- delete(db: Session, todo_id: UUID) -> bool
2. UserRepository class with methods:
- create(db: Session, email: str, password_hash: str) -> User
- get_by_email(db: Session, email: str) -> User | None
- get_by_id(db: Session, user_id: UUID) -> User | None
3. Use SQLAlchemy 2.0 select() syntax (not session.query())
4. Proper error handling: raise ValueError for invalid inputs, return None for not-found
5. No bare unwrap(); handle None gracefully
6. Docstrings with examples
Output: repositories.py
Generated output: repositories.py
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import User, Todo
from uuid import UUID
from datetime import datetime
class TodoRepository:
@staticmethod
def create(db: Session, user_id: UUID, title: str, description: str = None, due_date: datetime = None) -> Todo:
"""Create a new todo."""
if not title or not title.strip():
raise ValueError("Title cannot be empty")
todo = Todo(title=title, description=description, due_date=due_date, owner_id=user_id)
db.add(todo)
db.commit()
return todo
@staticmethod
def get_by_id(db: Session, todo_id: UUID) -> Todo | None:
"""Get a todo by ID."""
stmt = select(Todo).where(Todo.id == todo_id)
return db.execute(stmt).scalar_one_or_none()
@staticmethod
def list_for_user(db: Session, user_id: UUID) -> list[Todo]:
"""List all todos for a user."""
stmt = select(Todo).where(Todo.owner_id == user_id).order_by(Todo.created_at.desc())
return db.execute(stmt).scalars().all()
@staticmethod
def update(db: Session, todo_id: UUID, title: str | None = None, completed: bool | None = None) -> Todo:
"""Update a todo."""
todo = TodoRepository.get_by_id(db, todo_id)
if not todo:
raise ValueError(f"Todo {todo_id} not found")
if title is not None:
todo.title = title
if completed is not None:
todo.completed = completed
db.commit()
return todo
@staticmethod
def delete(db: Session, todo_id: UUID) -> bool:
"""Delete a todo."""
todo = TodoRepository.get_by_id(db, todo_id)
if not todo:
return False
db.delete(todo)
db.commit()
return True
class UserRepository:
@staticmethod
def create(db: Session, email: str, password_hash: str) -> User:
"""Create a new user."""
if "@" not in email:
raise ValueError("Invalid email format")
user = User(email=email, password_hash=password_hash)
db.add(user)
db.commit()
return user
@staticmethod
def get_by_email(db: Session, email: str) -> User | None:
"""Get a user by email."""
stmt = select(User).where(User.email == email)
return db.execute(stmt).scalar_one_or_none()
@staticmethod
def get_by_id(db: Session, user_id: UUID) -> User | None:
"""Get a user by ID."""
stmt = select(User).where(User.id == user_id)
return db.execute(stmt).scalar_one_or_none()
Step 3: Generate API Endpoints
Language: Python 3.11
Framework: FastAPI
Context:
- Use repositories from repositories.py (above)
- Use models from models.py
- FastAPI version 0.104+
- Already has: get_db() dependency for sessions, authenticated user extraction
Generate: API routes for todos
Requirements:
1. POST /todos - Create a new todo (authenticated)
Input: {title, description, due_date}
Output: {id, title, completed, created_at}
2. GET /todos/{todo_id} - Retrieve a todo (authenticated, user must own it)
Output: Full todo object
3. GET /todos - List all todos for authenticated user
Output: list[todo]
4. PATCH /todos/{todo_id} - Update a todo
Input: {title?, completed?}
Output: Updated todo
5. DELETE /todos/{todo_id} - Delete a todo
Response: {deleted: true}
Error handling: 404 for not found, 403 for unauthorized, 400 for invalid input
Generated output: routes.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.repositories import TodoRepository, UserRepository
from app.models import User, Todo
from uuid import UUID
from datetime import datetime
router = APIRouter(prefix="/todos", tags=["todos"])
def get_current_user(db: Session = Depends(get_db)) -> User:
"""Get current authenticated user (already implemented elsewhere)"""
pass
@router.post("/")
def create_todo(
title: str, description: str = None, due_date: datetime = None,
current_user: User = Depends(get_current_user), db: Session = Depends(get_db)
):
"""Create a new todo."""
todo = TodoRepository.create(db, current_user.id, title, description, due_date)
return {"id": todo.id, "title": todo.title, "completed": todo.completed, "created_at": todo.created_at}
@router.get("/{todo_id}")
def get_todo(todo_id: UUID, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""Get a specific todo."""
todo = TodoRepository.get_by_id(db, todo_id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized")
return todo
@router.get("/")
def list_todos(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""List all todos for the current user."""
return TodoRepository.list_for_user(db, current_user.id)
@router.patch("/{todo_id}")
def update_todo(
todo_id: UUID, title: str = None, completed: bool = None,
current_user: User = Depends(get_current_user), db: Session = Depends(get_db)
):
"""Update a todo."""
todo = TodoRepository.get_by_id(db, todo_id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized")
updated = TodoRepository.update(db, todo_id, title, completed)
return updated
@router.delete("/{todo_id}")
def delete_todo(todo_id: UUID, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""Delete a todo."""
todo = TodoRepository.get_by_id(db, todo_id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not authorized")
TodoRepository.delete(db, todo_id)
return {"deleted": True}
Step 4: Generate Integration Tests
Language: Python 3.11
Context: Complete todo API (models, repositories, routes)
Generate: Pytest integration tests
Requirements:
1. Setup: Create test database, populate test data
2. Test create_todo: User can create a todo
3. Test list_todos: User sees only their own todos
4. Test update_todo: Can mark todo as complete
5. Test delete_todo: Deleted todos are gone
6. Test authorization: User can't access another user's todos
7. Use pytest fixtures for test data and database
Output: test_todos.py
This layered approach is scalable and manageable: each prompt is focused, each layer can be tested independently, and you can iterate on one layer without affecting others.
State Management Across Prompts
When generating multiple related pieces, you need to maintain state (what was generated, what depends on what). Use a generation log:
## Code Generation Log for Todo API
### Generated Artifacts
1. models.py (Step 1)
- User: id, email, password_hash, created_at, updated_at, todos relationship
- Todo: id, title, description, completed, due_date, owner_id, created_at, updated_at
2. repositories.py (Step 2)
- TodoRepository: create, get_by_id, list_for_user, update, delete
- UserRepository: create, get_by_email, get_by_id
- Uses models from models.py
3. routes.py (Step 3)
- POST /todos: create
- GET /todos: list for user
- GET /todos/{id}: get one
- PATCH /todos/{id}: update
- DELETE /todos/{id}: delete
- Uses repositories from repositories.py
### Dependencies
models.py → repositories.py → routes.py → tests.py
### Next Steps
- Generate database migrations
- Generate authentication module
- Generate client-side code (React hooks)
This log prevents confusion about what was generated and in what order.
Handling Generation Failures and Gaps
If a layer fails generation, don't skip it—iterate before moving to the next layer.
Step 2 failed: TodoRepository.delete() has logic error
Error: deletes todo without checking authorization first
Iteration prompt:
"The delete method should verify the todo exists and belongs to the specified user.
Return False if todo not found OR user_id doesn't match.
Include both cases in the docstring examples."
Regenerate Step 2, then proceed to Step 3.
Advanced Pattern: Generating Interdependent Modules
For complex systems with circular or fuzzy dependencies, consider generating:
- Contracts first. Generate interface/protocol definitions that all modules conform to.
- Implementations. Generate each module implementation against the contracts.
- Integration. Generate glue code (factories, dependency injection) that wires everything together.
Contracts (interfaces)
↓
Interface 1: UserRepository ↓
Interface 2: TodoRepository → Implementations
Interface 3: AuthService ↓
↓
Dependency Injection / Factories
↓
Integration Tests
This approach reduces coupling and makes it easier to swap implementations.
Key Takeaways
- Generate complex systems layer by layer, not all at once.
- Each layer builds on the previous: models → repositories → services → endpoints.
- Use a generation log to track what was generated and dependencies.
- Iterate on failures before proceeding to the next layer.
- For interdependent modules, generate contracts first, then implementations, then integration.
- Test each layer after generation before moving to the next.
Frequently Asked Questions
How many layers should I use?
For small apps: 2–3 layers. For medium apps: 4–5 layers. For large systems: 6–8 layers. Don't over-engineer; use layers that match your codebase structure.
Can I parallelize layer generation?
Only if layers are independent. Model generation and repository generation depend on each other (bad parallel candidates). But unit tests for two independent modules can be generated in parallel.
What if a later layer needs to revise an earlier layer?
This is normal. If Step 4 (endpoints) reveals a missing field in Step 1 (models), go back and regenerate Step 1. Then regenerate steps 2–4 as needed. Keep a generation log to track these revisions.
Should I generate all tests at the end, or test each layer?
Test each layer as you generate it. Unit tests for models, integration tests for repositories, etc. This catches issues early.
How do I handle backward-incompatible changes between layers?
Avoid them by being thorough in earlier layers. If you must change an earlier layer, document it and regenerate all dependent layers. Example: "Changed User.email to User.emails (list). Regenerate repositories, services, and routes."