tessl/pypi-supabase

Supabase client for Python providing database operations, authentication, storage, real-time subscriptions, and edge functions.

Overview

Eval results

Files

Client Management

Name: tessl/pypi-supabase
Author: tessl

Client creation, configuration, and lifecycle management for both synchronous and asynchronous operations. Provides factory functions and client classes with comprehensive configuration options.

Capabilities

Synchronous Client Creation

Creates a synchronous Supabase client instance for standard blocking operations.

def create_client(
    supabase_url: str,
    supabase_key: str, 
    options: Optional[ClientOptions] = None
) -> SyncClient:
    """
    Create synchronous Supabase client.

    Parameters:
    - supabase_url: The URL to the Supabase instance
    - supabase_key: The API key for the Supabase instance  
    - options: Optional client configuration

    Returns:
    SyncClient instance

    Raises:
    SupabaseException: If URL or key is invalid
    """

Usage Example:

from supabase import create_client, ClientOptions

# Basic client creation
client = create_client("https://project.supabase.co", "your-api-key")

# With custom options
options = ClientOptions(
    schema="private",
    auto_refresh_token=False,
    persist_session=True
)
client = create_client("https://project.supabase.co", "your-api-key", options)

Asynchronous Client Creation

Creates an asynchronous Supabase client instance for non-blocking operations.

async def create_async_client(
    supabase_url: str,
    supabase_key: str,
    options: Optional[AsyncClientOptions] = None  
) -> AsyncClient:
    """
    Create asynchronous Supabase client.

    Parameters:
    - supabase_url: The URL to the Supabase instance
    - supabase_key: The API key for the Supabase instance
    - options: Optional async client configuration

    Returns:
    AsyncClient instance

    Raises:
    SupabaseException: If URL or key is invalid
    """

# Alternative name
async def acreate_client(
    supabase_url: str,
    supabase_key: str, 
    options: Optional[AsyncClientOptions] = None
) -> AsyncClient:
    """Alias for create_async_client"""

Usage Example:

from supabase import create_async_client, AsyncClientOptions
import asyncio

async def main():
    # Basic async client creation
    client = await create_async_client("https://project.supabase.co", "your-api-key")
    
    # With custom options
    options = AsyncClientOptions(
        schema="private", 
        auto_refresh_token=False
    )
    client = await create_async_client("https://project.supabase.co", "your-api-key", options)

asyncio.run(main())

Synchronous Client Class

Main synchronous client class providing access to all Supabase services.

class SyncClient:
    """Main synchronous Supabase client class."""
    
    def __init__(
        self,
        supabase_url: str,
        supabase_key: str,
        options: Optional[ClientOptions] = None
    ):
        """
        Initialize client instance.

        Parameters:
        - supabase_url: Supabase instance URL
        - supabase_key: API key
        - options: Client configuration options
        """
    
    @classmethod
    def create(
        cls,
        supabase_url: str, 
        supabase_key: str,
        options: Optional[ClientOptions] = None
    ) -> 'SyncClient':
        """Factory method for creating client with session handling."""

    # Service properties
    @property
    def auth(self) -> SyncSupabaseAuthClient:
        """Authentication client"""
    
    @property  
    def postgrest(self) -> SyncPostgrestClient:
        """Database client"""
        
    @property
    def storage(self) -> SupabaseStorageClient:
        """Storage client"""
        
    @property
    def functions(self) -> SyncFunctionsClient:
        """Functions client"""
        
    @property
    def realtime(self) -> SyncRealtimeClient:
        """Realtime client"""

Asynchronous Client Class

Main asynchronous client class providing access to all Supabase services with async/await support.

class AsyncClient:
    """Main asynchronous Supabase client class."""
    
    def __init__(
        self,
        supabase_url: str,
        supabase_key: str, 
        options: Optional[AsyncClientOptions] = None
    ):
        """
        Initialize async client instance.

        Parameters:
        - supabase_url: Supabase instance URL
        - supabase_key: API key
        - options: Async client configuration options
        """
    
    @classmethod
    async def create(
        cls,
        supabase_url: str,
        supabase_key: str,
        options: Optional[AsyncClientOptions] = None
    ) -> 'AsyncClient':
        """Async factory method for creating client with session handling."""

    # Service properties (same as sync but with async clients)
    @property
    def auth(self) -> AsyncSupabaseAuthClient:
        """Async authentication client"""
    
    @property
    def postgrest(self) -> AsyncPostgrestClient:
        """Async database client"""
        
    @property
    def storage(self) -> AsyncSupabaseStorageClient:
        """Async storage client"""
        
    @property
    def functions(self) -> AsyncFunctionsClient:
        """Async functions client"""
        
    @property
    def realtime(self) -> AsyncRealtimeClient:
        """Async realtime client"""

Client Configuration Options

Configuration options for customizing client behavior and service integration.

@dataclass
class ClientOptions:
    """Synchronous client configuration options."""
    
    schema: str = "public"
    """PostgreSQL schema for database operations"""
    
    headers: Dict[str, str] = field(default_factory=lambda: {"X-Client-Info": f"supabase-py/{__version__}"})
    """Custom headers for HTTP requests (includes default X-Client-Info header)"""
    
    auto_refresh_token: bool = True
    """Automatically refresh authentication tokens"""
    
    persist_session: bool = True  
    """Persist user session to storage"""
    
    storage: SyncSupportedStorage = field(default_factory=SyncMemoryStorage)
    """Storage provider for session persistence"""
    
    realtime: Optional[RealtimeClientOptions] = None
    """Realtime client configuration"""
    
    httpx_client: Optional[SyncHttpxClient] = None
    """Custom HTTP client for requests"""
    
    postgrest_client_timeout: Union[int, float, Timeout] = DEFAULT_POSTGREST_CLIENT_TIMEOUT
    """Database client request timeout"""
    
    storage_client_timeout: Union[int, float, Timeout] = DEFAULT_STORAGE_CLIENT_TIMEOUT
    """Storage client request timeout"""
    
    function_client_timeout: Union[int, float, Timeout] = DEFAULT_FUNCTION_CLIENT_TIMEOUT
    """Functions client request timeout"""
    
    flow_type: AuthFlowType = "pkce"
    """Authentication flow type (pkce, implicit)"""
    
    def replace(self, **kwargs) -> 'ClientOptions':
        """Create new ClientOptions instance with specified changes."""

@dataclass
class AsyncClientOptions(ClientOptions):
    """Asynchronous client configuration options."""
    
    storage: AsyncSupportedStorage = field(default_factory=AsyncMemoryStorage)
    """Async storage provider for session persistence"""
    
    httpx_client: Optional[AsyncHttpxClient] = None
    """Custom async HTTP client for requests"""
    
    def replace(self, **kwargs) -> 'AsyncClientOptions':
        """Create new AsyncClientOptions instance with specified changes."""

Usage Example:

from supabase import ClientOptions, AsyncClientOptions
from supabase_auth import SyncMemoryStorage
from httpx import Client as SyncHttpxClient

# Sync client options
sync_options = ClientOptions(
    schema="private",
    headers={"Custom-Header": "value"},
    auto_refresh_token=True,
    persist_session=True,
    storage=SyncMemoryStorage(),
    httpx_client=SyncHttpxClient(timeout=30),
    flow_type="pkce"
)

# Create new options with changes
modified_options = sync_options.replace(
    schema="public",
    auto_refresh_token=False
)

# Async client options
async_options = AsyncClientOptions(
    schema="private",
    auto_refresh_token=True
)

Error Handling

class SupabaseException(Exception):
    """Base exception for client errors."""
    
    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

# Sync and async variants
class SyncSupabaseException(SupabaseException):
    """Synchronous client exception"""

class AsyncSupabaseException(SupabaseException):  
    """Asynchronous client exception"""

Common Error Scenarios:

from supabase import create_client, SupabaseException

try:
    # Invalid URL format
    client = create_client("invalid-url", "key")
except SupabaseException as e:
    print(f"Client creation failed: {e.message}")

try:
    # Missing required parameters
    client = create_client("", "")
except SupabaseException as e:
    print(f"Missing parameters: {e.message}")

Install with Tessl CLI