flask-sqlalchemy-core 0.1

Set up

import os
from flask_sqlalchemy_core import FlaskSQLAlchemy

DATABASE_URL = os.environ['DATABASE_URL']

db = FlaskSQLAlchemy(DATABASE_URL)

Running queries

from sqlalchemy import select

# Create your query here
query = select(...)

with db.connect() as conn:
    result = conn.execute(query)
    # Do something with the result...

Transactions

with db.transaction() as conn:
    result = conn.execute(query)

The transaction will automatically committed upon a successful exit from the “with” block, or rolled back if an exception was raised.

Nested transactions

Simply nest with db.transaction(): blocks.

This allows for more reusable code, for example:

def create_user(...):
    with db.transaction() as conn:
        # Create record in the users table
        conn.execute(...)
        # Other data for the user in some other table
        conn.execute(...)

def create_client(...):
    with db.transaction() as conn:
        # Create record in the clients table
        conn.execute(...)
        # ...other data for this client...
        conn.execute(...)

def setup_new_client(client_name, user_name):
    with db.transaction():
        create_user(user_name)
        create_client(client_name)

Define tables

Just do as you normally would (create a Metadata instance, use it to define your schema).

Creating schema

metadata.create_all(db.get_engine())

Test fixtures

For use with pytest, place those in a conftest.py file in your tests directory.

Note: you might want to change your DATABASE_URL environment variable during testing, to avoid overriding your current development database.

import pytest

@pytest.fixture
def db(db_schema):
    with db.transaction(autocommit=False, rollback=True):
        # By wrapping execution in a transaction that automatically
        # gets rolled back, we can avoid having to recreate the whole
        # schema for every test function run.
        yield

@pytest.fixture(scope='session')
def db_schema():
    engine = db.get_engine()

    # Clean up, in case tables were left around from a previous run.
    # This can happen if the test process was abruptly killed.
    metadata.drop_all(engine)

    metadata.create_all(engine)

    yield

    metadata.drop_all(engine)

Database migrations

Use Alembic for creating database migrations.