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

Query Execution

Name: tessl/pypi-edgedb
Author: tessl

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

Capabilities

Query Methods

Execute EdgeQL queries with different expected result cardinalities and output formats.

def query(
    query: str,
    *args,
    **kwargs
) -> List[Any]:
    """
    Execute a query and return all results as a list.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    List of query results, empty list if no results
    
    Raises:
    - EdgeDBError: For query execution errors
    - QueryArgumentError: For invalid arguments
    """

def query_single(
    query: str,
    *args,
    **kwargs
) -> Optional[Any]:
    """
    Execute a query expecting at most one result.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    Single query result or None if no results
    
    Raises:
    - EdgeDBError: For query execution errors
    - ResultCardinalityMismatchError: If query returns more than one result
    - QueryArgumentError: For invalid arguments
    """

def query_required_single(
    query: str,
    *args,
    **kwargs
) -> Any:
    """
    Execute a query expecting exactly one result.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    Single query result
    
    Raises:
    - EdgeDBError: For query execution errors
    - NoDataError: If query returns no results
    - ResultCardinalityMismatchError: If query returns more than one result
    - QueryArgumentError: For invalid arguments
    """

def query_json(
    query: str,
    *args,
    **kwargs
) -> str:
    """
    Execute a query and return results as JSON string.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    JSON string containing query results
    
    Raises:
    - EdgeDBError: For query execution errors
    - QueryArgumentError: For invalid arguments
    """

def query_single_json(
    query: str,
    *args,
    **kwargs
) -> str:
    """
    Execute a query expecting at most one result as JSON.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    JSON string containing single result or null
    
    Raises:
    - EdgeDBError: For query execution errors
    - ResultCardinalityMismatchError: If query returns more than one result
    - QueryArgumentError: For invalid arguments
    """

def query_required_single_json(
    query: str,
    *args,
    **kwargs
) -> str:
    """
    Execute a query expecting exactly one result as JSON.
    
    Parameters:
    - query: EdgeQL query string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    JSON string containing single result
    
    Raises:
    - EdgeDBError: For query execution errors
    - NoDataError: If query returns no results
    - ResultCardinalityMismatchError: If query returns more than one result
    - QueryArgumentError: For invalid arguments
    """

def execute(
    query: str,
    *args,
    **kwargs
) -> None:
    """
    Execute a command that doesn't return data.
    
    Used for INSERT, UPDATE, DELETE, DDL commands, etc.
    
    Parameters:
    - query: EdgeQL command string
    - *args: Positional query arguments
    - **kwargs: Named query arguments
    
    Returns:
    None
    
    Raises:
    - EdgeDBError: For command execution errors
    - QueryArgumentError: For invalid arguments
    """

Query Context and Options

Query execution context and options for controlling query behavior.

class QueryContext(NamedTuple):
    """Context for query execution."""
    query: str
    cache: QueryCache
    query_options: Optional[Any]
    retry_options: Optional[RetryOptions]
    state: Optional[State]
    warning_handler: Optional[WarningHandler]

class ExecuteContext(NamedTuple):
    """Context for command execution."""
    query: str
    cache: QueryCache
    state: Optional[State]
    warning_handler: Optional[WarningHandler]

class QueryWithArgs(NamedTuple):
    """Query with its arguments."""
    query: str
    args: Tuple
    kwargs: Dict[str, Any]

class QueryCache(NamedTuple):
    """Query cache entry for prepared statements."""
    # Implementation details vary

Usage Examples

Basic Queries

import edgedb

client = edgedb.create_client()

# Query multiple results
users = client.query("SELECT User { name, email }")
print(f"Found {len(users)} users")

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

# Query single result (required)
admin = client.query_required_single("SELECT User { name } FILTER .role = 'admin'")
print(f"Admin: {admin.name}")

Parameterized Queries

import edgedb

client = edgedb.create_client()

# Using positional arguments
users = client.query(
    "SELECT User { name, email } FILTER .age > $0 AND .city = $1",
    25, "New York"
)

# Using named arguments
users = client.query(
    "SELECT User { name, email } FILTER .age > $min_age AND .city = $city",
    min_age=25,
    city="New York"
)

# Mixed arguments
user = client.query_single(
    "SELECT User { name, email } FILTER .id = $0 AND .active = $active",
    "123e4567-e89b-12d3-a456-426614174000",
    active=True
)

JSON Queries

import edgedb
import json

client = edgedb.create_client()

# Get results as JSON string
users_json = client.query_json("SELECT User { name, email, created_at }")
users_data = json.loads(users_json)

# Single result as JSON
user_json = client.query_single_json(
    "SELECT User { name, email, profile } FILTER .id = <uuid>$id",
    id="123e4567-e89b-12d3-a456-426614174000"
)
if user_json != "null":
    user_data = json.loads(user_json)

Execute Commands

import edgedb

client = edgedb.create_client()

# Insert data
client.execute(
    "INSERT User { name := $name, email := $email, age := $age }",
    name="John Doe",
    email="john@example.com", 
    age=30
)

# Update data
client.execute(
    "UPDATE User FILTER .email = $email SET { age := $new_age }",
    email="john@example.com",
    new_age=31
)

# Delete data
client.execute(
    "DELETE User FILTER .email = $email",
    email="john@example.com"
)

# DDL commands
client.execute("""
    CREATE TYPE Article {
        title: str;
        content: str;
        author: User;
        created_at: datetime {
            default := datetime_current();
        };
    }
""")

Async Query Execution

import asyncio
import edgedb

async def main():
    client = edgedb.create_async_client()
    
    # All query methods have async versions
    users = await client.query("SELECT User { name, email }")
    user = await client.query_single("SELECT User FILTER .id = <uuid>$id", 
                                   id="123e4567-e89b-12d3-a456-426614174000")
    
    await client.execute(
        "INSERT User { name := $name, email := $email }",
        name="Alice Smith",
        email="alice@example.com"
    )
    
    await client.aclose()

asyncio.run(main())

Complex Queries

import edgedb

client = edgedb.create_client()

# Complex query with nested selections
articles = client.query("""
    SELECT Article {
        title,
        content,
        author: { name, email },
        tags: { name },
        comment_count := count(.comments),
        created_at
    }
    FILTER .published = true
    ORDER BY .created_at DESC
    LIMIT 10
""")

# Query with computed properties
user_stats = client.query("""
    SELECT User {
        name,
        email,
        article_count := count(.articles),
        latest_article := (
            SELECT .articles {
                title,
                created_at
            }
            ORDER BY .created_at DESC
            LIMIT 1
        ),
        total_comments := sum(.articles.comment_count)
    }
    FILTER .active = true
""")

Error Handling

import edgedb

client = edgedb.create_client()

try:
    user = client.query_required_single(
        "SELECT User { name } FILTER .email = $email",
        email="nonexistent@example.com"
    )
except edgedb.NoDataError:
    print("User not found")
except edgedb.ResultCardinalityMismatchError:
    print("Multiple users found with same email")
except edgedb.QueryError as e:
    print(f"Query error: {e}")
except edgedb.EdgeDBError as e:
    print(f"Database error: {e}")

Install with Tessl CLI