f3da9f3b59
Some checks failed
api-ci-deploy / test-build-deploy (push) Has been cancelled
- client.py: set apikey explicitly in _create_base_client headers (Kong needs it on every request; for per-user clients apikey stays anon while Authorization carries the user JWT). Fixes the 2 stale header unit tests that asserted apikey in options.headers, and is robust against supabase-py default-header changes. - test_dev_stack: exact == seed counts → >= baselines. The greenfield seed sets a floor; additive exam-marker fixtures (S4-4 cohort) legitimately push live .94 counts above the old snapshot. >= still catches a broken/missing seed. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
113 lines
5.2 KiB
Python
113 lines
5.2 KiB
Python
import os
|
|
from typing import Dict, Optional, Any, TypedDict, List
|
|
from supabase import create_client, Client
|
|
from supabase.lib.client_options import SyncClientOptions
|
|
from supabase_auth import SyncMemoryStorage
|
|
from modules.logger_tool import initialise_logger
|
|
|
|
logger = initialise_logger(__name__, os.getenv("LOG_LEVEL"), os.getenv("LOG_PATH"), 'default', True)
|
|
|
|
class CreateBucketOptions(TypedDict, total=False):
|
|
"""Options for bucket creation, matching Supabase API requirements"""
|
|
public: bool
|
|
file_size_limit: int
|
|
allowed_mime_types: List[str]
|
|
name: str
|
|
|
|
def _create_base_client(url: str, key: str, access_token: Optional[str] = None, options: Optional[Dict[str, Any]] = None) -> Client:
|
|
"""Create a base Supabase client with given configuration.
|
|
|
|
If access_token is provided, it is used as the Authorization header (for per-user RLS).
|
|
Otherwise, the API key is used (service role bypasses RLS, anon key does not).
|
|
"""
|
|
# If an access token is provided, use it for Authorization (enables per-user RLS)
|
|
# Otherwise fall back to the API key
|
|
auth_header = f"Bearer {access_token}" if access_token else f"Bearer {key}"
|
|
|
|
# apikey is required by the Supabase gateway (Kong) on every request and is independent of
|
|
# Authorization: for a per-user client apikey stays the anon key while Authorization carries
|
|
# the user's JWT (so RLS sees auth.uid()). Set it explicitly rather than relying on
|
|
# create_client's internal default-header behaviour, which our options.headers override.
|
|
headers = {
|
|
"apikey": key,
|
|
"Authorization": auth_header,
|
|
}
|
|
if options:
|
|
headers.update(options.get("headers", {}))
|
|
|
|
client_options = SyncClientOptions(
|
|
schema="public",
|
|
storage=SyncMemoryStorage(),
|
|
headers=headers,
|
|
)
|
|
return create_client(url, key, options=client_options)
|
|
|
|
class SupabaseServiceRoleClient:
|
|
"""Supabase client for making authenticated requests using the service role key.
|
|
|
|
NOTE: Service role bypasses all RLS policies. Use only for admin operations
|
|
where you need to read/write any row regardless of ownership.
|
|
"""
|
|
|
|
def __init__(self, url: Optional[str] = None, service_role_key: Optional[str] = None):
|
|
"""Initialize the Supabase client with URL and service role key"""
|
|
self.url = url or os.environ.get("SUPABASE_URL", "http://localhost:8000")
|
|
self.service_role_key = service_role_key or os.environ.get("SERVICE_ROLE_KEY")
|
|
|
|
if not self.url or not self.service_role_key:
|
|
raise ValueError("SUPABASE_URL and SERVICE_ROLE_KEY must be provided")
|
|
|
|
# Initialize Supabase client with service role key (bypasses RLS)
|
|
self.supabase = _create_base_client(self.url, self.service_role_key)
|
|
|
|
def create_bucket(self, id: str, options: Optional[CreateBucketOptions] = None) -> Dict[str, Any]:
|
|
"""Create a storage bucket with the given ID and options"""
|
|
if options is None:
|
|
options = CreateBucketOptions()
|
|
if 'name' not in options:
|
|
options['name'] = id # Use ID as default name if not provided
|
|
return self.supabase.storage.create_bucket(id, options=options)
|
|
|
|
class SupabaseAnonClient:
|
|
"""Supabase client for making authenticated requests using the anon key.
|
|
|
|
When initialized with an access_token, per-user RLS policies are enforced
|
|
via auth.uid() in the JWT. Without an access_token, requests use the anon key
|
|
which does NOT enforce per-user RLS (only bucket-level storage rules apply).
|
|
"""
|
|
|
|
def __init__(self, url: Optional[str] = None, anon_key: Optional[str] = None, access_token: Optional[str] = None):
|
|
"""Initialize the Supabase client with URL and anon key.
|
|
|
|
Args:
|
|
url: Supabase URL
|
|
anon_key: Anon API key (fallback if no access_token)
|
|
access_token: User's JWT access token for per-user RLS enforcement
|
|
"""
|
|
self.url = url or os.environ.get("SUPABASE_URL", "http://localhost:8000")
|
|
self.anon_key = anon_key or os.environ.get("ANON_KEY")
|
|
self.access_token = access_token
|
|
|
|
if not self.url or not self.anon_key:
|
|
raise ValueError("SUPABASE_URL and ANON_KEY must be provided")
|
|
|
|
# Initialize Supabase client with anon key and optional access token for RLS
|
|
self.supabase = _create_base_client(self.url, self.anon_key, access_token=access_token)
|
|
|
|
def create_bucket(self, id: str, options: Optional[CreateBucketOptions] = None) -> Dict[str, Any]:
|
|
"""Create a storage bucket with the given ID and options"""
|
|
return self.supabase.storage.create_bucket(id, options=options)
|
|
|
|
@classmethod
|
|
def for_user(cls, access_token: str) -> 'SupabaseAnonClient':
|
|
"""Create a client instance for a specific user using their access token.
|
|
|
|
This enables per-user RLS enforcement via auth.uid() in the JWT.
|
|
"""
|
|
if not access_token or not access_token.strip():
|
|
raise ValueError("access_token is required for per-user Supabase clients")
|
|
token = access_token.strip()
|
|
if token.lower().startswith("bearer "):
|
|
token = token.split(None, 1)[1]
|
|
return cls(access_token=token)
|