edgedb-python has two APIs: blocking and asynchronous. Both are almost entirely equivalent, with the exception of pool functionality, which is currently only supported in asynchronous mode.
The interaction with the database normally starts with a call to
connect(), or
async_connect(),
which establishes a new database session and returns a new
BlockingIOConnection
or AsyncIOConnection instance
correspondingly. The connection instance provides methods to run queries
and manage transactions.
Blocking connection example:
import datetime
import edgedb
def main():
# Establish a connection to an existing database
# named "test" as an "edgedb" user.
conn = edgedb.connect(
'edgedb://edgedb@localhost/test')
# Create a User object type
conn.execute('''
CREATE TYPE User {
CREATE REQUIRED PROPERTY name -> str;
CREATE PROPERTY dob -> cal::local_date;
}
''')
# Insert a new User object
conn.query('''
INSERT User {
name := <str>$name,
dob := <cal::local_date>$dob
}
''', name='Bob', dob=datetime.date(1984, 3, 1))
# Select User objects.
user_set = conn.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)}}
print(user_set)
# Close the connection.
conn.close()
if __name__ == '__main__':
main()An equivalent example using the asyncio API:
import asyncio
import datetime
import edgedb
async def main():
# Establish a connection to an existing database
# named "test" as an "edgedb" user.
conn = await edgedb.async_connect(
'edgedb://edgedb@localhost/test')
# Create a User object type
await conn.execute('''
CREATE TYPE User {
CREATE REQUIRED PROPERTY name -> str;
CREATE PROPERTY dob -> cal::local_date;
}
''')
# Insert a new User object
await conn.query('''
INSERT User {
name := <str>$name,
dob := <cal::local_date>$dob
}
''', name='Bob', dob=datetime.date(1984, 3, 1))
# Select User objects.
user_set = await conn.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)}}
print(user_set)
# Close the connection.
await conn.aclose()
if __name__ == '__main__':
asyncio.run(main())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. The edgedb-python asyncio API provides an implementation of such a pool.
To create a connection pool, use the
edgedb.create_async_pool()
function. The resulting AsyncIOPool
object can then be used to borrow connections from the pool.
Below is an example of a connection pool usage:
import asyncio
import edgedb
from aiohttp import web
async def handle(request):
"""Handle incoming requests."""
pool = request.app['pool']
username = int(request.match_info.get('name'))
# Execute the query on any pool connection
result = await pool.query_one_json(
'''
SELECT User {first_name, email, bio}
FILTER .name = <str>$username
''', username=username)
return web.Response(
text=result,
content_type='application/json')
async def init_app():
"""Initialize the application server."""
app = web.Application()
# Create a database connection pool
app['pool'] = await edgedb.create_async_pool(
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 = loop.run_until_complete(init_app())
web.run_app(app)You can also acquire connection from the pool:
async with pool.acquire() as conn:
result = await conn.query_one_json(
'''
SELECT User {first_name, email, bio}
FILTER .name = <str>$username
''', username=username)But if you have a bunch of tightly related queries it’s better to use transactions.
See Connection Pools API documentation for more information.
The most robust way to create a
transaction is retry method:
Example:
for tx in connection.retrying_transaction():
with tx:
tx.execute("INSERT User {name := 'Don'}")or, if using the async API on connection pool:
async for tx in connection.retrying_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.
See Transactions API documentation for more information.
