Connection Management

Using Edgy is designed to be straightforward, but understanding connection management is crucial for optimal performance and stability.

Edgy is built on SQLAlchemy Core, but it's an asynchronous implementation. This raises questions about its integration with popular frameworks like Esmerald, Starlette, or FastAPI.

Edgy is framework-agnostic, meaning it can be seamlessly integrated into any framework that supports lifecycle events.

Lifecycle Events

Lifecycle events are common in frameworks built on Starlette, such as Esmerald and FastAPI. Other frameworks may offer similar functionality through different mechanisms.

The most common lifecycle events include:

• on_startup
• on_shutdown
• lifespan

This document focuses on lifespan, which is widely used.

Hooking Database Connections into Your Application

Integrating database connections is as simple as incorporating them into your framework's lifecycle events.

For illustrative purposes, we'll use Esmerald. However, the principles apply to any framework.

Using ASGI integration:

from esmerald import Esmerald

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


app = models.asgi(
    Esmerald(
        routes=[...],
    )
)

# load settings
monkay.evaluate_settings(ignore_import_errors=False)
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(registry=registry, app=app))

Manual integration (applicable to all frameworks):

from contextlib import asynccontextmanager
from esmerald import Esmerald

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


@asynccontextmanager
async def lifespan(app: Esmerald):
    async with models:
        yield


app = Esmerald(
    routes=[...],
    lifespan=lifespan,
)
# now required
monkay.evaluate_settings()
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(app=app, registry=registry))

Using an asynchronous context manager:

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


async def main():
    # load settings
    monkay.evaluate_settings(ignore_import_errors=False)
    # monkey-patch so you can use edgy shell
    monkay.set_instance(Instance(registry=registry))
    async with models:
        ...

Once the connection is integrated into your application's lifecycle, you can use the ORM throughout your application. Failing to do so will result in performance warnings, as the databasez backend will be reinitialized for each operation.

You can also define additional database connections in the registry and switch between them.

Django Integration

Django doesn't natively support the lifespan protocol. Therefore, we provide a keyword parameter for manual handling.

from django.core.asgi import get_asgi_application


from edgy import Registry, Instance

models = Registry(database="sqlite:///db.sqlite", echo=True)


application = models.asgi(handle_lifespan=True)(get_asgi_application())

# load settings
monkay.evaluate_settings(ignore_import_errors=False)
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(registry=registry, app=app))

Manual Integration

The __aenter__ and __aexit__ methods can be called as connect and disconnect. However, using context managers is recommended for simpler error handling.

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


async def main():
    # load settings
    monkay.evaluate_settings(ignore_import_errors=False)
    # monkey-patch app so you can use edgy shell
    monkay.set_instance(Instance(app=app, registry=registry))
    await models.__aenter__()
    try:
        ...
    finally:
        await models.__aexit__()

This approach is suitable for integration via on_startup and on_shutdown.

from contextlib import asynccontextmanager
from esmerald import Esmerald

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


app = Esmerald(
    routes=[...],
    on_startup=[models.__aenter__],
    on_shutdown=[models.__aexit__],
)
# load settings
monkay.evaluate_settings(ignore_import_errors=False)
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(app=app, registry=registry))

DatabaseNotConnectedWarning

This warning appears when an unconnected Database object is used.

Despite the warning being non-fatal, you should establish proper connections as demonstrated above. Synchronous environments require additional care.

Note

Ensure that Database objects passed via using are connected. They are not guaranteed to be connected outside of extra.

Integration in Synchronous Environments

When the framework is synchronous and no asynchronous loop is active, we can use run_sync. It's necessary to create an asynchronous environment using the with_async_env method of the registry. Otherwise, you'll encounter performance issues and DatabaseNotConnectedWarning warnings. run_sync calls must occur within the scope of with_async_env. with_async_env is re-entrant and accepts an optional loop parameter.

from edgy import Registry, Instance, monkay

models = Registry(database="sqlite:///db.sqlite", echo=True)


# load settings
monkay.evaluate_settings(ignore_import_errors=False)
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(registry=registry))


def main():
    with models.with_async_env():
        edgy.run_sync(User.query.all())

To maintain the loop for performance reasons, you can wrap the server worker loop or, for single-threaded servers, the server loop that runs the application. Alternatively, you can keep the asyncio event loop alive, which is easier for synchronous-first frameworks like Flask. Here's an example that's multi-threading safe.

import asyncio
from contextvars import ContextVar
from edgy import Registry, Instance, monkay, run_sync

models = Registry(database="sqlite:///db.sqlite", echo=True)


# multithreading safe
event_loop = ContextVar("event_loop", default=None)


def handle_request():
    loop = event_loop.get()
    if loop is None:
        # eventloops die by default with the thread
        loop = asyncio.new_event_loop()
        event_loop.set(loop)
    with models.with_loop(loop):
        edgy.run_sync(User.query.all())


def get_application():
    app = ...
    # load settings
    monkay.evaluate_settings(ignore_import_errors=False)
    # monkey-patch app so you can use edgy shell
    monkay.set_instance(Instance(registry=registry, app=app))
    return app


app = get_application()

That was complicated, right? Let's unroll it in a simpler example with explicit loop cleanup.

import asyncio
from edgy import Registry, Instance, monkay, run_sync

models = Registry(database="sqlite:///db.sqlite", echo=True)

# load settings
monkay.evaluate_settings(ignore_import_errors=False)
# monkey-patch app so you can use edgy shell
monkay.set_instance(Instance(registry=registry))

loop = asyncio.new_event_loop()
with models.with_loop(loop):
    edgy.run_sync(User.query.all())

# uses the same loop
with models.with_loop(loop):
    edgy.run_sync(User.query.all())


loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()

Note: with_async_env internally calls __aenter__ and __aexit__. Therefore, the database is connected during the with scope of with_async_env. This means you can use run_sync and run commands in another loop (e.g., via asyncio.run). Everything works without raising DatabaseNotConnectedWarning. This is used, for example, in edgy shell.

run_sync Function

run_sync requires further explanation. It integrates with the asynchronous environment created by with_async_env and prefers checking for an active running loop (unless an explicit loop is provided). If an active loop is found, a subloop is created, which is only terminated when the found loop (or explicit loop) is garbage collected. If an idling loop is found, it's reused instead of creating a subloop.

What is a subloop?

A subloop is an event loop running in a separate thread. This allows multiple event loops to run concurrently. They are removed when the parent event loop is garbage collected.

However, given that event loops can be sticky, we additionally check if the old loop has stopped.

Querying Other Schemas

Edgy supports querying other schemas. Refer to the tenancy section for details.

Multiple Connections

The Edgy Registry accepts an extra parameter for defining named additional Database objects or strings. Including them here ensures they're connected and disconnected appropriately.

You can switch between them using using.

Migrate from flask-migrate

See Migrations for more informations.

Note

Check the tips and tricks and learn how to make your connections even cleaner.