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_contenthero_meta_pointsprogram_pillarstrust_segmentsnova_tracksmarketing_pageslegal_documentsfooter_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_KEYto 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
-
Create route file (if needed)
- Add to
/src/routes/[feature].py - Create Blueprint:
feature_bp = Blueprint('feature', __name__)
- Add to
-
Add route handler
@feature_bp.route('/endpoint', methods=['POST'])def handler_function():try:# 1. Get and validate inputdata = 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'}), 401token = auth_header.replace('Bearer ', '')user_response = supabase_service.client.auth.get_user(token)user_id = user_response.user.id# 3. Perform database operationresult = supabase_service.some_operation(user_id, sanitized_data)# 4. Return responsereturn jsonify(result), 200except Exception as e:error_msg = config.sanitize_error(e, 'Operation failed')return jsonify({'error': error_msg}), 500 -
Register blueprint in
main.pyfrom src.routes.feature import feature_bpapp.register_blueprint(feature_bp, url_prefix='/api') -
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.dataexcept Exception as e:raise Exception(f"Failed to perform operation: {str(e)}") -
Create frontend service in
/src/services/[feature].jsexport 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();}; -
Test thoroughly
- Test with valid data
- Test with invalid data
- Test authentication
- Test error cases
Quick Reference Card
Python Supabase Query Patterns
| Operation | Pattern |
|---|---|
| Select | client.table('t').select('*').execute() |
| Insert | client.table('t').insert(data).execute() |
| Update | client.table('t').update(data).eq('id', id).execute() |
| Delete | client.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
| Code | Meaning | When to Use |
|---|---|---|
| 200 | OK | Successful GET, PUT, DELETE |
| 201 | Created | Successful POST (resource created) |
| 400 | Bad Request | Invalid input data |
| 401 | Unauthorized | Missing or invalid auth token |
| 403 | Forbidden | Valid auth but insufficient permissions |
| 404 | Not Found | Resource doesn't exist |
| 409 | Conflict | Duplicate entry (e.g., already on waitlist) |
| 500 | Internal Server Error | Server-side error |
| 503 | Service Unavailable | Database or external service down |
Additional Resources
- Flask Documentation
- Supabase Python Client Docs
- Supabase JavaScript Client Docs (for frontend)
- SUPABASE_API_ARCHITECTURE.md - Architecture patterns
- Database setup scripts - Local migration helper
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:
- This is a Python Flask backend, not Node.js/Express
- Python Supabase client syntax is different from JavaScript - no
.select()after.insert()or.update() - All database operations go through the backend API, not direct Supabase client calls from frontend
- Authentication uses Supabase JWT tokens - verify tokens in backend before database operations
- Use
SupabaseServiceclass methods instead of direct client calls when possible - Always validate user input using validation utilities
- Always handle errors with try-except and
config.sanitize_error() - 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.