or run

npx @tessl/cli init

Version

Tile

Overview

Evals

Files

docs

ai-integration.md client-management.md configuration-options.md data-types.md error-handling.md index.md query-execution.md schema-introspection.md transaction-management.md

tile.json

tessl/pypi-edgedb

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

Workspace: tessl
Visibility: Public
Created: 3 months ago
Last updated: 3 months ago
Describes: pkg:pypi/edgedb@2.2.x

To install, run

npx @tessl/cli install tessl/pypi-edgedb@2.2.0

EdgeDB Python Driver

The official Python driver for EdgeDB, a next-generation graph-relational database. Provides both blocking IO and asyncio implementations for connecting to and interacting with EdgeDB databases, enabling developers to execute EdgeQL queries, manage database connections, handle transactions, and work with EdgeDB's advanced type system.

Package Information

Package Name: edgedb
Language: Python
Installation: pip install edgedb
Python Version: >= 3.8

Core Imports

import edgedb

Common client imports:

from edgedb import create_client, create_async_client
from edgedb import Client, AsyncIOClient

Basic Usage

Synchronous Client

import edgedb

# Create a client
client = edgedb.create_client()

# Query for multiple results
users = client.query(
    "SELECT User { name, email } FILTER .name ILIKE $pattern",
    pattern="John%"
)

# Query for a single result
user = client.query_single(
    "SELECT User { name, email } FILTER .id = <uuid>$id",
    id="123e4567-e89b-12d3-a456-426614174000"
)

# Execute a command
client.execute(
    "INSERT User { name := $name, email := $email }",
    name="John Doe",
    email="john@example.com"
)

# Use transactions
with client.transaction() as tx:
    tx.execute("INSERT User { name := 'Alice' }")
    tx.execute("INSERT User { name := 'Bob' }")
    # Automatically commits or rolls back

client.close()

Asynchronous Client

import asyncio
import edgedb

async def main():
    # Create an async client
    client = edgedb.create_async_client()
    
    # Query for multiple results
    users = await client.query(
        "SELECT User { name, email } FILTER .name ILIKE $pattern",
        pattern="John%"
    )
    
    # Query for a single result
    user = await client.query_single(
        "SELECT User { name, email } FILTER .id = <uuid>$id",
        id="123e4567-e89b-12d3-a456-426614174000"
    )
    
    # Execute a command
    await client.execute(
        "INSERT User { name := $name, email := $email }",
        name="John Doe",
        email="john@example.com"
    )
    
    # Use transactions
    async with client.transaction() as tx:
        await tx.execute("INSERT User { name := 'Alice' }")
        await tx.execute("INSERT User { name := 'Bob' }")
        # Automatically commits or rolls back
    
    await client.aclose()

asyncio.run(main())

Architecture

The EdgeDB Python driver uses a layered architecture:

Client Layer: High-level Client and AsyncIOClient classes providing lazy connection pooling
Executor Layer: Abstract interfaces (Executor, AsyncIOExecutor) defining query execution contracts
Connection Layer: Low-level connection management with automatic retry and connection pooling
Protocol Layer: Binary protocol implementation for efficient communication with EdgeDB
Type System: Complete EdgeDB type mapping with Python-native representations
Error Handling: Comprehensive exception hierarchy mapping EdgeDB server errors

Capabilities

Client Creation and Management

Client factory functions and connection management for both synchronous and asynchronous programming models.

def create_client(
    dsn: Optional[str] = None,
    *,
    max_concurrency: Optional[int] = None,
    host: Optional[str] = None,
    port: Optional[int] = None,
    credentials: Optional[Credentials] = None,
    # ... additional connection parameters
) -> Client: ...

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,
    # ... additional connection parameters
) -> AsyncIOClient: ...

Client Management

Query Execution

Core query execution methods for EdgeQL queries with support for different result cardinalities and JSON output formats.

def query(query: str, *args, **kwargs) -> List[Any]: ...
def query_single(query: str, *args, **kwargs) -> Optional[Any]: ...
def query_required_single(query: str, *args, **kwargs) -> Any: ...
def query_json(query: str, *args, **kwargs) -> str: ...
def execute(query: str, *args, **kwargs) -> None: ...

Query Execution

Data Types

EdgeDB data type representations in Python, including scalar types, collections, and custom types.

class Object: ...
class Set(list): ...
class Array(list): ...
class NamedTuple(tuple): ...
class EnumValue: ...
class RelativeDuration: ...
class Range: ...
class MultiRange: ...

Data Types

Transaction Management

Transaction handling with context managers, retry logic, and isolation level control.

class TransactionOptions:
    def __init__(
        self,
        isolation: IsolationLevel = IsolationLevel.Serializable,
        readonly: bool = False,
        deferrable: bool = False
    ): ...

def transaction(
    *,
    options: Optional[TransactionOptions] = None,
    retry_options: Optional[RetryOptions] = None
) -> ContextManager: ...

Transaction Management

Configuration and Options

Client configuration options, retry policies, session state management, and connection parameters.

class RetryOptions:
    def __init__(self, attempts: int, backoff: Callable[[int], float]): ...

class State:
    def with_default_module(self, module: str) -> State: ...
    def with_module_aliases(self, **aliases: str) -> State: ...
    def with_config(self, **config: Any) -> State: ...
    def with_globals(self, **globals: Any) -> State: ...

Configuration and Options

Error Handling

Comprehensive exception hierarchy representing all EdgeDB server errors and client-side issues.

class EdgeDBError(Exception):
    def has_tag(self, tag: ErrorTag) -> bool: ...
    def get_code(self) -> int: ...

class QueryError(EdgeDBError): ...
class TransactionError(EdgeDBError): ...
class ClientError(EdgeDBError): ...

Error Handling

Schema Introspection

Schema introspection capabilities for examining database structure and type information.

class DescribeContext:
    query: str
    state: Optional[State]
    inject_type_names: bool
    output_format: OutputFormat
    expect_one: bool

class DescribeResult:
    input_type: Optional[AnyType]
    output_type: Optional[AnyType]
    output_cardinality: Cardinality
    capabilities: int

Schema Introspection

AI Integration

EdgeDB AI capabilities for RAG (retrieval-augmented generation) queries and embeddings generation.

def create_ai(client: Client, **kwargs) -> EdgeDBAI: ...
def create_async_ai(client: AsyncIOClient, **kwargs) -> AsyncEdgeDBAI: ...

class EdgeDBAI:
    def query_rag(self, message: str, context: Optional[QueryContext] = None) -> str: ...
    def stream_rag(self, message: str, context: Optional[QueryContext] = None) -> Iterator[str]: ...
    def generate_embeddings(self, *inputs: str, model: str) -> List[float]: ...

AI Integration

Types

# Core data types
Tuple = tuple
NamedTuple = collections.namedtuple
Set = list
Array = list
EnumValue = object
RelativeDuration = object
DateDuration = object
ConfigMemory = object
Object = object
Link = object
LinkSet = object
Range = object
MultiRange = object

# Client types
class Client:
    def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    def query_json(self, query: str, *args, **kwargs) -> str: ...
    def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...
    def execute(self, query: str, *args, **kwargs) -> None: ...
    def transaction(self, **kwargs) -> ContextManager: ...
    def ensure_connected(self) -> None: ...
    def close(self) -> None: ...

class AsyncIOClient:
    async def query(self, query: str, *args, **kwargs) -> List[Any]: ...
    async def query_single(self, query: str, *args, **kwargs) -> Optional[Any]: ...
    async def query_required_single(self, query: str, *args, **kwargs) -> Any: ...
    async def query_json(self, query: str, *args, **kwargs) -> str: ...
    async def query_single_json(self, query: str, *args, **kwargs) -> str: ...
    async def query_required_single_json(self, query: str, *args, **kwargs) -> str: ...
    async def execute(self, query: str, *args, **kwargs) -> None: ...
    def transaction(self, **kwargs) -> AsyncContextManager: ...
    async def ensure_connected(self) -> None: ...
    async def aclose(self) -> None: ...

# Enums
class Cardinality(Enum):
    NO_RESULT = "n"
    AT_MOST_ONE = "?"
    ONE = "1"
    MANY = "*"
    AT_LEAST_ONE = "+"
    
    def is_single(self) -> bool: ...
    def is_multi(self) -> bool: ...

class IsolationLevel:
    Serializable = "SERIALIZABLE"

class ElementKind(Enum):
    LINK = "link"
    PROPERTY = "property"
    LINK_PROPERTY = "linkprop"

Version

Tile

Files

tessl/pypi-edgedb

To install, run

index.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

EdgeDB Python Driver

Package Information

Core Imports

Basic Usage

Synchronous Client

Asynchronous Client

Architecture

Capabilities

Client Creation and Management

Query Execution

Data Types

Transaction Management

Configuration and Options

Error Handling

Schema Introspection

AI Integration

Types

index.mddocs/