$ cat /posts/supabase-with-python-backend-development-guide.md

Supabase with Python: Backend Development Guide

drwxr-xr-x2026-01-275 min0 views

Building Python backend applications with Supabase enables leveraging PostgreSQL database, authentication, real-time subscriptions, and storage through Python client library integrating with FastAPI, Flask, Django, and data science workflows creating powerful server-side applications combining Python's rich ecosystem with Supabase's managed infrastructure. Unlike JavaScript-focused tutorials emphasizing frontend frameworks, Python integration targets backend services including REST APIs serving mobile and web clients, data processing pipelines transforming and analyzing data, machine learning workflows training and serving models, automation scripts handling scheduled tasks, and microservices architectures building distributed systems. This comprehensive guide covers installing Supabase Python client and dependencies, connecting to Supabase from Python applications, performing database operations with async support, implementing authentication and user management, building REST APIs with FastAPI integration, handling real-time subscriptions, managing file uploads to storage, and deploying Python backends to production. Python integration demonstrates backend versatility supporting diverse use cases from web APIs to data pipelines. Before starting, review JavaScript client basics, database fundamentals, and authentication concepts.

Python Client Installation

pythonsetup.py

# Install Supabase Python client
pip install supabase

# Install additional dependencies
pip install python-dotenv  # Environment variables
pip install fastapi uvicorn  # FastAPI framework
pip install httpx  # Async HTTP client
pip install pydantic  # Data validation

# requirements.txt
supabase==2.3.4
python-dotenv==1.0.0
fastapi==0.109.0
uvicorn[standard]==0.27.0
httpx==0.26.0
pydantic==2.5.3
pydantic-settings==2.1.0

# Install from requirements
pip install -r requirements.txt

# .env - Environment configuration
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_KEY=your-anon-key
SUPABASE_SERVICE_KEY=your-service-role-key

# config.py - Configuration management
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    supabase_url: str
    supabase_key: str
    supabase_service_key: str
    environment: str = "development"
    debug: bool = True
    
    class Config:
        env_file = ".env"
        case_sensitive = False

@lru_cache()
def get_settings() -> Settings:
    return Settings()

# Database client initialization
from supabase import create_client, Client
from config import get_settings

settings = get_settings()

# Create Supabase client
supabase: Client = create_client(
    settings.supabase_url,
    settings.supabase_key
)

# Service role client (admin operations)
supabase_admin: Client = create_client(
    settings.supabase_url,
    settings.supabase_service_key
)

print("✓ Supabase client initialized")

Database Operations with Python

pythondatabase.py

# database.py - Database operations
from supabase import Client
from typing import List, Dict, Any, Optional
from datetime import datetime

class DatabaseService:
    def __init__(self, client: Client):
        self.client = client
    
    # SELECT operations
    async def get_all_users(self) -> List[Dict[str, Any]]:
        """Get all users"""
        response = self.client.table('users').select('*').execute()
        return response.data
    
    async def get_user_by_id(self, user_id: str) -> Optional[Dict[str, Any]]:
        """Get single user by ID"""
        response = self.client.table('users') \
            .select('*') \
            .eq('id', user_id) \
            .single() \
            .execute()
        return response.data
    
    async def search_users(self, search_term: str) -> List[Dict[str, Any]]:
        """Search users by name or email"""
        response = self.client.table('users') \
            .select('*') \
            .or_(f'name.ilike.%{search_term}%,email.ilike.%{search_term}%') \
            .execute()
        return response.data
    
    # INSERT operations
    async def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
        """Create new user"""
        response = self.client.table('users') \
            .insert(user_data) \
            .execute()
        return response.data[0]
    
    async def create_multiple_users(self, users: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Bulk insert users"""
        response = self.client.table('users') \
            .insert(users) \
            .execute()
        return response.data
    
    # UPDATE operations
    async def update_user(self, user_id: str, updates: Dict[str, Any]) -> Dict[str, Any]:
        """Update user by ID"""
        response = self.client.table('users') \
            .update(updates) \
            .eq('id', user_id) \
            .execute()
        return response.data[0]
    
    # DELETE operations
    async def delete_user(self, user_id: str) -> None:
        """Delete user by ID"""
        self.client.table('users') \
            .delete() \
            .eq('id', user_id) \
            .execute()
    
    # Complex queries with joins
    async def get_user_with_posts(self, user_id: str) -> Dict[str, Any]:
        """Get user with their posts"""
        response = self.client.table('users') \
            .select('*, posts(*)') \
            .eq('id', user_id) \
            .single() \
            .execute()
        return response.data
    
    # Filtering and pagination
    async def get_active_users_paginated(
        self,
        page: int = 1,
        per_page: int = 20
    ) -> Dict[str, Any]:
        """Get active users with pagination"""
        start = (page - 1) * per_page
        end = start + per_page - 1
        
        response = self.client.table('users') \
            .select('*', count='exact') \
            .eq('status', 'active') \
            .order('created_at', desc=True) \
            .range(start, end) \
            .execute()
        
        return {
            'data': response.data,
            'count': response.count,
            'page': page,
            'per_page': per_page,
            'total_pages': (response.count + per_page - 1) // per_page
        }
    
    # Aggregation
    async def get_user_statistics(self) -> Dict[str, int]:
        """Get user statistics"""
        # Total users
        total_response = self.client.table('users') \
            .select('*', count='exact', head=True) \
            .execute()
        
        # Active users
        active_response = self.client.table('users') \
            .select('*', count='exact', head=True) \
            .eq('status', 'active') \
            .execute()
        
        return {
            'total_users': total_response.count,
            'active_users': active_response.count,
            'inactive_users': total_response.count - active_response.count
        }

# Usage example
from database import DatabaseService
from config import supabase

db = DatabaseService(supabase)

# Create user
new_user = await db.create_user({
    'name': 'John Doe',
    'email': '[email protected]',
    'status': 'active'
})

# Get user with posts
user_data = await db.get_user_with_posts(new_user['id'])

# Paginated query
result = await db.get_active_users_paginated(page=1, per_page=10)
print(f"Total users: {result['count']}")
print(f"Page {result['page']} of {result['total_pages']}")

FastAPI REST API Integration

pythonmain.py

# main.py - FastAPI application
from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, EmailStr
from typing import List, Optional
from datetime import datetime
import uuid

from database import DatabaseService
from config import get_settings, supabase

app = FastAPI(
    title="Supabase Python API",
    description="REST API built with FastAPI and Supabase",
    version="1.0.0"
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Pydantic models
class UserCreate(BaseModel):
    name: str
    email: EmailStr
    bio: Optional[str] = None

class UserUpdate(BaseModel):
    name: Optional[str] = None
    email: Optional[EmailStr] = None
    bio: Optional[str] = None
    status: Optional[str] = None

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    bio: Optional[str]
    status: str
    created_at: datetime
    updated_at: datetime

class PaginatedResponse(BaseModel):
    data: List[UserResponse]
    count: int
    page: int
    per_page: int
    total_pages: int

# Dependency injection
def get_db() -> DatabaseService:
    return DatabaseService(supabase)

# Routes
@app.get("/")
async def root():
    return {
        "message": "Supabase Python API",
        "version": "1.0.0",
        "docs": "/docs"
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    try:
        # Test database connection
        supabase.table('users').select('id').limit(1).execute()
        return {
            "status": "healthy",
            "database": "connected",
            "timestamp": datetime.now().isoformat()
        }
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Database connection failed"
        )

@app.get("/users", response_model=PaginatedResponse)
async def get_users(
    page: int = 1,
    per_page: int = 20,
    db: DatabaseService = Depends(get_db)
):
    """Get all users with pagination"""
    result = await db.get_active_users_paginated(page, per_page)
    return result

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: str,
    db: DatabaseService = Depends(get_db)
):
    """Get user by ID"""
    user = await db.get_user_by_id(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user

@app.post("/users", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user: UserCreate,
    db: DatabaseService = Depends(get_db)
):
    """Create new user"""
    user_data = {
        "id": str(uuid.uuid4()),
        **user.model_dump(),
        "status": "active",
        "created_at": datetime.now().isoformat(),
        "updated_at": datetime.now().isoformat()
    }
    
    try:
        new_user = await db.create_user(user_data)
        return new_user
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=str(e)
        )

@app.patch("/users/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: str,
    user: UserUpdate,
    db: DatabaseService = Depends(get_db)
):
    """Update user"""
    # Check if user exists
    existing_user = await db.get_user_by_id(user_id)
    if not existing_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    
    # Update user
    updates = {
        **user.model_dump(exclude_unset=True),
        "updated_at": datetime.now().isoformat()
    }
    
    updated_user = await db.update_user(user_id, updates)
    return updated_user

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: str,
    db: DatabaseService = Depends(get_db)
):
    """Delete user"""
    # Check if user exists
    existing_user = await db.get_user_by_id(user_id)
    if not existing_user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    
    await db.delete_user(user_id)
    return None

@app.get("/users/search/{search_term}")
async def search_users(
    search_term: str,
    db: DatabaseService = Depends(get_db)
):
    """Search users by name or email"""
    users = await db.search_users(search_term)
    return {"results": users, "count": len(users)}

@app.get("/stats/users")
async def get_user_stats(
    db: DatabaseService = Depends(get_db)
):
    """Get user statistics"""
    stats = await db.get_user_statistics()
    return stats

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8000,
        reload=True
    )

Authentication in Python

pythonauth.py

# auth.py - Authentication service
from supabase import Client
from typing import Dict, Any, Optional
from fastapi import HTTPException, status, Header
import jwt

class AuthService:
    def __init__(self, client: Client):
        self.client = client
    
    def sign_up(self, email: str, password: str, user_metadata: Dict = None) -> Dict[str, Any]:
        """Register new user"""
        try:
            response = self.client.auth.sign_up({
                "email": email,
                "password": password,
                "options": {
                    "data": user_metadata or {}
                }
            })
            return {
                "user": response.user,
                "session": response.session
            }
        except Exception as e:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(e)
            )
    
    def sign_in(self, email: str, password: str) -> Dict[str, Any]:
        """Sign in user"""
        try:
            response = self.client.auth.sign_in_with_password({
                "email": email,
                "password": password
            })
            return {
                "user": response.user,
                "session": response.session,
                "access_token": response.session.access_token
            }
        except Exception as e:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid credentials"
            )
    
    def sign_out(self) -> None:
        """Sign out user"""
        self.client.auth.sign_out()
    
    def get_user(self, access_token: str) -> Optional[Dict[str, Any]]:
        """Get user from access token"""
        try:
            response = self.client.auth.get_user(access_token)
            return response.user
        except Exception:
            return None
    
    def refresh_session(self, refresh_token: str) -> Dict[str, Any]:
        """Refresh access token"""
        try:
            response = self.client.auth.refresh_session(refresh_token)
            return {
                "session": response.session,
                "access_token": response.session.access_token
            }
        except Exception as e:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid refresh token"
            )
    
    def reset_password_email(self, email: str) -> None:
        """Send password reset email"""
        self.client.auth.reset_password_for_email(email)
    
    def update_user(self, access_token: str, updates: Dict[str, Any]) -> Dict[str, Any]:
        """Update user profile"""
        self.client.auth.set_session(access_token, "")
        response = self.client.auth.update_user(updates)
        return response.user

# Authentication dependency
from fastapi import Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
) -> Dict[str, Any]:
    """Get current authenticated user"""
    token = credentials.credentials
    
    auth = AuthService(supabase)
    user = auth.get_user(token)
    
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )
    
    return user

# Protected routes example
from pydantic import BaseModel

class SignUpRequest(BaseModel):
    email: str
    password: str
    name: str

class SignInRequest(BaseModel):
    email: str
    password: str

@app.post("/auth/signup")
async def signup(request: SignUpRequest):
    """Register new user"""
    auth = AuthService(supabase)
    result = auth.sign_up(
        email=request.email,
        password=request.password,
        user_metadata={"name": request.name}
    )
    return result

@app.post("/auth/signin")
async def signin(request: SignInRequest):
    """Sign in user"""
    auth = AuthService(supabase)
    result = auth.sign_in(
        email=request.email,
        password=request.password
    )
    return result

@app.post("/auth/signout")
async def signout():
    """Sign out user"""
    auth = AuthService(supabase)
    auth.sign_out()
    return {"message": "Signed out successfully"}

@app.get("/auth/me")
async def get_me(current_user: Dict = Depends(get_current_user)):
    """Get current user profile"""
    return current_user

@app.get("/protected")
async def protected_route(current_user: Dict = Depends(get_current_user)):
    """Protected route example"""
    return {
        "message": "This is a protected route",
        "user_id": current_user.id,
        "email": current_user.email
    }

File Storage Management

pythonstorage.py

# storage.py - Storage service
from supabase import Client
from typing import BinaryIO, List, Dict
from fastapi import UploadFile
import uuid
import os

class StorageService:
    def __init__(self, client: Client, bucket: str):
        self.client = client
        self.bucket = bucket
    
    def upload_file(
        self,
        file_path: str,
        file_data: BinaryIO,
        content_type: str = None
    ) -> str:
        """Upload file to storage"""
        options = {}
        if content_type:
            options['content_type'] = content_type
        
        self.client.storage.from_(self.bucket).upload(
            file_path,
            file_data,
            file_options=options
        )
        
        # Get public URL
        return self.get_public_url(file_path)
    
    def download_file(self, file_path: str) -> bytes:
        """Download file from storage"""
        response = self.client.storage.from_(self.bucket).download(file_path)
        return response
    
    def delete_file(self, file_path: str) -> None:
        """Delete file from storage"""
        self.client.storage.from_(self.bucket).remove([file_path])
    
    def list_files(self, folder: str = "") -> List[Dict]:
        """List files in folder"""
        response = self.client.storage.from_(self.bucket).list(folder)
        return response
    
    def get_public_url(self, file_path: str) -> str:
        """Get public URL for file"""
        response = self.client.storage.from_(self.bucket).get_public_url(file_path)
        return response
    
    def create_signed_url(self, file_path: str, expires_in: int = 3600) -> str:
        """Create signed URL with expiry"""
        response = self.client.storage.from_(self.bucket).create_signed_url(
            file_path,
            expires_in
        )
        return response['signedURL']

# FastAPI file upload endpoint
from fastapi import File, UploadFile
import mimetypes

@app.post("/upload")
async def upload_file(
    file: UploadFile = File(...),
    current_user: Dict = Depends(get_current_user)
):
    """Upload file to Supabase Storage"""
    
    # Generate unique filename
    file_ext = os.path.splitext(file.filename)[1]
    file_name = f"{uuid.uuid4()}{file_ext}"
    file_path = f"uploads/{current_user.id}/{file_name}"
    
    # Get content type
    content_type = file.content_type or mimetypes.guess_type(file.filename)[0]
    
    # Upload to Supabase
    storage = StorageService(supabase, "files")
    
    try:
        file_content = await file.read()
        public_url = storage.upload_file(
            file_path,
            file_content,
            content_type
        )
        
        return {
            "message": "File uploaded successfully",
            "file_name": file_name,
            "file_path": file_path,
            "public_url": public_url,
            "size": len(file_content)
        }
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Upload failed: {str(e)}"
        )

@app.get("/files")
async def list_files(
    current_user: Dict = Depends(get_current_user)
):
    """List user's uploaded files"""
    storage = StorageService(supabase, "files")
    folder = f"uploads/{current_user.id}"
    
    try:
        files = storage.list_files(folder)
        return {"files": files, "count": len(files)}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e)
        )

@app.delete("/files/{file_path:path}")
async def delete_file(
    file_path: str,
    current_user: Dict = Depends(get_current_user)
):
    """Delete uploaded file"""
    # Verify file belongs to user
    if not file_path.startswith(f"uploads/{current_user.id}/"):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Access denied"
        )
    
    storage = StorageService(supabase, "files")
    
    try:
        storage.delete_file(file_path)
        return {"message": "File deleted successfully"}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=str(e)
        )

Real-time Subscriptions

pythonrealtime.py

# realtime.py - Real-time subscriptions
from supabase import Client
import asyncio
from typing import Callable, Dict, Any

class RealtimeService:
    def __init__(self, client: Client):
        self.client = client
        self.subscriptions = {}
    
    def subscribe_to_table(
        self,
        table: str,
        callback: Callable,
        event: str = "*",  # INSERT, UPDATE, DELETE, or *
        filter_column: str = None,
        filter_value: Any = None
    ):
        """Subscribe to table changes"""
        channel = self.client.channel(f"public:{table}")
        
        # Build filter if provided
        filter_str = None
        if filter_column and filter_value:
            filter_str = f"{filter_column}=eq.{filter_value}"
        
        # Subscribe to changes
        channel.on_postgres_changes(
            event=event,
            schema="public",
            table=table,
            callback=callback,
            filter=filter_str
        )
        
        channel.subscribe()
        self.subscriptions[table] = channel
    
    def unsubscribe(self, table: str):
        """Unsubscribe from table"""
        if table in self.subscriptions:
            self.subscriptions[table].unsubscribe()
            del self.subscriptions[table]
    
    def unsubscribe_all(self):
        """Unsubscribe from all channels"""
        for channel in self.subscriptions.values():
            channel.unsubscribe()
        self.subscriptions.clear()

# Usage example
def handle_user_change(payload):
    """Handle user table changes"""
    event = payload['event']  # INSERT, UPDATE, DELETE
    record = payload['record']  # New/updated record
    old_record = payload.get('old_record')  # For UPDATE/DELETE
    
    print(f"Event: {event}")
    print(f"Record: {record}")
    
    if event == "INSERT":
        print(f"New user created: {record['email']}")
    elif event == "UPDATE":
        print(f"User updated: {record['email']}")
    elif event == "DELETE":
        print(f"User deleted: {old_record['email']}")

# Subscribe to users table
realtime = RealtimeService(supabase)
realtime.subscribe_to_table(
    table="users",
    callback=handle_user_change,
    event="*"  # All events
)

# Subscribe with filter
def handle_active_user(payload):
    print(f"Active user changed: {payload['record']}")

realtime.subscribe_to_table(
    table="users",
    callback=handle_active_user,
    event="UPDATE",
    filter_column="status",
    filter_value="active"
)

# Keep connection alive
try:
    print("Listening for changes... Press Ctrl+C to stop")
    asyncio.get_event_loop().run_forever()
except KeyboardInterrupt:
    print("\nUnsubscribing...")
    realtime.unsubscribe_all()
    print("Done")

Python Integration Best Practices

Use Environment Variables: Store credentials securely using .env files and pydantic-settings
Implement Error Handling: Use try-except blocks catching Supabase exceptions gracefully
Type Hints: Use Python type hints with Pydantic models ensuring type safety
Dependency Injection: Use FastAPI dependencies for database and auth services
Async Operations: Use async/await for non-blocking database operations
Pagination: Implement pagination for list endpoints reducing data transfer
Authentication: Protect routes with JWT authentication using Bearer tokens
Validation: Use Pydantic models validating request/response data
Connection Pooling: Reuse Supabase client instances avoiding connection overhead
Logging: Implement structured logging tracking API requests and errors

Pro Tip: Use FastAPI's automatic OpenAPI documentation at /docs for API exploration. Implement proper error handling returning meaningful HTTP status codes. Use Pydantic models ensuring type safety throughout application. Leverage async operations for better performance. Review security practices and performance optimization.

Common Issues

Connection Errors: Verify SUPABASE_URL and SUPABASE_KEY in environment variables, check network connectivity
Authentication Failures: Ensure correct API key usage (anon vs service role), verify JWT token validity
Query Errors: Check table names and column spelling, verify RLS policies allow access
File Upload Issues: Verify storage bucket exists and RLS policies configured correctly

Conclusion

Building Python backend applications with Supabase enables leveraging PostgreSQL database, authentication, real-time subscriptions, and storage through Python client library creating powerful server-side applications. By installing Supabase Python client with proper dependencies, connecting to Supabase with configuration management, performing database operations with comprehensive CRUD support, implementing authentication and user management with JWT tokens, building REST APIs with FastAPI integration providing automatic documentation, handling real-time subscriptions monitoring database changes, managing file uploads to storage with security, and deploying Python backends to production, you build robust backend services serving diverse use cases. Python integration advantages include rich ecosystem accessing data science libraries, strong typing with Pydantic ensuring safety, async support enabling high performance, FastAPI integration providing modern API development, flexible deployment supporting various hosting platforms, and backend versatility handling web APIs through data pipelines. Always use environment variables securing credentials, implement error handling gracefully, use type hints with Pydantic, leverage dependency injection, implement async operations, use pagination limiting data, protect routes with authentication, validate data thoroughly, reuse client instances, and implement logging tracking operations. Python integration demonstrates backend development versatility supporting everything from REST APIs to machine learning workflows. Continue exploring query techniques and real-time features.