RapidKit

Posted on Feb 16

Build Production SaaS: Code Walkthrough

#webdev #python #microservices #tutorial

Most tutorials show you how to run commands.

This one shows you how the code works.

We'll build and dissect a 4-service SaaS backend with real implementation patterns used in production.

📦 Get the code: saas-starter-workspace

What We're Building

4 microservices with clear separation:

saas-api (FastAPI) — Product API
saas-admin (FastAPI) — Admin ops
saas-nest (NestJS) — Polyglot service
saas-webhooks (FastAPI) — Billing events + replay

Key features:

Dual auth (JWT + session cookies)
Webhook signature verification
Event replay for billing corrections
DDD structure with module injection

Prerequisites

python3 --version  # 3.10+
node --version     # 20+
pip install poetry
npm install -g rapidkit
git clone https://github.com/getrapidkit/rapidkit-examples.git
cd rapidkit-examples/saas-starter-workspace

💡 Faster Setup: RapidKit VS Code Extension

Skip manual cloning. Install the RapidKit VS Code Extension:

What you get:

Clone workspaces from GitHub with one click (Welcome Page → Example Workspaces)
Download & import workspaces directly
Run/test projects from the sidebar
Install modules via GUI
Integrated terminal with RapidKit commands

Install:

code --install-extension rapidkit.rapidkit-vscode

Open Command Palette (Cmd/Ctrl+Shift+P) → RapidKit: Import Workspace → done.

Architecture Pattern: DDD with Injection Markers

Project structure:

saas-api/src/
├── main.py              # Entry point with injection markers
├── app/main.py          # FastAPI factory
├── routing/
│   └── saas.py          # Business logic (485 lines)
└── modules/free/        # RapidKit modules
    ├── auth/core/       # Password hashing + JWT
    ├── auth/session/    # Cookie sessions
    ├── users/users_core/
    └── security/rate_limiting/

Entry point pattern (main.py):

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # <<<inject:startup>>> — modules add startup hooks here
    try:
        yield
    finally:
        # <<<inject:shutdown>>> — cleanup hooks
        pass

app: FastAPI = create_app(
    title="saas-api",
    lifespan=lifespan,
)
# <<<inject:routes>>> — modules mount routes here

Why this works:

Modules inject code at generation time
Clean separation of concerns
Easy to add/remove features

Code Deep Dive 1: Authentication Flow

Register endpoint (routing/saas.py):

@router.post("/auth/register", status_code=201)
async def register(
    payload: RegisterRequest,
    response: Response,
    users_service: UsersService = Depends(get_users_service),
    auth_runtime: AuthCoreRuntime = Depends(get_auth_core_runtime),
    _: Any = Depends(rate_limit_dependency()),
) -> dict[str, Any]:
    # 1. Hash password (PBKDF2, 100K iterations)
    password_hash = auth_runtime.hash_password(payload.password)

    # 2. Create user in database
    try:
        user_dto = UserCreateDTO(
            email=payload.email,
            password_hash=password_hash,
            full_name=payload.full_name,
        )
        created_user = await users_service.create_user(user_dto)
    except UserEmailConflictError:
        raise HTTPException(409, "Email already registered")

    # 3. Issue JWT token
    token_payload = {"sub": created_user.id, "email": created_user.email}
    access_token = auth_runtime.issue_token(token_payload)

    # 4. Create session + set cookie
    session_runtime = _get_session_runtime()
    session_token = session_runtime.create_session(user_id=created_user.id)
    _set_session_cookie(response, session_runtime, session_token)

    return {
        "user": created_user.model_dump(),
        "access_token": access_token,
        "token_type": "bearer",
    }

Pattern breakdown:

PBKDF2 hashing: 100K iterations prevents brute force
Service layer: Business logic isolated from HTTP
Dual auth: JWT (stateless) + session (server-side validation)
Rate limiting: Dependency injection pattern

Code Deep Dive 2: Authentication Resolution

How we resolve users from JWT OR cookies:

async def _get_current_user(
    request: Request,
    users_service: UsersService,
    auth_runtime: AuthCoreRuntime,
    session_runtime: SessionRuntime,
) -> UserDTO:
    user_id = None
    auth_header = request.headers.get("authorization")

    # Strategy 1: JWT Bearer Token
    if auth_header and auth_header.lower().startswith("bearer "):
        bearer_token = auth_header.split(" ", 1)[1]
        payload = auth_runtime.verify_token(bearer_token)
        user_id = payload.get("sub")

    # Strategy 2: Session Cookie
    else:
        session_token = request.cookies.get(session_runtime.settings.cookie.name)
        if not session_token:
            raise HTTPException(401, "Authentication required")

        session = session_runtime.verify_session_token(session_token)
        user_id = session.user_id

    # Fetch user from database
    return UserDTO.from_entity(await users_service.get_user(user_id))

Why dual auth?

Mobile apps → JWT (stateless, offline validation)
Web browsers → Cookies (server-side revocation)
Single function handles both

Usage in protected routes:

@router.get("/auth/me")
async def get_profile(
    request: Request,
    users: UsersService = Depends(get_users_service),
    auth: AuthCoreRuntime = Depends(get_auth_core_runtime),
    session: SessionRuntime = Depends(_get_session_runtime),
):
    user = await _get_current_user(request, users, auth, session)
    return {"user": user.model_dump()}

Code Deep Dive 3: Webhook Processing

Signature verification (saas-webhooks/routing/webhooks.py):

def _verify_signature(body: bytes, header: str, secret: str) -> bool:
    """Verify Stripe webhook signature using HMAC-SHA256."""

    timestamp, signature = _parse_signature_header(header)
    if not timestamp or not signature:
        return False

    # Reconstruct: "{timestamp}.{body}"
    signed_payload = f"{timestamp}.{body.decode()}"
    expected = hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()

    # Constant-time comparison (prevents timing attacks)
    return hmac.compare_digest(expected, signature)

Ingestion endpoint:

@router.post("/stripe", status_code=202)
async def receive_webhook(
    payload: StripeWebhookRequest,
    request: Request,
    background_tasks: BackgroundTasks,
):
    # 1. Verify signature
    raw_body = await request.body()
    signature = request.headers.get("stripe-signature")
    if signature and not _verify_signature(raw_body, signature, WEBHOOK_SECRET):
        raise HTTPException(400, "Invalid signature")

    # 2. Check idempotency
    if payload.id in _EVENTS:
        return {"status": "duplicate", "event_id": payload.id}

    # 3. Persist log
    _EVENTS[payload.id] = WebhookLogEntry(
        event_id=payload.id,
        event_type=payload.type,
        status="queued",
        received_at=_utc_now(),
    )

    # 4. Queue background processing
    background_tasks.add_task(_process_event, request, payload)
    return {"status": "accepted", "event_id": payload.id}

Why this pattern?

Signature first → reject fakes immediately
Idempotency → Stripe sends duplicates
Background tasks → don't block response
Logs → audit trail for billing events

Code Deep Dive 4: Webhook Replay

Critical for billing correctness:

@router.post("/replay/{event_id}")
async def replay_event(
    event_id: str,
    request: Request,
    background_tasks: BackgroundTasks,
):
    """Reprocess failed event without calling Stripe."""

    event = _EVENTS.get(event_id)
    if not event:
        raise HTTPException(404, "Event not found")

    event.replay_count += 1
    event.status = "replay_queued"

    # Reconstruct payload from stored data
    replay_payload = StripeWebhookRequest(
        id=event.event_id,
        type=event.event_type,
        data=event.metadata,
    )

    # Queue for processing
    background_tasks.add_task(_process_event, request, replay_payload)
    return {"status": "replay_accepted", "replay_count": event.replay_count}

When you need replay:

Handler had a bug → fix code, replay events
Database was down → replay after recovery
Need to backfill subscription states

Production upgrade:

Replace _EVENTS dict with PostgreSQL
Add exponential backoff for retries
Implement dead-letter queue

Code Deep Dive 5: NestJS Implementation

Auth service (saas-nest/src/auth/auth.service.ts):

@Injectable()
export class AuthService {
  private usersByEmail = new Map<string, UserRecord>();
  private tokens = new Map<string, string>();

  register(payload: { email: string; password: string }) {
    const email = payload.email.trim().toLowerCase();

    if (this.usersByEmail.has(email)) {
      throw new Error('Email already registered');
    }

    const user: UserRecord = {
      id: `user_${randomBytes(8).toString('hex')}`,
      email,
      passwordHash: this.hash(payload.password),
    };

    this.usersByEmail.set(email, user);
    const token = this.issueToken(user.id);

    return {
      user: this.publicUser(user),
      access_token: token,
      token_type: 'bearer'
    };
  }

  private hash(raw: string): string {
    return createHash('sha256').update(raw).digest('hex');
  }

  private issueToken(userId: string): string {
    const token = randomBytes(24).toString('hex');
    this.tokens.set(token, userId);
    return token;
  }
}

Controller:

@Controller('auth')
export class AuthController {
  constructor(private readonly authService: AuthService) {}

  @Post('register')
  register(@Body() payload: { email: string; password: string }) {
    try {
      return this.authService.register(payload);
    } catch (error) {
      throw new HttpException(error.message, HttpStatus.CONFLICT);
    }
  }

  @Get('me')
  me(@Headers('authorization') authorization?: string) {
    const token = extractBearer(authorization);
    if (!token) {
      throw new HttpException('Auth required', HttpStatus.UNAUTHORIZED);
    }
    return this.authService.me(token);
  }
}

Comparison with FastAPI version:

Same flow: hash → store → issue token
NestJS: TypeScript types + dependency injection
FastAPI: Async/await + Pydantic validation
Both: Service pattern isolates business logic

Run the Complete System

Terminal 1 — Product API:

cd saas-api && poetry shell && rapidkit init && rapidkit dev
# → http://127.0.0.1:8000

Terminal 2 — Admin API:

cd saas-admin && poetry shell && rapidkit init && rapidkit dev --port 8001
# → http://127.0.0.1:8001

Terminal 3 — NestJS:

cd saas-nest && npm install && rapidkit init && rapidkit dev --port 8002
# → http://127.0.0.1:8002

Terminal 4 — Webhooks:

cd saas-webhooks && poetry shell && rapidkit init && rapidkit dev --port 8003
# → http://127.0.0.1:8003

Test the Implementation

Auth flow:

import requests

# Register
r = requests.post('http://127.0.0.1:8000/api/auth/register', json={
    'email': 'dev@example.com',
    'password': 'SecurePass123!',
})
token = r.json()['access_token']

# Get profile
headers = {'Authorization': f'Bearer {token}'}
me = requests.get('http://127.0.0.1:8000/api/auth/me', headers=headers)
print(me.json())

Webhook flow:

# Send event
curl -X POST http://127.0.0.1:8003/api/webhooks/stripe \
  -H 'Content-Type: application/json' \
  -d '{"id":"evt_test","type":"customer.subscription.updated","data":{}}'

# View logs
curl http://127.0.0.1:8003/api/webhooks/logs | jq

# Replay
curl -X POST http://127.0.0.1:8003/api/webhooks/replay/evt_test

Key Patterns You Learned

1. Injection Marker System

# <<<inject:startup>>> — modules contribute code here
# <<<inject:routes>>> — dynamic route mounting

2. Service Layer Pattern

# Controller → calls → Service → calls → Repository
# Keeps business logic testable and framework-independent

3. Dependency Injection

async def endpoint(
    service: Service = Depends(get_service),
    auth: Auth = Depends(get_auth),
):
    # FastAPI resolves dependencies automatically

4. Background Processing

background_tasks.add_task(process_event, payload)
return {"status": "accepted"}  # Immediate response

5. Dual Authentication

# Check Bearer token first, fall back to session cookie
# Enables mobile (JWT) + web (session) clients

Production Hardening

Before going live:

1. Replace in-memory storage:

# Current: _EVENTS = {}
# Production: PostgreSQL with SQLAlchemy

2. Add retry logic:

for attempt in range(max_retries):
    try:
        await process()
        break
    except Exception:
        await asyncio.sleep(2 ** attempt)

3. Implement dead-letter queue:

if attempts >= max_retries:
    await send_to_dlq(event)

4. Add observability:

Structured logging with request IDs
APM (Sentry/DataDog)
Prometheus metrics

What You Built

✅ Multi-service architecture (4 services, independent scaling)

✅ Dual authentication (JWT + session cookies)

✅ Secure webhooks (signature verification + replay)

✅ DDD structure (clean separation of concerns)

✅ Polyglot support (FastAPI + NestJS coexist)

✅ Production patterns (background tasks, rate limiting, error handling)

This is the foundation real SaaS companies use to process millions in ARR.

Next Steps

Want Part 2 - Production Deployment?

Docker Compose orchestration
PostgreSQL + Redis setup
CI/CD with GitHub Actions
Kubernetes manifests
Secrets management

Drop a comment if interested 👇

Resources:

Found this useful? Follow for more production engineering patterns.

DEV Community