Supabase Python Client

from supabase import create_client, Client

from supabase import create_async_client, AsyncClient

from supabase import ClientOptions, AsyncClientOptions

import os
from supabase import create_client, Client

# Initialize the client
url = os.environ.get("SUPABASE_URL")
key = os.environ.get("SUPABASE_KEY")
supabase: Client = create_client(url, key)

# Database operations
data = supabase.table("countries").insert({"name": "Germany"}).execute()
countries = supabase.table("countries").select("*").execute()

# Authentication
user = supabase.auth.sign_up({"email": "user@example.com", "password": "password123"})
session = supabase.auth.sign_in_with_password({"email": "user@example.com", "password": "password123"})

# Storage operations  
supabase.storage.from_("bucket").upload("file.txt", b"file content")

# Real-time subscriptions
channel = supabase.channel("my-channel")
channel.on("postgres_changes", {"event": "*", "schema": "public", "table": "countries"}, callback)
channel.subscribe()

# Edge functions
response = supabase.functions.invoke("hello-world", {"name": "Functions"})

def create_client(supabase_url: str, supabase_key: str, options: Optional[ClientOptions] = None) -> SyncClient
async def create_async_client(supabase_url: str, supabase_key: str, options: Optional[AsyncClientOptions] = None) -> AsyncClient

def table(self, table_name: str)  # Query builder for table operations
def from_(self, table_name: str)  # Alternative table method
def schema(self, schema: str)     # Schema selection
def rpc(self, fn: str, params: Optional[Dict[Any, Any]] = None, count: Optional[CountMethod] = None, head: bool = False, get: bool = False)  # Stored procedures

# Auth client accessed via client.auth property
# Inherits complete GoTrue API including:
# - sign_up, sign_in_with_password, sign_out
# - OAuth providers, magic links, phone auth
# - Session management and token refresh
# - User profile and metadata management

# Storage client accessed via client.storage property  
# Returns SupabaseStorageClient (sync) or AsyncSupabaseStorageClient (async)
# Provides complete file storage operations including:
# - File upload, download, update, delete
# - Bucket creation and management
# - Access control and signed URLs
# - File metadata and transformations

def channel(self, topic: str, params: RealtimeChannelOptions = {}) -> SyncRealtimeChannel | AsyncRealtimeChannel
def get_channels(self)  # Get all active channels
def remove_channel(self, channel)  # Remove specific channel
def remove_all_channels(self)  # Remove all channels

# Functions client accessed via client.functions property
# Provides edge function invocation with:
# - Function execution with custom headers and data
# - Response handling and error management
# - Integration with auth context

# Core exceptions
class SupabaseException(Exception)
class SyncSupabaseException(SupabaseException)  
class AsyncSupabaseException(SupabaseException)

# Auth exceptions (from supabase_auth)
class AuthApiError(Exception)
class AuthError(Exception)
class AuthInvalidCredentialsError(Exception)
class AuthSessionMissingError(Exception)
class AuthWeakPasswordError(Exception)
class AuthImplicitGrantRedirectError(Exception)

# Database exceptions (from postgrest)
class PostgrestAPIError(Exception)

# Storage exceptions (from storage3)
class StorageException(Exception)

# Functions exceptions (from supabase_functions)
class FunctionsError(Exception)
class FunctionsHttpError(Exception)

# Realtime exceptions (from realtime)
class AuthorizationError(Exception)
class NotConnectedError(Exception)

# Client option types
@dataclass
class ClientOptions:
    schema: str = "public"
    headers: Dict[str, str] = field(default_factory=dict)
    auto_refresh_token: bool = True
    persist_session: bool = True
    storage: SyncSupportedStorage = field(default_factory=SyncMemoryStorage)
    realtime: Optional[RealtimeClientOptions] = None
    httpx_client: Optional[SyncHttpxClient] = None
    postgrest_client_timeout: Union[int, float, Timeout] = DEFAULT_POSTGREST_CLIENT_TIMEOUT
    storage_client_timeout: Union[int, float, Timeout] = DEFAULT_STORAGE_CLIENT_TIMEOUT
    function_client_timeout: Union[int, float, Timeout] = DEFAULT_FUNCTION_CLIENT_TIMEOUT
    flow_type: AuthFlowType = "pkce"

@dataclass  
class AsyncClientOptions(ClientOptions):
    storage: AsyncSupportedStorage = field(default_factory=AsyncMemoryStorage)
    httpx_client: Optional[AsyncHttpxClient] = None

# Realtime options
class RealtimeClientOptions(TypedDict, total=False):
    auto_reconnect: bool
    hb_interval: int
    max_retries: int
    initial_backoff: float