tessl/pypi-social-auth-core
Python social authentication framework with 195+ backend providers for OAuth, OpenID Connect, and SAML integration.
—
Pending
storage-models.mddocs/
Storage Models
Abstract base classes and mixins for implementing user accounts, social account associations, and authentication data storage with any database or ORM system. The storage models provide a framework-agnostic interface that can be adapted to Django, SQLAlchemy, MongoDB, or any other data persistence layer.
Capabilities
User Social Auth Mixin
The main mixin for implementing social authentication user associations that link local user accounts to social provider accounts.
class UserMixin:
"""
Mixin for user social auth models.
This mixin provides the core functionality for managing social authentication
data associated with user accounts, including access tokens, refresh tokens,
and provider-specific extra data.
"""
# Class constants
ACCESS_TOKEN_EXPIRED_THRESHOLD: int = 5 # Seconds before expiry to consider expired
# Required fields (must be implemented by concrete model)
provider: str = "" # Provider name (e.g., 'google-oauth2', 'facebook')
uid: str | None = None # Provider-specific user ID
extra_data: dict | None = None # Additional provider data (tokens, profile info)
def save(self):
"""
Save the model instance.
Abstract method that must be implemented by the concrete model
to persist changes to the database.
Raises:
NotImplementedError: Must be implemented by subclasses
"""
def get_backend(self, strategy):
"""
Get backend class for this social account.
Parameters:
- strategy: Strategy instance for backend access
Returns:
Backend class for this provider
Raises:
MissingBackend: If backend not found for provider
"""
def get_backend_instance(self, strategy):
"""
Get backend instance for this social account.
Parameters:
- strategy: Strategy instance for backend instantiation
Returns:
Backend instance configured for this provider, or None if backend missing
"""
@property
def access_token(self):
"""
Get access token from extra_data.
Returns:
Access token string or None if not available
"""
def refresh_token(self, strategy, *args, **kwargs):
"""
Refresh access token using refresh token.
Attempts to refresh the access token using the stored refresh token
if the backend supports token refresh functionality.
Parameters:
- strategy: Strategy instance
- Additional arguments passed to backend refresh method
Returns:
None (updates extra_data in place)
"""
def set_extra_data(self, extra_data):
"""
Set extra data for this social account.
Updates the extra_data field with new information from provider,
merging with existing data and handling special token fields.
Parameters:
- extra_data: Dictionary of additional provider data
"""
def expiration_datetime(self):
"""
Get access token expiration datetime.
Returns:
datetime object for token expiration or None if no expiry data
"""
def access_token_expired(self):
"""
Check if access token has expired.
Uses ACCESS_TOKEN_EXPIRED_THRESHOLD to consider tokens expired
slightly before their actual expiration time.
Returns:
Boolean indicating if token is expired or expires soon
"""
def get_access_token(self, strategy):
"""
Get access token for API requests.
Parameters:
- strategy: Strategy instance
Returns:
Valid access token string
"""
def expiration_timedelta(self):
"""
Get access token expiration as timedelta.
Returns:
timedelta object for token expiration or None
"""
# Class methods - must be implemented by concrete storage classes
@classmethod
def clean_username(cls, value):
"""
Clean and validate username value.
Parameters:
- value: Raw username string
Returns:
Cleaned username string
"""
@classmethod
def changed(cls, user):
"""
Handle user change notification.
Called when user data changes to trigger any necessary
cleanup or update operations.
Parameters:
- user: User instance that changed
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get_username(cls, user):
"""
Get username from user instance.
Parameters:
- user: User instance
Returns:
Username string
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def user_model(cls):
"""
Get the user model class.
Returns:
User model class for this storage implementation
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def username_max_length(cls):
"""
Get maximum allowed username length.
Returns:
Integer maximum username length
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def allowed_to_disconnect(cls, user, backend_name, association_id=None):
"""
Check if user is allowed to disconnect this social account.
Parameters:
- user: User instance
- backend_name: Provider backend name
- association_id: Optional association ID
Returns:
Boolean indicating if disconnection is allowed
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def disconnect(cls, entry):
"""
Disconnect (delete) social account association.
Parameters:
- entry: Social account association to remove
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def user_exists(cls, *args, **kwargs):
"""
Check if user exists.
Returns:
Boolean indicating if user exists
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def create_user(cls, *args, **kwargs):
"""
Create new user account.
Returns:
New user instance
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get_user(cls, pk):
"""
Get user by primary key.
Parameters:
- pk: Primary key value
Returns:
User instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get_users_by_email(cls, email):
"""
Get users by email address.
Parameters:
- email: Email address string
Returns:
List of user instances with matching email
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get_social_auth(cls, provider, uid):
"""
Get social auth record by provider and UID.
Parameters:
- provider: Provider name string
- uid: Provider user ID
Returns:
Social auth instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get_social_auth_for_user(cls, user, provider=None, id=None):
"""
Get social auth records for user.
Parameters:
- user: User instance
- provider: Optional provider name filter
- id: Optional association ID filter
Returns:
List of social auth instances
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def create_social_auth(cls, user, uid, provider):
"""
Create social auth association.
Parameters:
- user: User instance
- uid: Provider user ID
- provider: Provider name
Returns:
New social auth instance
Raises:
NotImplementedError: Must be implemented by subclasses
"""Nonce Mixin
Mixin for storing OpenID nonces to prevent replay attacks.
class NonceMixin:
"""
Mixin for OpenID nonce models.
Prevents replay attacks by tracking used nonces and their timestamps
to ensure each nonce is only used once within its valid timeframe.
"""
# Required fields
server_url: str # OpenID provider server URL
timestamp: int # Nonce timestamp
salt: str # Nonce salt value
@classmethod
def use(cls, server_url, timestamp, salt):
"""
Use (create or verify) a nonce.
Parameters:
- server_url: Provider server URL
- timestamp: Nonce timestamp
- salt: Nonce salt
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get(cls, server_url, salt):
"""
Get nonce by server URL and salt.
Parameters:
- server_url: Provider server URL
- salt: Nonce salt
Returns:
Nonce instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def delete(cls, nonce):
"""
Delete nonce record.
Parameters:
- nonce: Nonce instance to delete
Raises:
NotImplementedError: Must be implemented by subclasses
"""Association Mixin
Base mixin for association models that store OpenID associations.
class AssociationMixin:
"""
Mixin for association models used in OpenID authentication.
Stores association data for OpenID providers including handles,
secrets, and expiration information.
"""
# Required fields
server_url: str # OpenID provider server URL
handle: str # Association handle
secret: str # Association secret (base64 encoded)
issued: int # Issue timestamp
lifetime: int # Association lifetime in seconds
assoc_type: str # Association type
@classmethod
def oids(cls, server_url, handle=None):
"""
Get OpenID associations for server.
Parameters:
- server_url: Provider server URL
- handle: Optional association handle filter
Returns:
List of association instances
"""
@classmethod
def openid_association(cls, assoc):
"""
Convert to OpenID association object.
Parameters:
- assoc: Association instance
Returns:
OpenID Association object
"""
@classmethod
def store(cls, server_url, association):
"""
Store OpenID association.
Parameters:
- server_url: Provider server URL
- association: OpenID Association object
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def get(cls, server_url=None, handle=None):
"""
Get association by server URL and/or handle.
Parameters:
- server_url: Provider server URL (optional)
- handle: Association handle (optional)
Returns:
Association instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def remove(cls, ids_to_delete):
"""
Remove associations by IDs.
Parameters:
- ids_to_delete: List of association IDs to delete
Raises:
NotImplementedError: Must be implemented by subclasses
"""Code Mixin
Mixin for email validation code storage.
class CodeMixin:
"""
Mixin for email validation code models.
Stores validation codes sent to user email addresses
for email verification workflows.
"""
# Required fields
email: str # Email address
code: str # Validation code
verified: bool # Whether code has been verified
def save(self):
"""
Save the code record.
Raises:
NotImplementedError: Must be implemented by subclasses
"""
def verify(self):
"""
Mark code as verified.
Sets verified=True and saves the record.
"""
@classmethod
def generate_code(cls):
"""
Generate random validation code.
Returns:
Random code string
"""
@classmethod
def make_code(cls, email):
"""
Create new validation code for email.
Parameters:
- email: Email address
Returns:
New code instance
"""
@classmethod
def get_code(cls, code):
"""
Get validation code by code string.
Parameters:
- code: Code string to find
Returns:
Code instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""Partial Mixin
Mixin for partial pipeline data storage.
class PartialMixin:
"""
Mixin for partial pipeline data models.
Stores intermediate pipeline state when authentication
process is paused (e.g., for email validation).
"""
# Required fields
token: str # Partial pipeline token
data: dict # Pipeline data dictionary
next_step: int # Next pipeline step index
backend: str # Backend name
@property
def args(self):
"""Get pipeline args from data."""
@args.setter
def args(self, value):
"""Set pipeline args in data."""
@property
def kwargs(self):
"""Get pipeline kwargs from data."""
@kwargs.setter
def kwargs(self, value):
"""Set pipeline kwargs in data."""
def extend_kwargs(self, values):
"""
Extend kwargs with additional values.
Parameters:
- values: Dictionary of additional kwargs
"""
@classmethod
def generate_token(cls):
"""
Generate random partial token.
Returns:
Random token string
"""
@classmethod
def load(cls, token):
"""
Load partial pipeline data by token.
Parameters:
- token: Partial pipeline token
Returns:
Partial instance or None
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def destroy(cls, token):
"""
Delete partial pipeline data.
Parameters:
- token: Partial pipeline token
Raises:
NotImplementedError: Must be implemented by subclasses
"""
@classmethod
def prepare(cls, backend, next_step, data):
"""
Prepare partial pipeline data.
Parameters:
- backend: Backend name
- next_step: Next pipeline step index
- data: Pipeline data dictionary
Returns:
Prepared partial instance
"""
@classmethod
def store(cls, partial):
"""
Store partial pipeline data.
Parameters:
- partial: Partial instance to store
Returns:
Stored partial instance
"""Base Storage Interface
Abstract base class defining the storage interface that must be implemented for each framework.
class BaseStorage:
"""
Abstract base storage interface.
Defines the contract that storage implementations must fulfill
to provide data persistence for social authentication.
"""
user = None # User model class
nonce = None # Nonce model class
association = None # Association model class
def is_integrity_error(self, exception):
"""
Check if exception is an integrity error.
Parameters:
- exception: Exception instance to check
Returns:
Boolean indicating if this is a database integrity error
"""
def create_user(self, *args, **kwargs):
"""
Create new user account.
Returns:
New user instance
"""
def get_user(self, pk):
"""
Get user by primary key.
Parameters:
- pk: User primary key
Returns:
User instance or None if not found
"""
def create_social_auth(self, user, uid, provider):
"""
Create social authentication association.
Parameters:
- user: User instance
- uid: Provider user ID
- provider: Provider name
Returns:
New social auth instance
"""
def get_social_auth(self, provider, uid):
"""
Get social auth association.
Parameters:
- provider: Provider name
- uid: Provider user ID
Returns:
Social auth instance or None if not found
"""
def get_social_auth_for_user(self, user, provider=None):
"""
Get social auth associations for user.
Parameters:
- user: User instance
- provider: Provider name filter (optional)
Returns:
QuerySet or list of social auth instances
"""
def create_nonce(self, server_url, timestamp, salt):
"""
Create OpenID nonce.
Parameters:
- server_url: Provider server URL
- timestamp: Nonce timestamp
- salt: Nonce salt
Returns:
New nonce instance
"""
def get_nonce(self, server_url, timestamp, salt):
"""
Get OpenID nonce.
Parameters:
- server_url: Provider server URL
- timestamp: Nonce timestamp
- salt: Nonce salt
Returns:
Nonce instance or None if not found
"""
def create_association(self, server_url, handle, secret, issued, lifetime, assoc_type):
"""
Create OpenID association.
Parameters:
- server_url: Provider server URL
- handle: Association handle
- secret: Association secret
- issued: Issue timestamp
- lifetime: Lifetime in seconds
- assoc_type: Association type
Returns:
New association instance
"""
def get_association(self, server_url, handle=None):
"""
Get OpenID association.
Parameters:
- server_url: Provider server URL
- handle: Association handle (optional)
Returns:
Association instance or None if not found
"""
def remove_association(self, server_url, handle):
"""
Remove OpenID association.
Parameters:
- server_url: Provider server URL
- handle: Association handle
"""User Social Auth Manager
Base manager class for user social auth model management.
class BaseUserSocialAuthManager:
"""
Base manager for user social auth models.
Provides common query methods for social authentication associations
that can be customized by framework-specific implementations.
"""
def get_social_auth(self, provider, uid):
"""
Get social auth by provider and UID.
Parameters:
- provider: Provider name
- uid: Provider user ID
Returns:
Social auth instance or raises DoesNotExist
"""
def get_social_auth_for_user(self, user, provider=None):
"""
Get social auths for user.
Parameters:
- user: User instance
- provider: Provider name filter (optional)
Returns:
QuerySet of social auth instances
"""
def create_social_auth(self, user, uid, provider):
"""
Create social auth association.
Parameters:
- user: User instance
- uid: Provider user ID
- provider: Provider name
Returns:
New social auth instance
"""
def username_max_length(self):
"""
Get maximum username length.
Returns:
Integer maximum length for username field
"""
def user_model(self):
"""
Get user model class.
Returns:
User model class for this storage implementation
"""Framework Integration Examples
Django Implementation
from django.db import models
from django.contrib.auth.models import AbstractUser
from social_core.storage import UserMixin, AssociationMixin
class User(AbstractUser):
"""Custom user model."""
pass
class UserSocialAuth(UserMixin, models.Model):
"""Social authentication association model."""
user = models.ForeignKey(User, related_name='social_auth', on_delete=models.CASCADE)
provider = models.CharField(max_length=32)
uid = models.CharField(max_length=255)
extra_data = models.JSONField(default=dict)
class Meta:
unique_together = ('provider', 'uid')
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
class Association(AssociationMixin, models.Model):
"""OpenID association model."""
server_url = models.CharField(max_length=255)
handle = models.CharField(max_length=255)
secret = models.CharField(max_length=255)
issued = models.IntegerField()
lifetime = models.IntegerField()
assoc_type = models.CharField(max_length=64)
def save(self, *args, **kwargs):
super().save(*args, **kwargs)SQLAlchemy Implementation
from sqlalchemy import Column, Integer, String, JSON, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from social_core.storage import UserMixin, AssociationMixin
Base = declarative_base()
class User(Base):
"""User model."""
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(150), unique=True)
email = Column(String(254))
class UserSocialAuth(UserMixin, Base):
"""Social auth model."""
__tablename__ = 'user_social_auth'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
provider = Column(String(32))
uid = Column(String(255))
extra_data = Column(JSON)
def save(self):
session.add(self)
session.commit()The storage models provide the foundation for persisting social authentication data while maintaining flexibility to work with any database system or ORM through the abstract interface pattern.
Install with Tessl CLI
npx tessl i tessl/pypi-social-auth-core