Skills
skills/franciscosanchezn/easyfactu-es/supabase-integration

supabase-integration

SKILL.md

Supabase Integration

Integrate Supabase for authentication, database management, storage, and real-time features across Python and TypeScript applications.

When to Use This Skill

  • Setting up Supabase Auth (email/password, social, magic link)
  • Validating JWTs in Python or TypeScript backends
  • Creating and managing PostgreSQL schemas per project/environment
  • Implementing Row-Level Security (RLS) policies
  • Configuring Supabase clients in Python or TypeScript
  • Managing file uploads with Supabase Storage
  • Planning around free tier limitations

Free Tier Awareness

Limits (as of 2026)

Resource Free Tier Limit
Database size 500 MB
Bandwidth 5 GB / month
Monthly Active Users 50,000
Active projects 2
Storage 1 GB
Edge Function invocations 500K / month
Pause after inactivity 7 days

Strategy: One Project, Multiple Schemas

Use a single Supabase project with schema-per-project-environment isolation to stay within the 2-project limit:

-- Schema naming convention
CREATE SCHEMA easyfactu_prod;           -- EasyFactu production
CREATE SCHEMA easyfactu_dev;            -- EasyFactu development
CREATE SCHEMA easyfactu_tenant_acme;    -- Tenant-specific schema
CREATE SCHEMA shared;                   -- Cross-project shared data

Migration to Pro ($25/mo)

Migrate when approaching these thresholds:

  • Database > 400 MB (80% of 500 MB limit)
  • Need more than 2 active projects
  • Require point-in-time recovery
  • Need daily backups
  • Project pausing is unacceptable (production)

Supabase Auth

Email/Password Registration and Login

TypeScript Client

import { createClient } from "@supabase/supabase-js";

const supabase = createClient(
  import.meta.env.VITE_SUPABASE_URL,
  import.meta.env.VITE_SUPABASE_ANON_KEY,
);

// Sign up
async function signUp(email: string, password: string) {
  const { data, error } = await supabase.auth.signUp({
    email,
    password,
    options: {
      data: {
        tenant_id: "acme",  // Custom user metadata
        role: "user",
      },
    },
  });
  if (error) throw error;
  return data;
}

// Sign in
async function signIn(email: string, password: string) {
  const { data, error } = await supabase.auth.signInWithPassword({
    email,
    password,
  });
  if (error) throw error;
  return data;
}

// Sign out
async function signOut() {
  const { error } = await supabase.auth.signOut();
  if (error) throw error;
}

// Get current session
async function getSession() {
  const { data: { session } } = await supabase.auth.getSession();
  return session;
}

// Listen for auth state changes
supabase.auth.onAuthStateChange((event, session) => {
  console.log("Auth event:", event, session);
});

Python Client

from supabase import create_client, Client

supabase: Client = create_client(
    supabase_url="https://your-project.supabase.co",
    supabase_key="your-anon-key",
)

# Sign up
def sign_up(email: str, password: str) -> dict:
    response = supabase.auth.sign_up({
        "email": email,
        "password": password,
        "options": {
            "data": {"tenant_id": "acme", "role": "user"},
        },
    })
    return response

# Sign in
def sign_in(email: str, password: str) -> dict:
    response = supabase.auth.sign_in_with_password({
        "email": email,
        "password": password,
    })
    return response

JWT Validation (Python Backend)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import jwt, JWTError
from pydantic import BaseModel

security = HTTPBearer()

class AuthUser(BaseModel):
    """Authenticated user from Supabase JWT."""
    id: str
    email: str
    tenant_id: str | None = None
    role: str = "user"


async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> AuthUser:
    """Validate Supabase JWT and extract user info."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            SUPABASE_JWT_SECRET,
            algorithms=["HS256"],
            audience="authenticated",
        )
        return AuthUser(
            id=payload["sub"],
            email=payload.get("email", ""),
            tenant_id=payload.get("user_metadata", {}).get("tenant_id"),
            role=payload.get("user_metadata", {}).get("role", "user"),
        )
    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or expired token",
        ) from e

Social Login Providers

// Google OAuth
async function signInWithGoogle() {
  const { data, error } = await supabase.auth.signInWithOAuth({
    provider: "google",
    options: {
      redirectTo: `${window.location.origin}/auth/callback`,
    },
  });
  if (error) throw error;
  return data;
}

// Handle OAuth callback (in callback page component)
async function handleAuthCallback() {
  const { data: { session }, error } = await supabase.auth.getSession();
  if (error) throw error;
  if (session) {
    // Redirect to dashboard
    window.location.href = "/dashboard";
  }
}

Protected Routes (Frontend)

import { Navigate, Outlet } from "react-router-dom";
import { useAuth } from "@/hooks/useAuth";

export function ProtectedRoute() {
  const { user, isLoading } = useAuth();

  if (isLoading) {
    return <LoadingSpinner />;
  }

  if (!user) {
    return <Navigate to="/login" replace />;
  }

  return <Outlet />;
}

// In router config:
// <Route element={<ProtectedRoute />}>
//   <Route path="/dashboard" element={<Dashboard />} />
//   <Route path="/invoices" element={<InvoiceList />} />
// </Route>

Protected Endpoints (Backend)

from fastapi import APIRouter, Depends

router = APIRouter()

@router.get("/invoices")
async def list_invoices(
    user: AuthUser = Depends(get_current_user),  # Auth required
    db: AsyncSession = Depends(get_db),
) -> list[InvoiceResponse]:
    """Only authenticated users can list invoices."""
    return await invoice_service.list_by_tenant(db, user.tenant_id)


@router.delete("/invoices/{invoice_id}")
async def delete_invoice(
    invoice_id: str,
    user: AuthUser = Depends(require_admin),  # Admin only
    db: AsyncSession = Depends(get_db),
) -> None:
    """Only admins can delete invoices."""
    await invoice_service.delete(db, invoice_id)

Database Schema Management

Schema-per-Project-Environment

-- Create schema for each project + environment
CREATE SCHEMA IF NOT EXISTS easyfactu_prod;
CREATE SCHEMA IF NOT EXISTS easyfactu_dev;

-- Grant access to the authenticated role
GRANT USAGE ON SCHEMA easyfactu_prod TO authenticated;
GRANT ALL ON ALL TABLES IN SCHEMA easyfactu_prod TO authenticated;

-- Default privileges for future tables
ALTER DEFAULT PRIVILEGES IN SCHEMA easyfactu_prod
    GRANT ALL ON TABLES TO authenticated;

Schema-per-Tenant (Within a Project)

-- Create tenant schema dynamically
CREATE OR REPLACE FUNCTION create_tenant_schema(tenant_slug TEXT)
RETURNS VOID AS $$
DECLARE
    schema_name TEXT := 'easyfactu_tenant_' || tenant_slug;
BEGIN
    EXECUTE format('CREATE SCHEMA IF NOT EXISTS %I', schema_name);

    -- Create standard tables in the new schema
    EXECUTE format('
        CREATE TABLE IF NOT EXISTS %I.invoices (
            id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
            invoice_number TEXT NOT NULL,
            amount DECIMAL(12,2) NOT NULL CHECK (amount > 0),
            currency TEXT NOT NULL DEFAULT ''EUR'',
            status TEXT NOT NULL DEFAULT ''draft'',
            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        )', schema_name);

    -- Enable RLS
    EXECUTE format('ALTER TABLE %I.invoices ENABLE ROW LEVEL SECURITY', schema_name);

    -- Grant access
    EXECUTE format('GRANT USAGE ON SCHEMA %I TO authenticated', schema_name);
    EXECUTE format('GRANT ALL ON ALL TABLES IN SCHEMA %I TO authenticated', schema_name);
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

Migrations

Using SQL Migration Files

-- migrations/001_create_invoices.sql
BEGIN;

CREATE TABLE IF NOT EXISTS easyfactu_prod.invoices (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id TEXT NOT NULL,
    customer_id UUID NOT NULL,
    invoice_number TEXT NOT NULL,
    amount DECIMAL(12,2) NOT NULL CHECK (amount > 0),
    currency TEXT NOT NULL DEFAULT 'EUR',
    status TEXT NOT NULL DEFAULT 'draft'
        CHECK (status IN ('draft', 'sent', 'paid', 'cancelled')),
    issued_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    UNIQUE(tenant_id, invoice_number)
);

-- RLS
ALTER TABLE easyfactu_prod.invoices ENABLE ROW LEVEL SECURITY;

-- Indexes
CREATE INDEX idx_invoices_tenant ON easyfactu_prod.invoices(tenant_id);
CREATE INDEX idx_invoices_status ON easyfactu_prod.invoices(tenant_id, status);

COMMIT;

Using Alembic (Python)

# alembic/env.py - configure for schema-per-tenant
from alembic import context
import os

def run_migrations_online():
    schema = os.environ.get("TENANT_SCHEMA", "easyfactu_prod")
    connectable = engine_from_config(config.get_section("alembic"))

    with connectable.connect() as connection:
        connection.execute(text(f"SET search_path TO {schema}, public"))
        context.configure(connection=connection, target_metadata=metadata)

        with context.begin_transaction():
            context.run_migrations()

Row-Level Security (RLS)

Basic Tenant Isolation

-- Tenant isolation using app.current_tenant session variable
CREATE POLICY "tenant_isolation" ON easyfactu_prod.invoices
    FOR ALL
    USING (tenant_id = current_setting('app.current_tenant', true))
    WITH CHECK (tenant_id = current_setting('app.current_tenant', true));

-- Set tenant context per request
SET app.current_tenant = 'acme';

JWT-Based Policies

-- Use Supabase auth.jwt() for tenant resolution
CREATE POLICY "jwt_tenant_isolation" ON easyfactu_prod.invoices
    FOR ALL
    USING (
        tenant_id = (auth.jwt() -> 'user_metadata' ->> 'tenant_id')::TEXT
    );

-- Admin bypass
CREATE POLICY "admin_full_access" ON easyfactu_prod.invoices
    FOR ALL
    USING (
        (auth.jwt() -> 'user_metadata' ->> 'role')::TEXT = 'admin'
    );

-- Read-only for viewers
CREATE POLICY "viewer_read_only" ON easyfactu_prod.invoices
    FOR SELECT
    USING (
        tenant_id = (auth.jwt() -> 'user_metadata' ->> 'tenant_id')::TEXT
        AND (auth.jwt() -> 'user_metadata' ->> 'role')::TEXT IN ('viewer', 'user', 'admin')
    );

When to Use RLS vs Schema Isolation

Aspect RLS Schema Isolation
Setup complexity Simple per-table Complex (schema creation, migrations)
Query performance Slight overhead per query No overhead
Data isolation Logical (policy-based) Physical (separate schema)
Migration effort One migration per table One migration per schema
Best for Few tenants, shared queries Many tenants, strict isolation
Recommended MVP / early stage Production with compliance needs

Client Integration

TypeScript Client Setup

import { createClient } from "@supabase/supabase-js";

// Type-safe database types (generate with Supabase CLI)
import type { Database } from "./database.types";

export const supabase = createClient<Database>(
  import.meta.env.VITE_SUPABASE_URL,
  import.meta.env.VITE_SUPABASE_ANON_KEY,
);

// Type-safe queries
const { data: invoices, error } = await supabase
  .schema("easyfactu_prod")
  .from("invoices")
  .select("*")
  .eq("status", "draft")
  .order("created_at", { ascending: false });

Python Client Setup

from supabase import create_client, Client
from pydantic_settings import BaseSettings


class SupabaseSettings(BaseSettings):
    supabase_url: str
    supabase_key: str
    supabase_service_key: str  # For server-side operations

    model_config = {"env_prefix": "APP_"}


def get_supabase_client(settings: SupabaseSettings) -> Client:
    """Create Supabase client for authenticated user operations."""
    return create_client(settings.supabase_url, settings.supabase_key)


def get_supabase_admin(settings: SupabaseSettings) -> Client:
    """Create Supabase admin client (bypasses RLS)."""
    return create_client(settings.supabase_url, settings.supabase_service_key)

Generate TypeScript Types

# Install Supabase CLI
npx supabase login

# Generate types from remote database
npx supabase gen types typescript \
  --project-id your-project-id \
  --schema easyfactu_prod \
  > src/types/database.types.ts

# Or from local development
npx supabase gen types typescript \
  --local \
  --schema easyfactu_prod \
  > src/types/database.types.ts

Supabase Storage

Bucket Setup

-- Create a storage bucket for invoice PDFs
INSERT INTO storage.buckets (id, name, public)
VALUES ('invoices', 'invoices', false);

-- RLS policy: only authenticated users can access their tenant's files
CREATE POLICY "tenant_invoice_files" ON storage.objects
    FOR ALL
    USING (
        bucket_id = 'invoices'
        AND (storage.foldername(name))[1] = (auth.jwt() -> 'user_metadata' ->> 'tenant_id')
    );

File Upload (TypeScript)

async function uploadInvoicePdf(
  tenantId: string,
  invoiceId: string,
  file: File,
) {
  const path = `${tenantId}/${invoiceId}.pdf`;

  const { data, error } = await supabase.storage
    .from("invoices")
    .upload(path, file, {
      contentType: "application/pdf",
      upsert: true,
    });

  if (error) throw error;
  return data;
}

// Get signed URL (temporary access)
async function getInvoiceUrl(tenantId: string, invoiceId: string) {
  const { data, error } = await supabase.storage
    .from("invoices")
    .createSignedUrl(`${tenantId}/${invoiceId}.pdf`, 3600); // 1 hour

  if (error) throw error;
  return data.signedUrl;
}

File Upload (Python)

async def upload_invoice_pdf(
    supabase: Client,
    tenant_id: str,
    invoice_id: str,
    file_content: bytes,
) -> str:
    """Upload invoice PDF to Supabase Storage."""
    path = f"{tenant_id}/{invoice_id}.pdf"
    supabase.storage.from_("invoices").upload(
        path=path,
        file=file_content,
        file_options={"content-type": "application/pdf", "upsert": "true"},
    )
    return path

Testing Patterns

Testing with Supabase Local Development

# Start local Supabase
npx supabase start

# Run migrations against local
npx supabase db push

# Reset local database
npx supabase db reset

Mock Supabase in Tests (Python)

import pytest
from unittest.mock import AsyncMock, MagicMock


@pytest.fixture
def mock_supabase() -> MagicMock:
    """Mock Supabase client for testing."""
    client = MagicMock()
    client.auth.sign_in_with_password.return_value = {
        "user": {"id": "user-123", "email": "test@example.com"},
        "session": {"access_token": "test-token"},
    }
    client.table.return_value.select.return_value.execute.return_value = MagicMock(
        data=[{"id": "inv-1", "amount": 100}],
    )
    return client

Mock Supabase in Tests (TypeScript)

import { vi } from "vitest";

const mockSupabase = {
  auth: {
    getSession: vi.fn().mockResolvedValue({
      data: { session: { access_token: "test-token", user: { id: "user-123" } } },
      error: null,
    }),
    signInWithPassword: vi.fn(),
    signOut: vi.fn(),
    onAuthStateChange: vi.fn().mockReturnValue({
      data: { subscription: { unsubscribe: vi.fn() } },
    }),
  },
  from: vi.fn().mockReturnValue({
    select: vi.fn().mockReturnThis(),
    insert: vi.fn().mockReturnThis(),
    update: vi.fn().mockReturnThis(),
    eq: vi.fn().mockReturnThis(),
    single: vi.fn().mockResolvedValue({ data: {}, error: null }),
  }),
};

vi.mock("@/lib/supabase", () => ({ supabase: mockSupabase }));

Edge Functions

Deno-Based Edge Function Pattern

// supabase/functions/process-invoice/index.ts
import { serve } from 'https://deno.land/std@0.177.0/http/server.ts';
import { createClient } from 'https://esm.sh/@supabase/supabase-js@2';

serve(async (req: Request) => {
  try {
    const supabase = createClient(
      Deno.env.get('SUPABASE_URL')!,
      Deno.env.get('SUPABASE_SERVICE_ROLE_KEY')!,
    );

    const { invoice_id } = await req.json();

    // Process with service role (bypasses RLS)
    const { data, error } = await supabase
      .from('invoices')
      .update({ status: 'processed' })
      .eq('id', invoice_id)
      .select()
      .single();

    if (error) throw error;

    return new Response(JSON.stringify(data), {
      headers: { 'Content-Type': 'application/json' },
    });
  } catch (error) {
    return new Response(JSON.stringify({ error: error.message }), {
      status: 400,
      headers: { 'Content-Type': 'application/json' },
    });
  }
});

Connection Management

Supavisor Modes

  • Use transaction mode for short-lived connections (API handlers)
  • Use session mode when SET search_path or prepared statements are needed
  • Pool connections in Python with SQLAlchemy async

Schema-Aware Database Session (Python)

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import text

# Use Supavisor connection string for pooling
engine = create_async_engine(
    settings.supabase_db_url,
    pool_size=5,
    max_overflow=10,
)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db_session(
    user: AuthUser = Depends(get_current_user),
) -> AsyncIterator[AsyncSession]:
    """Create a database session with tenant schema isolation."""
    async with async_session() as session:
        await session.execute(text(f"SET search_path TO {settings.db_schema}"))
        await session.execute(
            text(f"SET app.current_tenant TO '{user.tenant_id}'")
        )
        yield session
        await session.commit()

Free Tier Migration Guide

When to Upgrade

Signal Action
DB approaching 400 MB Plan upgrade within 1 month
Bandwidth > 4 GB/mo Consider upgrading
Need daily backups Upgrade to Pro
Need more than 2 projects Upgrade to Pro
Project keeps pausing Enable keep-alive or upgrade

Migration to Self-Hosted PostgreSQL (on K3s)

  1. Export data with pg_dump
  2. Deploy PostgreSQL pod in K3s
  3. Update connection strings in app config
  4. Migrate auth to self-hosted Supabase or switch to custom JWT
  5. Trade-off: Unlimited storage but uses VPS RAM (~256-512 MB)

Guidelines

  • Use a single Supabase project with schema-per-project-environment isolation
  • Always enable RLS on tables containing user or tenant data
  • Validate JWTs server-side using python-jose or equivalent
  • Use Supabase Auth for all authentication flows
  • Generate TypeScript types from the database schema for type safety
  • Plan for free tier limits — monitor storage and bandwidth usage
  • Use signed URLs for private file access (never expose service key to clients)
  • Test RLS policies with different user contexts before deploying
  • Use Supavisor (connection pooler) for serverless workloads
  • Store the service role key only in backend environments (never in frontend)
  • Use Edge Functions for server-side processing that bypasses RLS
  • Use transaction mode for API handlers, session mode for search_path
Weekly Installs
1
Repository
franciscosanche…factu-es
First Seen
12 days ago
Security Audits
Gen Agent Trust HubPass
SocketPass
SnykWarn
Installed on
mcpjam1
claude-code1
junie1
windsurf1
zencoder1
crush1