Integrations¶

SQLAlchemy¶

Elefast can read sqlalchemy.MetaData objects to create your database schema. Just use the elefast.MetadataMigrator and pass it your metadata object:

from elefast import DatabaseServer, MetadataMigrator
from my_app.models import Base

def db_server() -> DatabaseServer:
    server = DatabaseServer(
        engine=getenv("TESTING_DB_URL"), 
        schema=MetadataMigrator(Base.metadata),
    )  
    await server.ensure_is_ready()
    return server

from sqlalchemy.orm import DeclarativeBase, Mapped, MappedAsDataclass, mapped_column

class Base(MappedAsDataclass, DeclarativeBase):
    pass


class Post(Base):
    __tablename__ = "posts"

    slug: Mapped[str] = mapped_column(primary_key=True)
    headline: Mapped[str]
    body: Mapped[str]

Alembic¶

If you want to build your schema using your migrations, use the AlembicMigrator and pass it the path to your alembic config file.

from pathlib import Path
from elefast import DatabaseServer
from elefast.extras.alembic import AlembicMigrator

def db_server() -> DatabaseServer:
    alembic_config = Path(__file__).parent.parent / "alembic.ini" # (1)!
    server = DatabaseServer(
        engine=getenv("TESTING_DB_URL"), 
        schema=AlembicMigrator(alembic_config),
    )  
    await server.ensure_is_ready()
    return server

Then adjust your env.py to allow re-using an existing connection. This is the trickiest part to document properly, given that everyone is free to make their own little tweaks to the default env.py generated by Alembic. So let's walk through adjustments

@contextmanager
def get_connection() -> Iterator[Connection]:
    connection = config.attributes.get("connection") 
    if isinstance(connection, Connection):# (1)!
        yield connection
    else:# (2)!
        connectable = engine_from_config(
            config.get_section(config.config_ini_section, {}), 
            prefix="sqlalchemy.", 
            poolclass=pool.NullPool,
        )
        yield connectable.connect()

1. This path is hit while testing and forwards the connection to the database created by fixtures via DatabaseServer

2. This branch contains the usual code from the default config

Then adjust run_migrations_online to use the new get_connection function. Delete the construction of the connectable, which we've moved to get_connection.

def run_migrations_online() -> None:
    with get_connection() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()
            return

Docker¶

pip install 'elefast[docker]'

Sticky Container¶

TODO: Compare to testcontainers As mentioned previously, you

Optimizations¶

• tmpfs
• io stuff
• turning optimizations off

FastAPI¶

We have an example project demonstrating the use of the async API. It also uses pytest_asyncio in its fixtures.

uv¶

The elefast-example-uv-monorepo example shows you how you can create a repo-local Pytest plugin in your uv workspace.