SQLAlchemy Helpers

This project contains a tools to use SQLAlchemy and Alembic in a project.

It has a Flask integration, and other framework integrations could be added in the future.

The full documentation is on ReadTheDocs.

You can install it from PyPI.

Features

Here’s what sqlalchemy-helpers provides:

• Alembic integration:

  • programmatically create or upgrade your schema,

  • get information about schema versions and status

  • drop your tables without leaving alembic information behind

  • use a function in your env.py script to retrieve the database URL, and thus avoid repeating your configuration in two places.

  • migration helper functions such as is_sqlite() or exists_in_db()

• SQLAlchemy naming convention for easier schema upgrades

• Automatically activate foreign keys on SQLite

• Addition of some useful query properties on your models

• A query function get_or_create() that you can call directly or use on your model classes

• Optional Flask integration: you can use sqlalchemy-helpers outside of a Flask app and feel at home

• The models created with sqlalchemy-helpers work both inside and outside the Flask application context

• Support for asyncio and FastAPI.

This project has 100% code coverage and aims at reliably sharing some of the basic boilerplate between applications that use SQLAlchemy.

Check out the User Guide to learn how to use it in your application, with or without a web framework.

FAQ

• Why not use Flask-SQLAlchemy and Flask-Migrate?

Those projects are great, but we also have apps that are not based on Flask and that would benefit from the features provided by sqlalchemy-helpers.

User Guide

Standalone

Even without a framework extension, sqlalchemy-helpers brings many interesting features. Here’s how you can use it.

Writing models

All the models must inherit from the sqlalchemy_helpers.manager.Base class. It is equivalent to SQLAlchemy’s declarative_base() with a constraint naming convention and some extra features.

Example:

from sqlalchemy import Column, Integer, Unicode
from sqlalchemy_helpers import Base

class User(Base):

    __tablename__ = "users"

    id = Column("id", Integer, primary_key=True)
    name = Column(Unicode(254), index=True, unique=True, nullable=False)
    full_name = Column(Unicode(254), nullable=False)
    timezone = Column(Unicode(127), nullable=True)

As you can see, it is very similar to what you would do with plain SQLAlchemy.

If you need to, you can have a different base class for your models, using the sqlalchemy_helpers.manager.get_base() function:

from sqlalchemy_helpers import get_base

Base = get_base(cls=CustomBase)

All the arguments passed to the get_base() function will be transferred to the sqlalchemy.orm.declarative_base() function.

The database manager

Most of the integration work in sqlalchemy-helpers is done via the sqlalchemy_helpers.manager.DatabaseManager. It can be instanciated with:

from sqlalchemy_helpers import DatabaseManager
db = DatabaseManager("sqlite:///", "path/to/alembic")

The first argument is the database URI, the second argument is the path to the alembic directory is where Alembic’s env.py resides. The third argument is a dictionary of additional keyword arguments that will be passed to the create_engine factory along with the URI. The fourth argument is the custom base class, if you have defined any (it is optional).

You can call the Database Manager’s functions to get information about your database or to migrate its schema.

Making queries

The Database Manager has a Session property mapping to SQLAlchemy’s Session factory, scoped for multithreading use. Get a session by calling:

session = db.Session()
user = session.query(User).filter_by(name="foo").one()

This library also provides a get_or_create() function, as popularized by Django:

from sqlalchemy_helpers import get_or_create

user, created = get_or_create(User, name="foo")

For convenience, this function is also available as a model method:

user, created = User.get_or_create(name="foo")

Other useful model methods are:

user = User.get_one(name="foo")
user = User.get_by_pk(42)

Migrations

The manager can create and update your database. It also has a sync() method that will create the database if it does not exist or update it if it is not at the latest schema revision. The sync() call will return the result of the operation as a member of the SyncResult enum so you can react accordingly.

You can also find a couple helper functions for your migrations: is_sqlite() and exists_in_db().

Flask integration

This project provides a Flask integration layer for Flask >= 2.0.0. This is how you can use it.

Base setup

First, create a python module to instanciate the DatabaseExtension, and re-export some useful helpers:

# database.py

from sqlalchemy_helpers import Base, get_or_create, is_sqlite, exists_in_db
from sqlalchemy_helpers.flask_ext import DatabaseExtension, get_or_404, first_or_404

db = DatabaseExtension()

In the application factory, import the instance and call its init_app() method:

# app.py

from flask import Flask
from .database import db

def create_app():
    """See https://flask.palletsprojects.com/en/1.1.x/patterns/appfactories/"""

    app = Flask(__name__)

    # Load the optional configuration file
    if "FLASK_CONFIG" in os.environ:
        app.config.from_envvar("FLASK_CONFIG")

    # Database
    db.init_app(app)

    return app

If you need to define a custom base class, you can pass it to the extension using the base_model argument of the __init__() constructor or the init_app() function.

Models

You can declare your models as you usually would with SQLAlchemy, just inherit from the Base class that you re-exported in database.py:

# models.py

from sqlalchemy import Column, Integer, Unicode

from .database import Base


class User(Base):

    __tablename__ = "users"

    id = Column("id", Integer, primary_key=True)
    name = Column(Unicode(254), index=True, unique=True, nullable=False)
    full_name = Column(Unicode(254), nullable=False)
    timezone = Column(Unicode(127), nullable=True)

Note: these models do not depend on the Flask extension, only the main part of sqlalchemy-helpers. They will import and work just fine without Flask.

Also note that if you want to move your models away from sqlalchemy-helpers and back to plain SQLAlchemy, all you have to do is replace the Base import with:

from sqlalchemy.orm import declarative_base

Base = decalarative_base()

Views

Now in your views, you can use the instance’s session property to access the SQLAlchemy session object. There are also functions to ease classical view patterns such as getting an object by ID or returning a 404 error if not found:

# views.py

from .database import db, get_or_404
from .models import User


@bp.route("/")
def root():
    users = db.session.query(User).all()
    return render_template("index.html", users=users)


@bp.route("/user/<int:user_id>")
def profile(user_id):
    user = get_or_404(User, user_id)
    return render_template("profile.html", user=user)

Migrations

You can adjust alembic’s env.py file to get the database URL from your app’s configuration:

# migrations/env.py

from my_flask_app.app import create_app
from my_flask_app.database import Base
from sqlalchemy_helpers.flask_ext import get_url_from_app

url = get_url_from_app(create_app)
config.set_main_option("sqlalchemy.url", url)
target_metadata = Base.metadata

# ...rest of the env.py file...

Also set script_location in you alembic.ini file in order to use it with the alembic command-line tool:

# migrations/alembic.ini

[alembic]
script_location = %(here)s

Features summary

And that’s it! You’ll gain the following features:

• a per-request session you can use with db.session

• recursive auto-import of your models

• a db subcommand to sync your models: just run flask db sync

• two view utility functions: get_or_404() and first_or_404(), which let you query the database and return 404 errors if the expected record is not found

• the alembic command is still functional as documented upstream by pointing at the alembic.ini file

Full example

In Fedora Infrastructure we use a cookiecutter template that showcases this Flask integration, feel free to check it out or even use it if it suits your needs.

Openshift health checks

Being able to programmatically know whether the database schema is up-to-date is very useful when working with cloud services that check that your application is actually available, such as OpenShift/Kubernetes. If you’re using flask-healthz you can write a pretty clever readiness function such as:

from flask_healthz import HealthError
from sqlalchemy_helpers import DatabaseStatus
from .database import db

def liveness():
    pass

def readiness():
    try:
        status = db.manager.get_status()
    except Exception as e:
        raise HealthError(f"Can't get the database status: {e}")
    if status is DatabaseStatus.NO_INFO:
        raise HealthError("Can't connect to the database")
    if status is DatabaseStatus.UPGRADE_AVAILABLE:
        raise HealthError("The database schema needs to be updated")

With this function, OpenShift will not forward requests to the updated version of your application if there are pending schema changes, and will keep serving from the old version until you’ve applied the database migration.

Asynchronous connections

The sqlalchemy-helpers library supports AsyncIO connections in SQLAlchemy and Alembic.

All the models must still inherit from the sqlalchemy_helpers.manager.Base class. As a convenience, it is also exported in the sqlalchemy_helpers.aio module.

Usage and differences

The database manager

The async-enabled database manager is sqlalchemy_helpers.aio.AsyncDatabaseManager. It can be instanciated with:

from sqlalchemy_helpers.aio import AsyncDatabaseManager
db = AsyncDatabaseManager("sqlite:///", "path/to/alembic")

The arguments are the same as the synchronous manager.

Making queries

Like the synchronous manager, the async manager has a Session property mapping to SQLAlchemy’s AsyncSession factory. Get a session by calling:

session = db.Session()
result = await session.execute(select(User).filter_by(name="foo"))
user = result.one()

The Database Manager will also setup async query methods on your models, similar to the synchronous versions, but a bit different because you need to provide the session as a first argument:

user = await User.get_one(session, name="foo")
# or
user = await User.get_by_pk(session, 42)
# or
user, created = await User.get_or_create(session, name="foo")

Alembic

You will need to modify Alembic’s env.py script slightly to make it support async operations. An example is provided in the docs/ directory here in the source code.

You can keep defining your database url with the sync drivers, such as sqlite, postgresql, etc. The database manager will automatically translate them to their async counterparts. As a consequence, you will still be able to use the alembic command with the sync drivers, as usual.

Migrations

The manager’s migration operations are async and will need to be awaited. Besides that, they work as their synchronous counterparts.

FastAPI integration

This project provides a few FastAPI integration functions.

Making a manager

The sqlalchemy_helpers.fastapi.manager_from_config() function will build a sqlalchemy_helpers.aio.AsyncDatabaseManager instance using Pydantic settings

It assumes a layout such as:

class SQLAlchemyModel(BaseModel):
    url: stricturl(tld_required=False, host_required=False) = "sqlite:///:memory:"

class AlembicModel(BaseModel):
    migrations_path: DirectoryPath = Path(__file__).parent.joinpath("migrations").absolute()

class Settings(BaseSettings):
    sqlalchemy: SQLAlchemyModel = SQLAlchemyModel()
    alembic: AlembicModel = AlembicModel()

You can, of course, pass a subset of the configuration to the function. It also understands plain dictionaries.

Sync CLI

A function wrapping the manager’s sync method is provided in sqlalchemy_helpers.fastapi.syncdb(). You can hook it up to your click-based CLI, it takes the Pydantic settings as only argument.

Base setup

The library provides functions that you can use as a dependencies in your FastAPI path operations. First, create a python module to integrate those functions with your Pydantic settings:

# database.py

from collections.abc import Iterator
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy_helpers.fastapi import AsyncDatabaseManager, make_db_session, manager_from_config
from sqlalchemy_helpers.aio import Base
from .config import get_settings
from . import models

async def gen_db_manager() -> AsyncDatabaseManager:
    db_settings = get_settings().database
    return manager_from_config(db_settings)

async def gen_db_session(
    db_manager: AsyncDatabaseManager = Depends(gen_db_manager),
) -> Iterator[AsyncSession]:
    async for session in make_db_session(db_manager):
        yield session

We also recommend re-exporting the sqlalchemy_helpers.aio.Base class for convenience and ease of refactoring.

In the main module, declare the application. This example uses routers for modularity:

# main.py

from fastapi import FastAPI
from .views import router

app = FastAPI()
app.include_router(router)

Models

You can declare your models as you usually would with SQLAlchemy, just inherit from the Base class that you re-exported in database.py:

# models.py

from sqlalchemy import Column, Integer, UnicodeText
from .database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(UnicodeText, nullable=False, unique=True)

Note: these models do not depend on the FastAPI extension, only the main part of sqlalchemy-helpers. They will import and work just fine without FastAPI.

Also note that if you want to move your models away from sqlalchemy-helpers and back to plain SQLAlchemy, all you have to do is replace the Base import with:

from sqlalchemy.orm import declarative_base

Base = decalarative_base()

Access in path operations

Now, you can use FastAPI’s dependency injection to get the database session in your path operations:

# views.py

from fastapi import APIRouter, Depends
from .database import gen_db_session
from .models import User

router = APIRouter(prefix="/users")

@router.get("/user/{name}")
async def get_user(name: str, db_session: AsyncSession = Depends(gen_db_session)):
    user = await User.get_one(db_session, name=name)
    return user

Migrations

You can adjust alembic’s env.py file to get the database URL from your app’s configuration:

# migrations/env.py

from my_fastapi_app.config import get_settings
from my_fastapi_app.database import Base

url = get_settings().database.sqlalchemy.url
config.set_main_option("sqlalchemy.url", url)
target_metadata = Base.metadata

# ...rest of the env.py file...

Also set script_location in you alembic.ini file in order to use it with the alembic command-line tool:

# migrations/alembic.ini

[alembic]
script_location = %(here)s

Contributing

Thanks for considering contributing to SQLAlchemy Helpers, we really appreciate it!

Quickstart:

1. Look for an existing issue about the bug or feature you’re interested in. If you can’t find an existing issue, create a new one.

2. Fork the repository on GitHub.

3. Fix the bug or add the feature, and then write one or more tests which show the bug is fixed or the feature works.

4. Submit a pull request and wait for a maintainer to review it.

More detailed guidelines to help ensure your submission goes smoothly are below.

Note

If you do not wish to use GitHub, please send patches to infrastructure@lists.fedoraproject.org.

Guidelines

Python Support

SQLAlchemy Helpers supports Python 3.8 or greater. This is automatically enforced by the continuous integration (CI) suite.

Code Style

We follow the PEP8 style guide for Python. This is automatically enforced by the CI suite.

We are using Black <https://github.com/ambv/black> to automatically format the source code. It is also checked in CI. The Black webpage contains instructions to configure your editor to run it on the files you edit.

Handle every possible case, and do so where it makes sense.

Security

Remember to keep the code simple enough that it can be easily reviewed for security concerns.

Code that touches security-critical paths must be signed off by two people. People who sign off are agreeing to have reviewed the code thoroughly and thought about edge cases.

Tests

The test suites can be run using tox by simply running tox from the repository root. All code must have test coverage or be explicitly marked as not covered using the #  pragma: no cover comment. This should only be done if there is a good reason to not write tests.

Your pull request should contain tests for your new feature or bug fix. If you’re not certain how to write tests, we will be happy to help you.

Release Notes

To add entries to the release notes, create a file in the news directory in the source.type name format, where the source part of the filename is:

• 42 when the change is described in issue 42

• PR42 when the change has been implemented in pull request 42, and there is no associated issue

• Cabcdef when the change has been implemented in changeset abcdef, and there is no associated issue or pull request.

And where the extension type is one of:

• bic: for backwards incompatible changes

• dependency: for dependency changes

• feature: for new features

• bug: for bug fixes

• dev: for development improvements

• docs: for documentation improvements

• other: for other changes

The content of the file will end up in the release notes. It should not end with a . (full stop).

If it is not present already, add a file in the news directory named username.author where username is the first part of your commit’s email address, and containing the name you want to be credited as. There is a script to generate a list of authors that we run before releasing, but creating the file manually allows you to set a custom name.

A preview of the release notes can be generated with towncrier build --draft.

Licensing

Your commit messages must include a Signed-off-by tag with your name and e-mail address, indicating that you agree to the Developer Certificate of Origin version 1.1:

Developer Certificate of Origin
Version 1.1

Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
1 Letterman Drive
Suite D4700
San Francisco, CA, 94129

Everyone is permitted to copy and distribute verbatim copies of this
license document, but changing it is not allowed.


Developer's Certificate of Origin 1.1

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.

Use git commit -s to add the Signed-off-by tag.

Releasing

When cutting a new release, follow these steps:

1. Update the version in pyproject.toml

2. Run poetry install to update the version in the metadata

3. Add missing authors to the release notes fragments by changing to the news directory and running the get-authors.py script, but check for duplicates and errors

4. Generate the release notes by running poetry run towncrier (in the base directory)

5. Adjust the release notes in docs/release_notes.rst.

6. Generate the docs with tox -e docs and check them in docs/_build/html.

7. Commit the changes

8. Push the commit to the upstream Github repository (via a PR or not).

9. Change to the stable branch and cherry-pick the commit (or merge if appropriate)

10. Run the checks one last time to be sure: tox,

11. Tag the commit with -s to generate a signed tag

12. Push the commit to the upstream Github repository with git push, and the new tag with git push --tags

13. Generate a tarball and push to PyPI with the command poetry publish --build

14. Create the release on GitHub and copy the release notes in there,

15. Announce.

Release notes

All notable changes to this project will be documented in this file.

The format is based on Keep a Changelog, and this project adheres to Semantic Versioning.

This project uses towncrier and the changes for the upcoming release can be found in https://github.com/fedora-infra/sqlalchemy-helpers/tree/develop/news/.

Version 0.13.0

Released on 2023-11-16. This is a feature release that adds customization of the model base class.

Features

• Allow customization of the model base class (bc26cd3).

Version 0.12.1

Released on 2023-10-12. This is a minor release that adds docs and development improvements

Development Improvements

• Automatically publish to PyPI and release (c572657).

Documentation Improvements

• Release notes: show dependency changes further down the page (499ec7a).

• Convert the release notes to Markdown (841e1fb).

Version 0.12.0

Released on 2023-08-09. This is a feature release that adds MySQL/MariaDB support in the async mode.

Features

• Add support for MySQL/MariaDB in the async mode (#325).

Bug Fixes

• The psycopg driver raises a ProgrammingError where sqlite raises an OperationalError (469d9c7).

Dependency Changes

• Fix a minor compatibility issue with SQLAlchemy 2.0 (3f379e2).

• Support Pydantic 2.0+ and Pydantic Settings (#323).

Version 0.11.0

Released on 2023-06-23. This is a major release that adds AsyncIO and FastAPI support.

Dependency Changes

• Drop the query_property as it is considered legacy by SQLAlchemy. Instead, add get_by_pk() and get_one() methods (2702667).

• Fix compatibility with Flask 2.3 and above (6040394).

Features

• Support for asyncio-based connections, and FastAPI integration (#317).

• Allow passing extra arguments to create_engine() and create_async_engine() (#319).

sqlalchemy_helpers package

SQLAlchemy Helpers

A set of tools to integrate SQLAlchemy and Alembic in your project, with sane defauts.

sqlalchemy_helpers.__version__

this package’s version.

Type:

str

Submodules

sqlalchemy_helpers.aio module

Database management (async).

This must remain independent from any web framework.

sqlalchemy_helpers.aio.Base

SQLAlchemy’s base class for models.

Type:

object

class sqlalchemy_helpers.aio.AsyncDatabaseManager(uri, alembic_location, engine_args=None, base_model=None)

Bases: DatabaseManager

Helper for a SQLAlchemy and Alembic-powered database, asynchronous version.

Parameters:

• uri (str) – the database URI

• alembic_location (str) – a path to the alembic directory

• engine_args (dict) – additional arguments passed to create_async_engine

alembic_cfg

the Alembic configuration object

Type:

alembic.config.Config

engine

the SQLAlchemy Engine instance

Type:

sqlalchemy.engine.Engine

Session

the SQLAlchemy scoped session factory

Type:

sqlalchemy.orm.scoped_session

configured_connection(f)

async create()

Create the database tables.

async drop()

Drop all the database tables.

async get_current_revision(session)

Get the current alembic database revision.

async get_status()

Get the status of the database.

Returns:

see DatabaseStatus.

Return type:

DatabaseStatus member

async sync()

Create or update the database schema.

Returns:

see SyncResult.

Return type:

SyncResult member

async upgrade(target='head')

Upgrade the database schema.

async sqlalchemy_helpers.aio.get_by_pk(pk, *, session, model)

Get a model instance using its primary key.

Example: user = get_by_pk(42, session=session, model=User)

async sqlalchemy_helpers.aio.get_one(session: AsyncSession, model, **attrs) → Base

Get an object from the datbase.

Parameters:

• session – The SQLAlchemy session to use

• model – The SQLAlchemy model to query

Returns:

the object

async sqlalchemy_helpers.aio.get_or_create(session, model, **attrs)

Function like Django’s get_or_create() method.

It will return a tuple, the first argument being the instance and the second being a boolean: True if the instance has been created and False otherwise.

Example: user, created = get_or_create(session, User, name="foo")

sqlalchemy_helpers.fastapi module

FastAPI integration of database management.

async sqlalchemy_helpers.fastapi.make_db_session(manager) → Iterator[AsyncSession]

Generate database sessions for FastAPI request handlers.

This lets users declare the session as a dependency in request handler functions, e.g.:

@app.get("/path")
def process_path(db_session: AsyncSession = Depends(make_db_session)):
    query = select(Model).filter_by(...)
    result = await db_session.execute(query)
    ...

Returns:

A sqlalchemy.ext.asyncio.AsyncSession object for the current request

sqlalchemy_helpers.fastapi.manager_from_config(db_settings, *args, **kwargs)

Get the database manager using the Flask app’s configuration.

async sqlalchemy_helpers.fastapi.syncdb(db_settings)

Run DatabaseManager.sync() on the command-line.

sqlalchemy_helpers.flask_ext module

Flask integration of database management.

class sqlalchemy_helpers.flask_ext.DatabaseExtension(app=None, base_model=None)

Bases: object

A Flask extension to configure the database manager according the the app’s configuration.

It cleans up database connections at the end of the requests, and creates the CLI endpoint to sync the database schema.

before_request()

Prepare the database manager at the start of each request.

This is necessary to allow access to the Model.get_* methods.

init_app(app, base_model=None)

Initialize the extention on the provided Flask app

Parameters:

app (flask.Flask) – the Flask application.

property manager

the instance of the database manager.

Type:

DatabaseManager

property session

the database Session instance to use.

Type:

sqlalchemy.session.Session

teardown(exception)

Close the database connection at the end of each requests.

sqlalchemy_helpers.flask_ext.first_or_404(query, description=None)

Like query.first but aborts with 404 if not found.

Parameters:

• query (sqlalchemy.orm.Query) – a query to retrieve.

• description (str, optional) – a message for the 404 error if no records are found.

sqlalchemy_helpers.flask_ext.get_or_404(Model, pk, description=None)

Like query.get but aborts with 404 if not found.

Parameters:

• Model (manager.Base) – a model class.

• pk (int or str) – the primary key of the desired record.

• description (str, optional) – a message for the 404 error if not found.

sqlalchemy_helpers.flask_ext.get_url_from_app(app_factory)

Get the DB URI from the app configuration

Create the application if it hasn’t been created yet. This is useful in Alembic’s env.py.

Args: app_factory (callable): the Flask application factory, to be called if this function is

called outside of and application context.

sqlalchemy_helpers.manager module

Database management.

This must remain independent from any web framework.

sqlalchemy_helpers.manager.Base

SQLAlchemy’s base class for models.

Type:

object

class sqlalchemy_helpers.manager.DatabaseManager(uri, alembic_location, engine_args=None, base_model=None)

Bases: object

Helper for a SQLAlchemy and Alembic-powered database

Parameters:

• uri (str) – the database URI

• alembic_location (str) – a path to the alembic directory

• engine_args (dict) – additional arguments passed to create_engine

alembic_cfg

the Alembic configuration object

Type:

alembic.config.Config

engine

the SQLAlchemy Engine instance

Type:

sqlalchemy.engine.Engine

Session

the SQLAlchemy scoped session factory

Type:

sqlalchemy.orm.scoped_session

create()

Create the database tables.

drop()

Drop all the database tables.

get_current_revision(session)

Get the current alembic database revision.

get_latest_revision()

Get the most up-to-date alembic database revision available.

get_status()

Get the status of the database.

Returns:

see DatabaseStatus.

Return type:

DatabaseStatus member

sync()

Create or update the database schema.

Returns:

see SyncResult.

Return type:

SyncResult member

upgrade(target='head')

Upgrade the database schema.

class sqlalchemy_helpers.manager.DatabaseStatus(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

The status of the database.

NO_INFO = 2

Returned when the database couldn’t be connected to.

UPGRADE_AVAILABLE = 3

Returned when the database schema can be upgraded.

UP_TO_DATE = 1

Returned when the database schema is up-to-date.

class sqlalchemy_helpers.manager.SyncResult(value, names=None, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

The result of a sync() call.

ALREADY_UP_TO_DATE = 1

Returned when the database schema was already up-to-date.

CREATED = 2

Returned when the database has been created.

UPGRADED = 3

Returned when the database schema has been upgraded.

sqlalchemy_helpers.manager.exists_in_db(bind, tablename, columnname=None)

Check whether a table and optionally a column exist in the database.

Parameters:

• bind (sqlalchemy.engine.Engine) – the database engine or connection.

• tablename (str) – the table to look for.

• columnname (str, optional) – the column to look for, if any. Defaults to None.

Returns:

Whether the database (and column) exist.

Return type:

bool

sqlalchemy_helpers.manager.get_base(*args, **kwargs)

A wrapper for declarative_base().

sqlalchemy_helpers.manager.get_by_pk(pk, *, session, model)

Get a model instance using its primary key.

Example: user = get_by_pk(42, session=session, model=User)

sqlalchemy_helpers.manager.get_one(session, model, **attrs)

Get a model instance using filters.

Example: user = get_one(session, User, name="foo")

sqlalchemy_helpers.manager.get_or_create(session, model, **attrs)

Function like Django’s get_or_create() method.

It will return a tuple, the first argument being the instance and the second being a boolean: True if the instance has been created and False otherwise.

Example: user, created = get_or_create(session, User, name="foo")

sqlalchemy_helpers.manager.is_sqlite(bind)

Check whether the database is SQLite.

Returns:

whether the database is SQLite.

Return type:

bool

sqlalchemy_helpers.manager.model_property(func)

Add a model property to call a function that uses the database model.

sqlalchemy_helpers.manager.session_and_model_property(Session, func)

Add a model property that uses the database session.

sqlalchemy_helpers.manager.set_sqlite_pragma(dbapi_connection, connection_record)

Automatically activate foreign keys on SQLite databases.