tessl/pypi-flask-security
Simple security for Flask apps
data-storage.mddocs/
Data Storage and Persistence
Flask-Security provides datastore classes that abstract user and role persistence across different database backends and ORMs. The datastore pattern allows Flask-Security to work with SQLAlchemy, MongoEngine, Peewee, and Pony ORM while providing a consistent interface.
Capabilities
SQLAlchemy Datastore
The most commonly used datastore for SQLAlchemy-based applications.
class SQLAlchemyUserDatastore:
"""SQLAlchemy implementation of UserDatastore"""
def __init__(self, db, user_model, role_model):
"""
Initialize SQLAlchemy datastore.
Args:
db: SQLAlchemy database instance
user_model: User model class
role_model: Role model class
"""
def get_user(self, identifier):
"""
Find user by ID or identity attributes (email).
Args:
identifier: User ID or email address
Returns:
User or None: Found user or None
"""
def find_user(self, **kwargs):
"""
Find user by filter criteria.
Args:
**kwargs: Filter criteria (e.g., email='user@example.com')
Returns:
User or None: Found user or None
"""
def find_role(self, role):
"""
Find role by name.
Args:
role (str): Role name
Returns:
Role or None: Found role or None
"""SQLAlchemy Session Datastore
Alternative SQLAlchemy datastore that uses a custom session instead of db.session.
class SQLAlchemySessionUserDatastore:
"""SQLAlchemy session-based implementation of UserDatastore"""
def __init__(self, session, user_model, role_model):
"""
Initialize session-based SQLAlchemy datastore.
Args:
session: SQLAlchemy session instance
user_model: User model class
role_model: Role model class
"""
def commit(self):
"""Commit the session"""MongoEngine Datastore
Datastore implementation for MongoEngine ODM.
class MongoEngineUserDatastore:
"""MongoEngine implementation of UserDatastore"""
def __init__(self, db, user_model, role_model):
"""
Initialize MongoEngine datastore.
Args:
db: MongoEngine database instance
user_model: User document class
role_model: Role document class
"""
def get_user(self, identifier):
"""
Find user by ID or identity attributes.
Args:
identifier: User ID or email address
Returns:
User or None: Found user or None
"""
def find_user(self, **kwargs):
"""
Find user using MongoEngine query methods.
Args:
**kwargs: Filter criteria
Returns:
User or None: Found user or None
"""
def find_role(self, role):
"""
Find role by name.
Args:
role (str): Role name
Returns:
Role or None: Found role or None
"""Peewee Datastore
Datastore implementation for Peewee ORM.
class PeeweeUserDatastore:
"""Peewee implementation of UserDatastore"""
def __init__(self, db, user_model, role_model, role_link):
"""
Initialize Peewee datastore.
Args:
db: Peewee database instance
user_model: User model class
role_model: Role model class
role_link: Many-to-many relationship model
"""
def get_user(self, identifier):
"""Find user by ID or identity attributes"""
def find_user(self, **kwargs):
"""Find user by filter criteria"""
def find_role(self, role):
"""Find role by name"""
def create_user(self, **kwargs):
"""Create user with custom role handling"""
def add_role_to_user(self, user, role):
"""Add role to user with custom implementation"""
def remove_role_from_user(self, user, role):
"""Remove role from user with custom implementation"""Pony ORM Datastore
Datastore implementation for Pony ORM with automatic session management.
class PonyUserDatastore:
"""Pony ORM implementation of UserDatastore"""
def __init__(self, db, user_model, role_model):
"""
Initialize Pony datastore.
Args:
db: Pony database instance
user_model: User entity class
role_model: Role entity class
"""
def get_user(self, identifier):
"""Find user by ID or identity attributes (with Pony session)"""
def find_user(self, **kwargs):
"""Find user by filter criteria (with Pony session)"""
def find_role(self, role):
"""Find role by name (with Pony session)"""
def add_role_to_user(self, user, role):
"""Add role to user (with Pony session)"""
def create_user(self, **kwargs):
"""Create user (with Pony session)"""
def create_role(self, **kwargs):
"""Create role (with Pony session)"""Base Datastore Interface
All datastores inherit from these base classes:
class Datastore:
"""Base datastore class"""
def __init__(self, db):
"""Initialize with database instance"""
def commit(self):
"""Commit database transaction"""
def put(self, model):
"""Save model to database"""
def delete(self, model):
"""Delete model from database"""
class UserDatastore(Datastore):
"""Base user datastore with user/role management"""
def __init__(self, user_model, role_model):
"""Initialize with user and role models"""
def get_user(self, id_or_email):
"""Get user by ID or email (abstract method)"""
def find_user(self, *args, **kwargs):
"""Find user by criteria (abstract method)"""
def find_role(self, *args, **kwargs):
"""Find role by criteria (abstract method)"""
def add_role_to_user(self, user, role):
"""Add role to user"""
def remove_role_from_user(self, user, role):
"""Remove role from user"""
def toggle_active(self, user):
"""Toggle user active status"""
def deactivate_user(self, user):
"""Deactivate user account"""
def activate_user(self, user):
"""Activate user account"""
def create_role(self, **kwargs):
"""Create new role"""
def find_or_create_role(self, name, **kwargs):
"""Find existing role or create new one"""
def create_user(self, **kwargs):
"""Create new user"""
def delete_user(self, user):
"""Delete user account"""Usage Examples
SQLAlchemy Setup
from flask_sqlalchemy import SQLAlchemy
from flask_security import Security, SQLAlchemyUserDatastore
db = SQLAlchemy(app)
# Create datastore
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
security = Security(app, user_datastore)
# Using the datastore
with app.app_context():
# Create a user
user_datastore.create_user(email='user@example.com', password='password')
# Create a role
user_datastore.create_role(name='admin', description='Administrator')
# Find and assign role
user = user_datastore.find_user(email='user@example.com')
role = user_datastore.find_role('admin')
user_datastore.add_role_to_user(user, role)
# Commit changes
db.session.commit()MongoEngine Setup
from flask_mongoengine import MongoEngine
from flask_security import Security, MongoEngineUserDatastore
db = MongoEngine(app)
# Create datastore
user_datastore = MongoEngineUserDatastore(db, User, Role)
security = Security(app, user_datastore)
# Using the datastore
with app.app_context():
# Create and save user
user = user_datastore.create_user(email='user@example.com', password='password')
# Create and save role
role = user_datastore.create_role(name='admin')
# Assign role
user_datastore.add_role_to_user(user, role)Multiple Datastores
You can use different datastores for different purposes:
# Primary user datastore
user_datastore = SQLAlchemyUserDatastore(db, User, Role)
# Read-only datastore for reporting
class ReadOnlyUserDatastore(SQLAlchemyUserDatastore):
def put(self, model):
raise NotImplementedError("Read-only datastore")
def delete(self, model):
raise NotImplementedError("Read-only datastore")
reporting_datastore = ReadOnlyUserDatastore(db, User, Role)Custom Datastore Methods
You can extend datastores with custom methods:
class CustomUserDatastore(SQLAlchemyUserDatastore):
def find_users_by_role(self, role_name):
"""Find all users with a specific role"""
role = self.find_role(role_name)
if role:
return role.users
return []
def get_active_users_count(self):
"""Count active users"""
return self.user_model.query.filter_by(active=True).count()
def find_users_needing_confirmation(self):
"""Find users who haven't confirmed their email"""
return self.user_model.query.filter_by(confirmed_at=None).all()tessl i tessl/pypi-flask-security@3.0.0