Add SQLAlchemy support to your Flask application.
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Core extension class initialization, Flask app integration, and database configuration management. This includes setting up database URIs, engine options, and binding multiple databases.
The main extension class that integrates SQLAlchemy with Flask applications. Handles initialization, configuration, and provides access to database resources.
class SQLAlchemy:
def __init__(
self,
app: Flask | None = None,
*,
metadata: sa.MetaData | None = None,
session_options: dict[str, Any] | None = None,
query_class: type[Query] = Query,
model_class = Model,
engine_options: dict[str, Any] | None = None,
add_models_to_shell: bool = True,
disable_autonaming: bool = False,
):
"""
Initialize the SQLAlchemy extension.
Parameters:
- app: Flask application instance or None for later initialization
- metadata: Custom metadata instance for table definitions
- session_options: Options passed to sessionmaker
- query_class: Custom query class for Model.query
- model_class: Custom model base class or declarative base
- engine_options: Default options for all database engines
- add_models_to_shell: Add db and models to Flask shell context
- disable_autonaming: Disable automatic table name generation
"""
def init_app(self, app: Flask) -> None:
"""
Initialize the extension with a Flask application.
Must be called before accessing database resources. Sets up:
- Database configuration from app.config
- Engine creation for all bind keys
- Session teardown handlers
- Optional shell context processors
Parameters:
- app: Flask application to configure
"""
def __repr__(self) -> str:
"""String representation showing engine info and bind count."""Properties and methods for accessing and managing database engines across multiple databases.
@property
def engine(self) -> sa.engine.Engine:
"""The default database engine for the current application."""
@property
def engines(self) -> Mapping[str | None, sa.engine.Engine]:
"""Map of bind keys to engine instances for current application."""
def get_engine(
self, bind_key: str | None = None, **kwargs: Any
) -> sa.engine.Engine:
"""
Get the engine for the given bind key (deprecated).
Parameters:
- bind_key: The name of the engine, None for default
Returns:
Engine instance for the specified bind key
Note: Deprecated in favor of engines[key] property
"""Methods for creating, dropping, and reflecting database schemas across all or specific bind keys.
def create_all(self, bind_key: str | None | list[str | None] = "__all__") -> None:
"""
Create tables that do not exist in the database.
Parameters:
- bind_key: Bind key(s) to create tables for, "__all__" for all binds
"""
def drop_all(self, bind_key: str | None | list[str | None] = "__all__") -> None:
"""
Drop all tables from the database.
Parameters:
- bind_key: Bind key(s) to drop tables from, "__all__" for all binds
"""
def reflect(self, bind_key: str | None | list[str | None] = "__all__") -> None:
"""
Load table definitions from the database.
Parameters:
- bind_key: Bind key(s) to reflect tables from, "__all__" for all binds
"""Properties for accessing metadata instances that define table schemas.
@property
def metadata(self) -> sa.MetaData:
"""The default metadata used by Model and Table."""
@property
def metadatas(self) -> dict[str | None, sa.MetaData]:
"""Map of bind keys to MetaData instances."""Direct access to SQLAlchemy classes and functions through the extension instance, plus enhanced relationship methods that automatically apply the extension's Query class.
def relationship(self, *args: Any, **kwargs: Any) -> sa_orm.RelationshipProperty[Any]:
"""
Create a SQLAlchemy relationship that applies this extension's Query class
for dynamic relationships and backrefs.
This is a wrapper around sqlalchemy.orm.relationship that automatically
sets the query_class for dynamic relationships and backref configurations.
Parameters:
- args: Arguments passed to sqlalchemy.orm.relationship
- kwargs: Keyword arguments passed to sqlalchemy.orm.relationship
Returns:
RelationshipProperty configured with Flask-SQLAlchemy's Query class
"""
def dynamic_loader(self, argument: Any, **kwargs: Any) -> sa_orm.RelationshipProperty[Any]:
"""
Create a dynamic loader relationship that applies this extension's Query class.
This is a wrapper around sqlalchemy.orm.dynamic_loader that automatically
sets the query_class for relationships and backref configurations.
Parameters:
- argument: Mapper or class to relate to
- kwargs: Keyword arguments passed to sqlalchemy.orm.dynamic_loader
Returns:
RelationshipProperty configured for dynamic loading with Flask-SQLAlchemy's Query class
"""
def __getattr__(self, name: str) -> Any:
"""
Proxy access to sqlalchemy and sqlalchemy.orm modules.
Allows convenient access like db.Column, db.Integer, db.relationship, etc.
Special handling for 'event' -> sqlalchemy.event and deprecated 'relation'.
"""from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)
# or
db = SQLAlchemy()
db.init_app(app)app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///main.db"
app.config["SQLALCHEMY_BINDS"] = {
"users": "sqlite:///users.db",
"logs": "postgresql://user:pass@localhost/logs"
}
db = SQLAlchemy(app)
# Engines are automatically created for each bind
with app.app_context():
main_engine = db.engine # Default bind
users_engine = db.engines["users"]
logs_engine = db.engines["logs"]from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
db = SQLAlchemy(
app,
model_class=Base,
engine_options={"pool_timeout": 30},
disable_autonaming=True
)class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
posts = db.relationship('Post', backref='author')
# or explicitly use db.relationship for Flask-SQLAlchemy Query class
posts = db.relationship('Post', backref=db.backref('author', lazy='dynamic'))
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
# Dynamic loading with Flask-SQLAlchemy Query class
comments = db.dynamic_loader('Comment', backref='post')
# Query using the enhanced Query class
user = User.query.first()
user.posts.filter_by(published=True).all() # Uses Flask-SQLAlchemy Query methodsInstall with Tessl CLI
npx tessl i tessl/pypi-flask-sqlalchemy