Supabase + Python Skill

Load with: base.md + supabase.md + python.md

FastAPI patterns with Supabase Auth and SQLAlchemy/SQLModel for database access.

Sources: Supabase Python Client | SQLModel

Core Principle

SQLAlchemy/SQLModel for queries, Supabase for auth/storage.

Use SQLAlchemy or SQLModel for type-safe database access. Use supabase-py for auth, storage, and realtime. FastAPI for the API layer.

Project Structure

project/
├── src/
│   ├── api/
│   │   ├── __init__.py
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── auth.py
│   │   │   ├── posts.py
│   │   │   └── users.py
│   │   └── deps.py              # Dependencies (auth, db)
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py            # Settings
│   │   └── security.py          # Auth helpers
│   ├── db/
│   │   ├── __init__.py
│   │   ├── session.py           # Database session
│   │   └── models.py            # SQLModel models
│   ├── services/
│   │   ├── __init__.py
│   │   └── supabase.py          # Supabase client
│   └── main.py                  # FastAPI app
├── supabase/
│   ├── migrations/
│   └── config.toml
├── alembic/                     # Alembic migrations (alternative)
├── alembic.ini
├── pyproject.toml
└── .env

Setup

Install Dependencies

bash
1pip install fastapi uvicorn supabase python-dotenv sqlmodel asyncpg alembic

pyproject.toml

toml
1[project]
2name = "my-app"
3version = "0.1.0"
4dependencies = [
5    "fastapi>=0.109.0",
6    "uvicorn[standard]>=0.27.0",
7    "supabase>=2.0.0",
8    "python-dotenv>=1.0.0",
9    "sqlmodel>=0.0.14",
10    "asyncpg>=0.29.0",
11    "alembic>=1.13.0",
12    "pydantic-settings>=2.0.0",
13]
14
15[project.optional-dependencies]
16dev = [
17    "pytest>=7.0.0",
18    "pytest-asyncio>=0.23.0",
19    "httpx>=0.26.0",
20]

Environment Variables

bash
1# .env
2SUPABASE_URL=http://localhost:54321
3SUPABASE_ANON_KEY=<from supabase start>
4SUPABASE_SERVICE_ROLE_KEY=<from supabase start>
5DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:54322/postgres

Configuration

src/core/config.py

python
1from pydantic_settings import BaseSettings
2from functools import lru_cache
3
4
5class Settings(BaseSettings):
6    # Supabase
7    supabase_url: str
8    supabase_anon_key: str
9    supabase_service_role_key: str
10
11    # Database
12    database_url: str
13
14    # App
15    debug: bool = False
16
17    class Config:
18        env_file = ".env"
19        env_file_encoding = "utf-8"
20
21
22@lru_cache
23def get_settings() -> Settings:
24    return Settings()

Database Setup

src/db/session.py

python
1from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
2from sqlalchemy.orm import sessionmaker
3from src.core.config import get_settings
4
5settings = get_settings()
6
7engine = create_async_engine(
8    settings.database_url,
9    echo=settings.debug,
10    pool_pre_ping=True,
11)
12
13AsyncSessionLocal = sessionmaker(
14    engine,
15    class_=AsyncSession,
16    expire_on_commit=False,
17)
18
19
20async def get_db() -> AsyncSession:
21    async with AsyncSessionLocal() as session:
22        try:
23            yield session
24        finally:
25            await session.close()

src/db/models.py

python
1from datetime import datetime
2from typing import Optional
3from uuid import UUID, uuid4
4from sqlmodel import SQLModel, Field
5
6
7class ProfileBase(SQLModel):
8    email: str
9    name: Optional[str] = None
10    avatar_url: Optional[str] = None
11
12
13class Profile(ProfileBase, table=True):
14    __tablename__ = "profiles"
15
16    id: UUID = Field(primary_key=True)  # References auth.users
17    created_at: datetime = Field(default_factory=datetime.utcnow)
18    updated_at: datetime = Field(default_factory=datetime.utcnow)
19
20
21class ProfileCreate(ProfileBase):
22    id: UUID
23
24
25class ProfileRead(ProfileBase):
26    id: UUID
27    created_at: datetime
28
29
30class PostBase(SQLModel):
31    title: str
32    content: Optional[str] = None
33    published: bool = False
34
35
36class Post(PostBase, table=True):
37    __tablename__ = "posts"
38
39    id: UUID = Field(default_factory=uuid4, primary_key=True)
40    author_id: UUID = Field(foreign_key="profiles.id")
41    created_at: datetime = Field(default_factory=datetime.utcnow)
42
43
44class PostCreate(PostBase):
45    pass
46
47
48class PostRead(PostBase):
49    id: UUID
50    author_id: UUID
51    created_at: datetime

Supabase Client

src/services/supabase.py

python
1from supabase import create_client, Client
2from src.core.config import get_settings
3
4settings = get_settings()
5
6
7def get_supabase_client() -> Client:
8    """Get Supabase client with anon key (respects RLS)."""
9    return create_client(
10        settings.supabase_url,
11        settings.supabase_anon_key
12    )
13
14
15def get_supabase_admin() -> Client:
16    """Get Supabase client with service role (bypasses RLS)."""
17    return create_client(
18        settings.supabase_url,
19        settings.supabase_service_role_key
20    )

Auth Dependencies

src/api/deps.py

python
1from typing import Annotated
2from fastapi import Depends, HTTPException, status
3from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
4from sqlalchemy.ext.asyncio import AsyncSession
5from supabase import Client
6
7from src.db.session import get_db
8from src.services.supabase import get_supabase_client
9
10security = HTTPBearer()
11
12
13async def get_current_user(
14    credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
15) -> dict:
16    """Validate JWT and return user."""
17    supabase = get_supabase_client()
18
19    try:
20        # Verify token with Supabase
21        user = supabase.auth.get_user(credentials.credentials)
22        if not user or not user.user:
23            raise HTTPException(
24                status_code=status.HTTP_401_UNAUTHORIZED,
25                detail="Invalid token",
26            )
27        return user.user
28    except Exception as e:
29        raise HTTPException(
30            status_code=status.HTTP_401_UNAUTHORIZED,
31            detail="Invalid token",
32        )
33
34
35# Type alias for dependency injection
36CurrentUser = Annotated[dict, Depends(get_current_user)]
37DbSession = Annotated[AsyncSession, Depends(get_db)]

API Routes

src/api/routes/auth.py

python
1from fastapi import APIRouter, HTTPException, status
2from pydantic import BaseModel, EmailStr
3
4from src.services.supabase import get_supabase_client
5
6router = APIRouter(prefix="/auth", tags=["auth"])
7
8
9class SignUpRequest(BaseModel):
10    email: EmailStr
11    password: str
12
13
14class SignInRequest(BaseModel):
15    email: EmailStr
16    password: str
17
18
19class AuthResponse(BaseModel):
20    access_token: str
21    refresh_token: str
22    user_id: str
23
24
25@router.post("/signup", response_model=AuthResponse)
26async def sign_up(request: SignUpRequest):
27    supabase = get_supabase_client()
28
29    try:
30        response = supabase.auth.sign_up({
31            "email": request.email,
32            "password": request.password,
33        })
34
35        if response.user is None:
36            raise HTTPException(
37                status_code=status.HTTP_400_BAD_REQUEST,
38                detail="Signup failed",
39            )
40
41        return AuthResponse(
42            access_token=response.session.access_token,
43            refresh_token=response.session.refresh_token,
44            user_id=str(response.user.id),
45        )
46    except Exception as e:
47        raise HTTPException(
48            status_code=status.HTTP_400_BAD_REQUEST,
49            detail=str(e),
50        )
51
52
53@router.post("/signin", response_model=AuthResponse)
54async def sign_in(request: SignInRequest):
55    supabase = get_supabase_client()
56
57    try:
58        response = supabase.auth.sign_in_with_password({
59            "email": request.email,
60            "password": request.password,
61        })
62
63        return AuthResponse(
64            access_token=response.session.access_token,
65            refresh_token=response.session.refresh_token,
66            user_id=str(response.user.id),
67        )
68    except Exception as e:
69        raise HTTPException(
70            status_code=status.HTTP_401_UNAUTHORIZED,
71            detail="Invalid credentials",
72        )
73
74
75@router.post("/signout")
76async def sign_out():
77    supabase = get_supabase_client()
78    supabase.auth.sign_out()
79    return {"message": "Signed out"}

src/api/routes/posts.py

python
1from uuid import UUID
2from fastapi import APIRouter, HTTPException, status
3from sqlmodel import select
4
5from src.api.deps import CurrentUser, DbSession
6from src.db.models import Post, PostCreate, PostRead
7
8router = APIRouter(prefix="/posts", tags=["posts"])
9
10
11@router.get("/", response_model=list[PostRead])
12async def list_posts(
13    db: DbSession,
14    published_only: bool = True,
15):
16    query = select(Post)
17    if published_only:
18        query = query.where(Post.published == True)
19    query = query.order_by(Post.created_at.desc())
20
21    result = await db.execute(query)
22    return result.scalars().all()
23
24
25@router.get("/me", response_model=list[PostRead])
26async def list_my_posts(
27    db: DbSession,
28    user: CurrentUser,
29):
30    query = select(Post).where(Post.author_id == UUID(user.id))
31    result = await db.execute(query)
32    return result.scalars().all()
33
34
35@router.post("/", response_model=PostRead, status_code=status.HTTP_201_CREATED)
36async def create_post(
37    db: DbSession,
38    user: CurrentUser,
39    post_in: PostCreate,
40):
41    post = Post(
42        **post_in.model_dump(),
43        author_id=UUID(user.id),
44    )
45    db.add(post)
46    await db.commit()
47    await db.refresh(post)
48    return post
49
50
51@router.get("/{post_id}", response_model=PostRead)
52async def get_post(
53    db: DbSession,
54    post_id: UUID,
55):
56    result = await db.execute(select(Post).where(Post.id == post_id))
57    post = result.scalar_one_or_none()
58
59    if not post:
60        raise HTTPException(
61            status_code=status.HTTP_404_NOT_FOUND,
62            detail="Post not found",
63        )
64
65    return post
66
67
68@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
69async def delete_post(
70    db: DbSession,
71    user: CurrentUser,
72    post_id: UUID,
73):
74    result = await db.execute(
75        select(Post).where(Post.id == post_id, Post.author_id == UUID(user.id))
76    )
77    post = result.scalar_one_or_none()
78
79    if not post:
80        raise HTTPException(
81            status_code=status.HTTP_404_NOT_FOUND,
82            detail="Post not found",
83        )
84
85    await db.delete(post)
86    await db.commit()

Main Application

src/main.py

python
1from fastapi import FastAPI
2from fastapi.middleware.cors import CORSMiddleware
3
4from src.api.routes import auth, posts
5
6app = FastAPI(title="My API")
7
8# CORS
9app.add_middleware(
10    CORSMiddleware,
11    allow_origins=["*"],  # Configure for production
12    allow_credentials=True,
13    allow_methods=["*"],
14    allow_headers=["*"],
15)
16
17# Routes
18app.include_router(auth.router, prefix="/api")
19app.include_router(posts.router, prefix="/api")
20
21
22@app.get("/health")
23async def health_check():
24    return {"status": "healthy"}

Alembic Migrations

Initialize Alembic

bash
1alembic init alembic

alembic/env.py (key changes)

python
1from src.db.models import SQLModel
2from src.core.config import get_settings
3
4settings = get_settings()
5
6# Use async engine
7config.set_main_option("sqlalchemy.url", settings.database_url)
8
9target_metadata = SQLModel.metadata
10
11
12def run_migrations_online():
13    # For async
14    import asyncio
15    from sqlalchemy.ext.asyncio import create_async_engine
16
17    connectable = create_async_engine(settings.database_url)
18
19    async def do_run_migrations():
20        async with connectable.connect() as connection:
21            await connection.run_sync(do_run_migrations_sync)
22
23    def do_run_migrations_sync(connection):
24        context.configure(
25            connection=connection,
26            target_metadata=target_metadata,
27        )
28        with context.begin_transaction():
29            context.run_migrations()
30
31    asyncio.run(do_run_migrations())

Migration Commands

bash
1# Create migration
2alembic revision --autogenerate -m "create posts table"
3
4# Apply migrations
5alembic upgrade head
6
7# Rollback
8alembic downgrade -1

Storage

Upload File

python
1from fastapi import UploadFile
2from src.services.supabase import get_supabase_client
3
4
5async def upload_avatar(user_id: str, file: UploadFile) -> str:
6    supabase = get_supabase_client()
7
8    file_content = await file.read()
9    file_path = f"{user_id}/avatar.{file.filename.split('.')[-1]}"
10
11    response = supabase.storage.from_("avatars").upload(
12        file_path,
13        file_content,
14        {"content-type": file.content_type, "upsert": "true"},
15    )
16
17    # Get public URL
18    url = supabase.storage.from_("avatars").get_public_url(file_path)
19    return url

Download File

python
1def get_avatar_url(user_id: str) -> str:
2    supabase = get_supabase_client()
3    return supabase.storage.from_("avatars").get_public_url(f"{user_id}/avatar.png")

Realtime (Async)

python
1import asyncio
2from supabase import create_client
3
4
5async def listen_to_posts():
6    supabase = create_client(
7        settings.supabase_url,
8        settings.supabase_anon_key
9    )
10
11    def handle_change(payload):
12        print(f"Change received: {payload}")
13
14    channel = supabase.channel("posts")
15    channel.on_postgres_changes(
16        event="*",
17        schema="public",
18        table="posts",
19        callback=handle_change,
20    ).subscribe()
21
22    # Keep listening
23    while True:
24        await asyncio.sleep(1)

Testing

tests/conftest.py

python
1import pytest
2from httpx import AsyncClient, ASGITransport
3from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
4from sqlalchemy.orm import sessionmaker
5
6from src.main import app
7from src.db.session import get_db
8from src.db.models import SQLModel
9
10TEST_DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:54322/postgres_test"
11
12engine = create_async_engine(TEST_DATABASE_URL)
13TestingSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
14
15
16@pytest.fixture(scope="function")
17async def db_session():
18    async with engine.begin() as conn:
19        await conn.run_sync(SQLModel.metadata.create_all)
20
21    async with TestingSessionLocal() as session:
22        yield session
23
24    async with engine.begin() as conn:
25        await conn.run_sync(SQLModel.metadata.drop_all)
26
27
28@pytest.fixture
29async def client(db_session):
30    async def override_get_db():
31        yield db_session
32
33    app.dependency_overrides[get_db] = override_get_db
34
35    async with AsyncClient(
36        transport=ASGITransport(app=app),
37        base_url="http://test",
38    ) as ac:
39        yield ac
40
41    app.dependency_overrides.clear()

tests/test_posts.py

python
1import pytest
2from httpx import AsyncClient
3
4
5@pytest.mark.asyncio
6async def test_list_posts(client: AsyncClient):
7    response = await client.get("/api/posts/")
8    assert response.status_code == 200
9    assert isinstance(response.json(), list)

Running the App

bash
1# Development
2uvicorn src.main:app --reload --port 8000
3
4# Production
5uvicorn src.main:app --host 0.0.0.0 --port 8000 --workers 4

Anti-Patterns

• Using Supabase client for DB queries - Use SQLAlchemy/SQLModel
• Sync database calls - Use async with asyncpg
• Hardcoded credentials - Use environment variables
• No connection pooling - asyncpg handles this
• Missing auth dependency - Always validate JWT
• Not closing sessions - Use context managers
• Blocking I/O in async - Use async libraries