Light
Dark
System

Basic Usage

To start using EdgeDB in Python, create an edgedb.Client instance using edgedb.create_client():

Copy
import datetime
import edgedb

client = edgedb.create_client()

client.query("""
    INSERT User {
        name := <str>$name,
        dob := <cal::local_date>$dob
    }
""", name="Bob", dob=datetime.date(1984, 3, 1))

user_set = client.query(
    "SELECT User {name, dob} FILTER .name = <str>$name", name="Bob")
# *user_set* now contains
# Set{Object{name := 'Bob', dob := datetime.date(1984, 3, 1)}}

client.close()

When used with asyncio, this should be replaced with edgedb.create_async_client() which creates an instance of the AsyncIOClient:

Copy
import asyncio
import datetime
import edgedb

client = edgedb.create_async_client()

async def main():
    await client.query("""
        INSERT User {
            name := <str>$name,
            dob := <cal::local_date>$dob
        }
    """, name="Bob", dob=datetime.date(1984, 3, 1))

    user_set = await client.query(
        "SELECT User {name, dob} FILTER .name = <str>$name", name="Bob")
    # *user_set* now contains
    # Set{Object{name := 'Bob', dob := datetime.date(1984, 3, 1)}}

    await client.aclose()

asyncio.run(main())

The examples above only work under an EdgeDB project. You could also provide your own connection parameters, refer to the Client Library Connection docs for details.

edgedb-python automatically converts EdgeDB types to the corresponding Python types and vice versa. See Datatypes for details.

For server-type type applications that handle frequent requests and need the database connection for a short period time while handling a request, the use of a connection pool is recommended. Both edgedb.Client and edgedb.AsyncIOClient come with such a pool.

For edgedb.Client, all methods are thread-safe. You can share the same client instance safely across multiple threads, and run queries concurrently. Likewise, AsyncIOClient is designed to be shared among different asyncio.Task/coroutines for concurrency.

Below is an example of a web API server running aiohttp:

Copy
import asyncio
import edgedb
from aiohttp import web


async def handle(request):
    """Handle incoming requests."""
    client = request.app['client']
    username = int(request.match_info.get('name'))

    # Execute the query on any pool connection
    result = await client.query_single_json(
        '''
            SELECT User {first_name, email, bio}
            FILTER .name = <str>$username
        ''', username=username)
    return web.Response(
        text=result,
        content_type='application/json')


def init_app():
    """Initialize the application server."""
    app = web.Application()
    # Create a database client
    app['client'] = edgedb.create_async_client(
        database='my_service',
        user='my_service')
    # Configure service routes
    app.router.add_route('GET', '/user/{name:\w+}', handle)
    return app


loop = asyncio.get_event_loop()
app = init_app()
web.run_app(app)

Note that the client is created synchronously. Pool connections are created lazily as they are needed. If you want to explicitly connect to the database in init_app(), use the ensure_connected() method on the client.

For more information, see API documentation of the blocking client and the asynchronous client.

The most robust way to create a transaction is the transaction() method:

Example:

Copy
for tx in client.transaction():
    with tx:
        tx.execute("INSERT User {name := 'Don'}")

or, if using the async API:

Copy
async for tx in client.transaction():
    async with tx:
        await tx.execute("INSERT User {name := 'Don'}")

When not in an explicit transaction block, any changes to the database will be applied immediately.

For more information, see API documentation of transactions for the blocking client and the asynchronous client.

Light
Dark
System