SQLAlchemy-Utils

0.42.1 · active · verified Sun Mar 29

SQLAlchemy-Utils is a library that provides various utility functions, new data types, and helpers for SQLAlchemy. It extends SQLAlchemy's functionality by offering additional features to simplify common database tasks like managing database existence, custom column types (e.g., ChoiceType, UUIDType, EmailType), and ORM helpers. The library is actively maintained with frequent updates, typically a few minor releases per year, and is currently at version 0.42.1.

Warnings

breaking Version 0.42.0 dropped support for Python 3.7 and 3.8. Users on these Python versions must either upgrade their Python environment or pin `sqlalchemy-utils<0.42.0` to avoid compatibility issues.
Fix: Upgrade Python to 3.9+ or pin `sqlalchemy-utils` version: `pip install 'sqlalchemy-utils<0.42.0'`.
breaking Version 0.42.0 dropped support for SQLAlchemy 1.3. SQLAlchemy 2.0 support was added in 0.40.0 and subsequent versions include fixes for SQLAlchemy 2.0.x compatibility. Ensure your SQLAlchemy version is 1.4 or 2.x.
Fix: Upgrade SQLAlchemy to 1.4.x or 2.x: `pip install 'sqlalchemy>=1.4,<3.0'`.
gotcha When using `ColorType`, be aware of a potential import name conflict if the `colour-science` package is also installed, as it shares the same import name (`colour`) as the package `sqlalchemy-utils` expects. This was fixed in 0.39.0 to avoid crashes.
Fix: Upgrade to `sqlalchemy-utils>=0.39.0` or avoid installing `colour-science` alongside older versions.
gotcha `create_database` and `database_exists` functions handle different database dialects (e.g., PostgreSQL, MySQL, SQLite) differently. For PostgreSQL and MySQL, they connect to a default/master database to check/create the target database, requiring appropriate credentials and server access, not just database-specific access.
Fix: Ensure the provided URL has credentials for a database with sufficient privileges to create new databases (e.g., `postgres` user for PostgreSQL) and that the database server is running.
gotcha A specific `AttributeError` could occur with `Sequence` defaults in `instant_defaults_listener`. This issue was resolved in version 0.42.1.
Fix: Upgrade to `sqlalchemy-utils>=0.42.1`.

Install

pip install sqlalchemy-utils Install latest version

Imports

database_exists

from sqlalchemy_utils import database_exists

create_database

from sqlalchemy_utils import create_database

drop_database

from sqlalchemy_utils import drop_database

ChoiceType
```
from sqlalchemy_utils import ChoiceType
```
Timestamp
```
from sqlalchemy_utils import Timestamp
```

generic_repr

from sqlalchemy_utils import generic_repr

Quickstart

This quickstart demonstrates how to use `sqlalchemy-utils` to create and manage a PostgreSQL database programmatically, define a model with a custom `ChoiceType`, and perform basic ORM operations. It utilizes `database_exists`, `create_database`, and `ChoiceType` from the utility library. Environment variables are used for database credentials for security and flexibility. To run this, ensure a PostgreSQL server is accessible and you have `psycopg2-binary` or `psycopg` installed.

import os
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy_utils import database_exists, create_database, drop_database, ChoiceType

# --- Database Setup ---
DB_USER = os.environ.get('DB_USER', 'testuser')
DB_PASS = os.environ.get('DB_PASS', 'testpass')
DB_HOST = os.environ.get('DB_HOST', 'localhost')
DB_NAME = os.environ.get('DB_NAME', 'test_db')
DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}/{DB_NAME}"

engine = create_engine(DATABASE_URL)

# Create database if it doesn't exist
if not database_exists(engine.url):
    create_database(engine.url)
    print(f"Database '{DB_NAME}' created.")
else:
    print(f"Database '{DB_NAME}' already exists.")

Session = sessionmaker(bind=engine)
Base = declarative_base()

# --- Define a Model with ChoiceType ---
class TaskPriority:
    LOW = 'low'
    MEDIUM = 'medium'
    HIGH = 'high'

class Task(Base):
    __tablename__ = 'tasks'

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    priority = Column(ChoiceType(TaskPriority, impl=String(10)))

    def __repr__(self):
        return f"<Task(id={self.id}, name='{self.name}', priority='{self.priority}')>"

# --- ORM Operations ---
Base.metadata.create_all(engine) # Create tables

session = Session()

# Add tasks
task1 = Task(name="Write documentation", priority=TaskPriority.HIGH)
task2 = Task(name="Review code", priority=TaskPriority.MEDIUM)
task3 = Task(name="Deploy update", priority=TaskPriority.LOW)

session.add_all([task1, task2, task3])
session.commit()

# Query tasks
print("\nAll tasks:")
for task in session.query(Task).all():
    print(task)

# Clean up (optional: uncomment to drop the database)
# session.close()
# drop_database(engine.url)
# print(f"Database '{DB_NAME}' dropped.")

view raw JSON →