Skip to main content

SQLAlchemy Core integration for Flask

Project description

SQLAlchemy Core integration for Flask application.

Provides an easy way for setting up and using SQLAlchemy Core in Flask (or Werkzeug-based, really) applications.

The main benefit of using this package over plain SQLAlchemy is the ability of sharing the current connection in a context-local “global” object.

This allows creating nested transactions without having to explicitly pass the connection object across functions.

This in turn enables running test functions inside a transaction, which will be rolled back on test teardown, greatly reducing test run time, by removing the need of continuously dropping and re-creating the whole database schema for each test function run.


Set up

import os
from flask_sqlalchemy_core import FlaskSQLAlchemy


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...


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
        # Other data for the user in some other table

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

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

Define tables

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

Creating schema


Test fixtures

For use with pytest, place those in a 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

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.

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.




Database migrations

Use Alembic for creating database migrations.

Database support

The library is currently tested with PostgreSQL (10).

Everything should work with other backends too (except nested transactions, on backends that don’t support checkpoints).


Before running the test suite, you’ll need to start a SQL database and set the DATABASE_URL environment variable.

For convenience, you can use the bin/run-test-database script, which will automatically run a PostgreSQL instance via Docker.

The script will print a suitable value for DATABASE_URL as well.

Oncer you’re done, simply kill it via Ctrl-C.

To install test dependencies:

pip install -r test_requirements.txt

To run the test suite:

pytest -vvv ./tests

To run tests using SQLite backend:

DATABASE_URL="sqlite:///:memory:" pytest -vvv ./tests

Warning: some tests will be skipped, as SQLite doesn’t support nested transactions.

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

flask_sqlalchemy_core-0.1.1.tar.gz (5.6 kB view hashes)

Uploaded source

Built Distribution

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page