tessl/pypi-edgedb

EdgeDB Python driver providing both blocking IO and asyncio implementations for connecting to and interacting with EdgeDB databases.

—

Pending

Overview

Eval results

Files

Client Management

Name: tessl/pypi-edgedb
Author: tessl

Client creation and connection management for EdgeDB, supporting both synchronous and asynchronous programming models with lazy connection pooling and automatic connection management.

Capabilities

Client Factory Functions

Creates EdgeDB clients with flexible connection options including DSN strings, individual connection parameters, and connection pooling configuration.

def create_client(
    dsn: Optional[str] = None,
    *,
    max_concurrency: Optional[int] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    credentials: Optional[Credentials] = None,
    credentials_file: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    secret_key: Optional[str] = None,
    database: Optional[str] = None,
    branch: Optional[str] = None,
    tls_ca: Optional[str] = None,
    tls_ca_file: Optional[str] = None,
    tls_security: Optional[str] = None,
    wait_until_available: int = 30,
    timeout: int = 10
) -> Client:
    """
    Create a synchronous EdgeDB client.
    
    Parameters:
    - dsn: Connection string in EdgeDB DSN format
    - max_concurrency: Maximum number of concurrent connections
    - host: Database host (default: localhost)
    - port: Database port (default: 5656)
    - credentials: Credentials object for authentication
    - credentials_file: Path to credentials file
    - user: Database user name
    - password: Database password
    - secret_key: Secret key for authentication
    - database: Database name
    - branch: Database branch name
    - tls_ca: TLS certificate authority
    - tls_ca_file: Path to TLS CA file
    - tls_security: TLS security mode ('strict', 'no_host_verification', 'insecure')
    - wait_until_available: Seconds to wait for server availability
    - timeout: Connection timeout in seconds
    
    Returns:
    Client instance for synchronous operations
    """

def create_async_client(
    dsn: Optional[str] = None,
    *,
    max_concurrency: Optional[int] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    credentials: Optional[Credentials] = None,
    credentials_file: Optional[str] = None,
    user: Optional[str] = None,
    password: Optional[str] = None,
    secret_key: Optional[str] = None,
    database: Optional[str] = None,
    branch: Optional[str] = None,
    tls_ca: Optional[str] = None,
    tls_ca_file: Optional[str] = None,
    tls_security: Optional[str] = None,
    wait_until_available: int = 30,
    timeout: int = 10
) -> AsyncIOClient:
    """
    Create an asynchronous EdgeDB client.
    
    Parameters: Same as create_client()
    
    Returns:
    AsyncIOClient instance for asynchronous operations
    """

Client Classes

The main client classes providing lazy connection pooling and query execution capabilities.

class Client:
    """
    Synchronous lazy connection pool client for EdgeDB.
    
    Provides blocking I/O operations and automatic connection management
    with connection pooling and retry logic.
    """
    
    def query(self, query: str, *args, **kwargs) -> List[Any]:
        """Execute query and return all results as a list."""
    
    def query_single(self, query: str, *args, **kwargs) -> Optional[Any]:
        """Execute query expecting at most one result."""
    
    def query_required_single(self, query: str, *args, **kwargs) -> Any:
        """Execute query expecting exactly one result."""
    
    def query_json(self, query: str, *args, **kwargs) -> str:
        """Execute query and return results as JSON string."""
    
    def query_single_json(self, query: str, *args, **kwargs) -> str:
        """Execute query expecting at most one result as JSON."""
    
    def query_required_single_json(self, query: str, *args, **kwargs) -> str:
        """Execute query expecting exactly one result as JSON."""
    
    def execute(self, query: str, *args, **kwargs) -> None:
        """Execute command that doesn't return data."""
    
    def transaction(
        self,
        *,
        options: Optional[TransactionOptions] = None,
        retry_options: Optional[RetryOptions] = None
    ) -> ContextManager:
        """Create a transaction context manager."""
    
    def ensure_connected(self) -> None:
        """Ensure the client is connected to the database."""
    
    def close(self) -> None:
        """Close the client and all connections."""

class AsyncIOClient:
    """
    Asynchronous lazy connection pool client for EdgeDB.
    
    Provides async/await operations and automatic connection management
    with connection pooling and retry logic.
    """
    
    async def query(self, query: str, *args, **kwargs) -> List[Any]:
        """Execute query and return all results as a list."""
    
    async def query_single(self, query: str, *args, **kwargs) -> Optional[Any]:
        """Execute query expecting at most one result."""
    
    async def query_required_single(self, query: str, *args, **kwargs) -> Any:
        """Execute query expecting exactly one result."""
    
    async def query_json(self, query: str, *args, **kwargs) -> str:
        """Execute query and return results as JSON string."""
    
    async def query_single_json(self, query: str, *args, **kwargs) -> str:
        """Execute query expecting at most one result as JSON."""
    
    async def query_required_single_json(self, query: str, *args, **kwargs) -> str:
        """Execute query expecting exactly one result as JSON."""
    
    async def execute(self, query: str, *args, **kwargs) -> None:
        """Execute command that doesn't return data."""
    
    def transaction(
        self,
        *,
        options: Optional[TransactionOptions] = None,
        retry_options: Optional[RetryOptions] = None
    ) -> AsyncContextManager:
        """Create an async transaction context manager."""
    
    async def ensure_connected(self) -> None:
        """Ensure the client is connected to the database."""
    
    async def aclose(self) -> None:
        """Close the client and all connections."""

Abstract Executor Interfaces

Abstract base classes defining the contract for EdgeDB query executors, enabling flexible client implementations.

class Executor(ABC):
    """
    Abstract interface for clients that can execute both read-only
    and modification queries.
    """
    
    @abstractmethod
    def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    
    @abstractmethod
    def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    
    @abstractmethod
    def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    
    @abstractmethod
    def query_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    def execute(self, query: str, *args, **kwargs) -> None: ...

class AsyncIOExecutor(ABC):
    """
    Abstract interface for asynchronous clients that can execute both
    read-only and modification queries.
    """
    
    @abstractmethod
    async def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    
    @abstractmethod
    async def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    
    @abstractmethod
    async def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    
    @abstractmethod
    async def query_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    async def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    async def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    async def execute(self, query: str, *args, **kwargs) -> None: ...

class ReadOnlyExecutor(ABC):
    """
    Abstract interface for clients that can execute read-only queries only.
    """
    
    @abstractmethod
    def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    
    @abstractmethod
    def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    
    @abstractmethod
    def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    
    @abstractmethod
    def query_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...

class AsyncIOReadOnlyExecutor(ABC):
    """
    Abstract interface for asynchronous clients that can execute
    read-only queries only.
    """
    
    @abstractmethod
    async def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    
    @abstractmethod
    async def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    
    @abstractmethod
    async def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    
    @abstractmethod
    async def query_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    async def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    
    @abstractmethod
    async def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...

Usage Examples

DSN Connection

import edgedb

# Connect using DSN
client = edgedb.create_client("edgedb://user:pass@localhost:5656/mydb")

# Async version
async_client = edgedb.create_async_client("edgedb://user:pass@localhost:5656/mydb")

Individual Parameters

import edgedb

# Connect using individual parameters
client = edgedb.create_client(
    host="localhost",
    port=5656,
    user="edgedb",
    password="secret",
    database="mydb",
    tls_security="strict"
)

Connection Pooling

import edgedb

# Configure connection pool
client = edgedb.create_client(
    dsn="edgedb://localhost:5656/mydb",
    max_concurrency=20,  # Maximum 20 concurrent connections
    timeout=30           # 30 second connection timeout
)

Environment Variables

The client automatically reads from environment variables:

EDGEDB_HOST
EDGEDB_PORT
EDGEDB_USER
EDGEDB_PASSWORD
EDGEDB_DATABASE
EDGEDB_BRANCH
EDGEDB_DSN
EDGEDB_CREDENTIALS_FILE

import edgedb

# Uses environment variables automatically
client = edgedb.create_client()

Install with Tessl CLI

npx tessl i tessl/pypi-edgedb

docs

ai-integration.md

client-management.md

configuration-options.md

schema-introspection.md

transaction-management.md

tile.json

tessl/pypi-edgedb