Skip to main content

Backend Developer Guide

Overview

This document provides comprehensive information about the backend architecture, API endpoints, and common patterns to help developers (and LLMs) work effectively with this Python Flask backend.


Tech Stack

Backend

  • Language: Python 3.13+
  • Framework: Flask (Python web framework)
  • Database: Supabase (PostgreSQL)
  • Database Client: supabase-py (Python client library)
  • CORS: Flask-CORS
  • Authentication: Supabase Auth (JWT tokens)

Frontend

  • Language: JavaScript/JSX
  • Framework: React with Vite
  • Router: React Router v6
  • Database Client: @supabase/supabase-js (JavaScript client)
  • Authentication: Supabase Auth (sessions managed in AuthContext)

Project Structure

.
├── main.py # Application entry point
├── src/
│ ├── routes/ # API route handlers
│ │ ├── enrollments.py # Enrollment endpoints
│ │ ├── waitlist.py # Waitlist endpoints
│ │ ├── content.py # Content management endpoints
│ │ ├── user.py # User management endpoints
│ │ └── email.py # Email sending endpoints
│ ├── services/
│ │ └── supabase_service.py # Supabase database service
│ ├── config/
│ │ ├── environment.py # Environment configuration
│ │ └── database.py # Database initialization
│ ├── utils/
│ │ └── validation.py # Input validation utilities
│ └── models/ # SQLAlchemy models (if needed)
├── pages/ # React page components
├── components/ # React UI components
└── supabase/
└── migrations/ # Database migration SQL files

Python Supabase Client - IMPORTANT PATTERNS

⚠️ CRITICAL: Python vs JavaScript Client Differences

The Python supabase-py client has different syntax than the JavaScript @supabase/supabase-js client. Many AI models trained on JavaScript examples will generate incorrect Python code.

❌ WRONG - JavaScript Pattern (Doesn't Work in Python)

# DON'T DO THIS - This will cause AttributeError
response = client.table('table_name').insert(data).select('*').execute()
response = client.table('table_name').update(data).eq('id', id).select('*').execute()

Error: 'SyncQueryRequestBuilder' object has no attribute 'select'

✅ CORRECT - Python Pattern

# DO THIS - No .select() chaining after .insert() or .update()
response = client.table('table_name').insert(data).execute()
response = client.table('table_name').update(data).eq('id', id).execute()

# The response already contains all columns by default
data = response.data[0] if response.data else None

Common Query Patterns

SELECT Queries

# Select all columns
response = client.table('table_name').select('*').execute()

# Select specific columns
response = client.table('table_name').select('id, name, email').execute()

# With filters
response = client.table('table_name').select('*').eq('user_id', user_id).execute()

# With multiple filters
response = (client.table('table_name')
.select('*')
.eq('status', 'active')
.gt('created_at', '2024-01-01')
.execute())

# With joins/relationships
response = (client.table('course_enrollments')
.select('id, status, track:track_id(id, title, description)')
.eq('user_id', user_id)
.execute())

# Ordering
response = (client.table('table_name')
.select('*')
.order('created_at', desc=True)
.execute())

INSERT Operations

# Single insert
data = {'name': 'John', 'email': 'john@example.com'}
response = client.table('users').insert(data).execute()
result = response.data[0] if response.data else None

# Multiple inserts
data = [
{'name': 'John', 'email': 'john@example.com'},
{'name': 'Jane', 'email': 'jane@example.com'}
]
response = client.table('users').insert(data).execute()
results = response.data

UPDATE Operations

# Update by ID
data = {'name': 'Updated Name'}
response = client.table('users').update(data).eq('id', user_id).execute()
result = response.data[0] if response.data else None

# Update multiple records
data = {'status': 'archived'}
response = (client.table('enrollments')
.update(data)
.lt('created_at', '2023-01-01')
.execute())

DELETE Operations

# Delete by ID
response = client.table('users').delete().eq('id', user_id).execute()

# Delete with conditions
response = (client.table('sessions')
.delete()
.lt('expires_at', datetime.now().isoformat())
.execute())

EXISTS Checks

# Check if record exists
response = (client.table('enrollments')
.select('id')
.eq('user_id', user_id)
.eq('track_id', track_id)
.execute())

exists = len(response.data) > 0

Filter Operators

.eq('column', value) # Equal to
.neq('column', value) # Not equal to
.gt('column', value) # Greater than
.gte('column', value) # Greater than or equal
.lt('column', value) # Less than
.lte('column', value) # Less than or equal
.like('column', '%pattern%') # Pattern matching
.ilike('column', '%pattern%') # Case-insensitive pattern matching
.is_('column', None) # IS NULL
.in_('column', [val1, val2]) # IN array
.contains('column', value) # Array/JSON contains

API Endpoints Reference

Authentication Pattern

All endpoints (except public content) require authentication via Bearer token:

# Backend token verification pattern
auth_header = request.headers.get('Authorization')
if not auth_header or not auth_header.startswith('Bearer '):
return jsonify({'error': 'Authorization required'}), 401

token = auth_header.replace('Bearer ', '')

try:
user_response = supabase_service.client.auth.get_user(token)
if not user_response or not user_response.user:
return jsonify({'error': 'Invalid or expired token'}), 401

user_id = user_response.user.id
except Exception as e:
return jsonify({'error': 'Authentication failed'}), 401

Enrollments API (/api/enrollments)

GET /api/enrollments/dashboard

Description: Get learner dashboard (enrollments + resources) Auth: Required Response:

{
"enrollments": [
{
"id": 1,
"status": "active",
"track_id": "track-slug",
"created_at": "2024-01-01T00:00:00",
"track": {
"id": "track-slug",
"title": "Track Title",
"description": "..."
}
}
],
"resources": [
{
"id": 1,
"track_id": "track-slug",
"title": "Resource Title",
"url": "https://..."
}
]
}

GET /api/enrollments/track/<track_id>

Description: Get enrollment for specific track Auth: Required Response:

{
"enrollment": {
"id": 1,
"status": "active",
"track_id": "track-slug",
"created_at": "2024-01-01T00:00:00"
}
}

POST /api/enrollments

Description: Enroll user in a track Auth: Required Body:

{
"trackId": "track-slug",
"pricingTier": "free" // or "paid"
}

Response:

{
"enrollment": {
"id": 1,
"status": "active",
"track_id": "track-slug",
"created_at": "2024-01-01T00:00:00"
},
"alreadyEnrolled": false
}

Waitlist API (/api/waitlist)

GET /api/waitlist/check

Description: Check if user is on waitlist Auth: Not required Query Params: trackId, email Response:

{
"on_waitlist": true,
"entry": {
"id": 1,
"created_at": "2024-01-01T00:00:00",
"email": "user@example.com"
}
}

POST /api/waitlist

Description: Add user to waitlist Auth: Not required Body:

{
"trackId": "track-slug",
"email": "user@example.com",
"name": "User Name",
"notes": "Optional notes"
}

DELETE /api/waitlist

Description: Remove user from waitlist Auth: Not required Body:

{
"trackId": "track-slug",
"email": "user@example.com"
}

Content API (/api/content)

GET /api/content?section=<section_name>

Description: Get content by section Auth: Not required for public content Sections:

  • homepage_content
  • hero_meta_points
  • program_pillars
  • trust_segments
  • nova_tracks
  • marketing_pages
  • legal_documents
  • footer_navigation_sections

POST /api/content

Description: Create new content Auth: Required (admin only) Body:

{
"section": "section_name",
"title": "Content Title",
"content": "Content body"
}

PUT /api/content/<content_id>

Description: Update content Auth: Required (admin only)

DELETE /api/content/<content_id>

Description: Delete content Auth: Required (admin only)


SupabaseService Class

The SupabaseService class in /src/services/supabase_service.py provides a centralized interface for all database operations.

Key Methods

User Operations

get_all_users() -> List[Dict[str, Any]]
get_user_by_email(email: str) -> Optional[Dict[str, Any]]
get_user_by_id(user_id: int) -> Optional[Dict[str, Any]]
create_user(email: str, name: str) -> Dict[str, Any]
update_user(user_id: int, data: Dict[str, Any]) -> Dict[str, Any]
delete_user(user_id: int) -> bool

Enrollment Operations

get_user_enrollments(user_id: str, include_track_details: bool = True) -> List[Dict[str, Any]]
get_enrollment_by_track(user_id: str, track_id: str) -> Optional[Dict[str, Any]]
create_enrollment(user_id: str, track_id: str, status: str = 'active', source: str = 'web-free') -> Dict[str, Any]
update_enrollment(enrollment_id: int, data: Dict[str, Any]) -> Dict[str, Any]
get_course_resources(track_ids: List[str]) -> List[Dict[str, Any]]

Waitlist Operations

get_waitlist_entries(track_id: Optional[int] = None) -> List[Dict[str, Any]]
create_waitlist_entry(track_id: int, email: str, name: Optional[str] = None, notes: Optional[str] = None, source: str = 'web') -> Dict[str, Any]
delete_waitlist_entry(entry_id: int) -> bool
check_waitlist_status(track_id: str, email: str) -> Optional[Dict[str, Any]]

Content Operations

get_all_content() -> List[Dict[str, Any]]
get_content_by_section(section: str) -> List[Dict[str, Any]]
get_content_by_id(content_id: int) -> Optional[Dict[str, Any]]
create_content(section: str, title: str, content: str) -> Dict[str, Any]
update_content(content_id: int, data: Dict[str, Any]) -> Dict[str, Any]
delete_content(content_id: int) -> bool

Track Operations

get_nova_tracks() -> List[Dict[str, Any]]
get_nova_track_by_id(track_id: int) -> Optional[Dict[str, Any]]

Authentication Helper

validate_auth_token(token: str) -> Optional[Dict[str, Any]]

Common Patterns and Best Practices

1. Error Handling

Always Use Try-Except Blocks

from src.config.environment import config

@blueprint.route('/endpoint', methods=['POST'])
def endpoint():
try:
# Your logic here
result = supabase_service.some_operation()
return jsonify(result), 200
except Exception as e:
error_msg = config.sanitize_error(e, 'Failed to perform operation')
return jsonify({'error': error_msg}), 500

Environment-Aware Error Messages

# config.sanitize_error() returns:
# - Full error in development (DEBUG=True)
# - Generic message in production (DEBUG=False)
error_msg = config.sanitize_error(e, 'User-friendly fallback message')

2. Input Validation

from src.utils.validation import validate_enrollment_data, validate_email

# Validate complex data
is_valid, error_msg, sanitized_data = validate_enrollment_data(data)
if not is_valid:
return jsonify({'error': error_msg}), 400

# Use sanitized data (prevents injection attacks)
track_id = sanitized_data['trackId']

3. JSON Request Handling

# Safely get JSON data
data = request.get_json(force=True, silent=True) or {}

# This handles:
# - Missing Content-Type header
# - Empty request body
# - Invalid JSON

4. Service Initialization

# Initialize service at module level for reuse
supabase_service = SupabaseService()

# Don't create new instance in every function
# ❌ BAD: supabase_service = SupabaseService() # In function
# ✅ GOOD: Use module-level instance

5. Response Formatting

# Consistent response format
return jsonify({
'success': True,
'data': result,
'message': 'Operation successful'
}), 200

# Error response format
return jsonify({
'error': 'Error message',
'details': additional_info # Optional, only in debug mode
}), 400

Database Schema Key Tables

course_enrollments

- id (integer, primary key)
- user_id (uuid, foreign key to auth.users)
- track_id (text, foreign key to nova_tracks)
- status (text: 'active', 'pending-payment', 'completed', 'cancelled')
- source (text: 'web-free', 'web-paid', 'admin')
- created_at (timestamptz)
- updated_at (timestamptz)

track_waitlist

- id (integer, primary key)
- track_id (text, foreign key to nova_tracks)
- email (text)
- name (text, optional)
- notes (text, optional)
- source (text: 'web', 'admin')
- created_at (timestamptz)

nova_tracks

- id (text, primary key)
- slug (text, unique)
- code (text)
- title (text)
- description (text)
- duration (text)
- level_label (text)
- pricing_tier (text: 'free', 'paid')
- display_order (integer)
- is_waitlist_only (boolean)

course_resources

- id (integer, primary key)
- track_id (text, foreign key to nova_tracks)
- title (text)
- description (text)
- resource_type (text: 'video', 'pdf', 'link', 'code')
- url (text)
- position (integer)

Common Pitfalls and Solutions

1. The .select() Chaining Issue ⚠️

Problem: Copying JavaScript Supabase patterns

# ❌ This FAILS in Python
response = client.table('table').insert(data).select('*').execute()

Solution: Remove .select() after .insert() or .update()

# ✅ This WORKS in Python
response = client.table('table').insert(data).execute()
# Response already contains the inserted data

2. UUID vs Integer IDs

Problem: Mixing UUID (auth.users) with integer IDs (other tables)

# user_id is UUID from Supabase Auth
user_id = user_response.user.id # UUID string

# But some tables use integer IDs
enrollment_id = 123 # Integer

Solution: Be aware of field types in your queries

# UUID comparison (user_id)
.eq('user_id', user_id) # user_id is string UUID

# Integer comparison (id)
.eq('id', enrollment_id) # enrollment_id is int

3. Datetime Handling

Problem: Inconsistent datetime formats

# ❌ Bad
created_at = datetime.now() # Naive datetime

Solution: Always use ISO format strings

# ✅ Good
from datetime import datetime
created_at = datetime.utcnow().isoformat()

4. Empty Response Handling

Problem: Assuming data exists

# ❌ Bad
response = client.table('users').select('*').eq('id', user_id).execute()
return response.data[0] # May crash if no results

Solution: Check before accessing

# ✅ Good
response = client.table('users').select('*').eq('id', user_id).execute()
return response.data[0] if response.data else None

5. RLS Policies

Problem: Queries fail silently due to Row Level Security (RLS)

# Query returns empty even though data exists
response = client.table('enrollments').select('*').execute()
# Empty because RLS blocks it

Solution: Use Service Role Key for backend operations

# In SupabaseService.__init__
# Service role key bypasses RLS (use carefully!)
self.client = create_client(self.url, self.service_key)

Environment Variables

Required Backend Variables

# Supabase
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_SERVICE_ROLE_KEY=your-service-role-key # Backend only!
VITE_SUPABASE_ANON_KEY=your-anon-key # Fallback

# Flask
SECRET_KEY=your-secret-key
NODE_ENV=development # or 'production'

# Database
DATABASE_URL=postgresql://... # Optional, uses Supabase by default

# Email (Optional)
SENDGRID_API_KEY=your-sendgrid-key
EMAIL_ENABLED=true

Security Notes

  • Never expose SUPABASE_SERVICE_ROLE_KEY to frontend
  • Service role key bypasses all RLS policies
  • Use anon key for frontend Supabase Auth only
  • Backend validates auth tokens server-side

Testing

Manual API Testing with cURL

# Get auth token (from browser dev tools or login response)
TOKEN="your-jwt-token-here"

# Test enrollment dashboard
curl -H "Authorization: Bearer $TOKEN" \
http://localhost:5001/api/enrollments/dashboard

# Test creating enrollment
curl -X POST \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"trackId":"ai-foundations","pricingTier":"free"}' \
http://localhost:5001/api/enrollments

# Test waitlist check (no auth)
curl "http://localhost:5001/api/waitlist/check?trackId=ai-foundations&email=test@example.com"

# Test waitlist join (no auth)
curl -X POST \
-H "Content-Type: application/json" \
-d '{"trackId":"ai-foundations","email":"test@example.com","name":"Test User"}' \
http://localhost:5001/api/waitlist

Running the Backend

# Install dependencies
pip install -r requirements.txt

# Run Flask development server
python main.py

# Server runs on http://localhost:5001

When Adding New Endpoints

Checklist for New API Endpoint

  1. Create route file (if needed)

    • Add to /src/routes/[feature].py
    • Create Blueprint: feature_bp = Blueprint('feature', __name__)
  2. Add route handler

    @feature_bp.route('/endpoint', methods=['POST'])
    def handler_function():
    try:
    # 1. Get and validate input
    data = request.get_json(force=True, silent=True) or {}
    is_valid, error_msg, sanitized_data = validate_data(data)
    if not is_valid:
    return jsonify({'error': error_msg}), 400

    # 2. Authenticate (if required)
    auth_header = request.headers.get('Authorization')
    if not auth_header:
    return jsonify({'error': 'Authorization required'}), 401
    token = auth_header.replace('Bearer ', '')
    user_response = supabase_service.client.auth.get_user(token)
    user_id = user_response.user.id

    # 3. Perform database operation
    result = supabase_service.some_operation(user_id, sanitized_data)

    # 4. Return response
    return jsonify(result), 200
    except Exception as e:
    error_msg = config.sanitize_error(e, 'Operation failed')
    return jsonify({'error': error_msg}), 500
  3. Register blueprint in main.py

    from src.routes.feature import feature_bp
    app.register_blueprint(feature_bp, url_prefix='/api')
  4. Add service method to SupabaseService (if needed)

    def some_operation(self, param: str) -> Dict[str, Any]:
    """Description of operation"""
    try:
    response = self.client.table('table_name').select('*').eq('field', param).execute()
    return response.data
    except Exception as e:
    raise Exception(f"Failed to perform operation: {str(e)}")
  5. Create frontend service in /src/services/[feature].js

    export const someOperation = async (param) => {
    const token = await getAuthToken();
    const response = await fetch('/api/endpoint', {
    method: 'POST',
    headers: {
    'Authorization': `Bearer ${token}`,
    'Content-Type': 'application/json',
    },
    body: JSON.stringify({ param }),
    });
    if (!response.ok) throw new Error('Operation failed');
    return response.json();
    };
  6. Test thoroughly

    • Test with valid data
    • Test with invalid data
    • Test authentication
    • Test error cases

Quick Reference Card

Python Supabase Query Patterns

OperationPattern
Selectclient.table('t').select('*').execute()
Insertclient.table('t').insert(data).execute()
Updateclient.table('t').update(data).eq('id', id).execute()
Deleteclient.table('t').delete().eq('id', id).execute()
Filter.eq('col', val), .neq(), .gt(), .lt(), .in_()
Order.order('col', desc=True)
Limit.limit(10)
Join.select('*, relation:fk(col1, col2)')

HTTP Status Codes

CodeMeaningWhen to Use
200OKSuccessful GET, PUT, DELETE
201CreatedSuccessful POST (resource created)
400Bad RequestInvalid input data
401UnauthorizedMissing or invalid auth token
403ForbiddenValid auth but insufficient permissions
404Not FoundResource doesn't exist
409ConflictDuplicate entry (e.g., already on waitlist)
500Internal Server ErrorServer-side error
503Service UnavailableDatabase or external service down

Additional Resources


Last Updated

Date: October 31, 2025 Status: ✅ Active Development Python Version: 3.13+ Flask Version: 3.x Supabase Python Client: Latest


For LLMs Reading This

If you're an AI assistant working with this codebase:

  1. This is a Python Flask backend, not Node.js/Express
  2. Python Supabase client syntax is different from JavaScript - no .select() after .insert() or .update()
  3. All database operations go through the backend API, not direct Supabase client calls from frontend
  4. Authentication uses Supabase JWT tokens - verify tokens in backend before database operations
  5. Use SupabaseService class methods instead of direct client calls when possible
  6. Always validate user input using validation utilities
  7. Always handle errors with try-except and config.sanitize_error()
  8. Check this guide first before making assumptions about syntax or patterns

When in doubt, refer to existing code in /src/routes/ for patterns and examples.