Python Social Authで知っておくと便利な仕様

SOCIAL_AUTH_PIPELINE = (
    "social_core.pipeline.social_auth.social_details",
    "social_core.pipeline.social_auth.social_uid",
    "social_core.pipeline.social_auth.social_user",
    "social_core.pipeline.user.get_username",
    "social_core.pipeline.social_auth.associate_by_email",
    "social_core.pipeline.user.create_user",
    "social_core.pipeline.social_auth.associate_user",
    "social_core.pipeline.social_auth.load_extra_data",
    "social_core.pipeline.user.user_details",
    "apps.users.pipeline.set_user_data",
)

def set_user_data(backend, strategy, details, response, user=None, *args, **kwargs):
    # max_length以上の長さの文字列が保存されないようにスライスする.
    display_name_max_length = user._meta.get_field('display_name').max_length
    if backend.name == "twitter":
        user.display_name = response["name"][:display_name_max_length]

from django.utils.crypto import get_random_string

def get_username(backend, strategy, details, response, user=None, *args, **kwargs):
    """
    social_core.pipeline.user.get_usernameのカスタム用関数.
    新規ユーザーのusernameは `user_xxxxxxxxxx` を割り当てる. 
    """
    if not user:
        # 最大文字数以上にならないようにスライスする.
        username = f"user_{ get_random_string(10) }"[:usernameフィールドのmax_length]
    else:
        username = strategy.storage.user.get_username(user)
    return {"username": username}

def social_user(backend, uid, user=None, *args, **kwargs):
    provider = backend.name
    social = backend.strategy.storage.user.get_social_auth(provider, uid)
    if social:
        # user(ログイン中のUser) と social.user(プロバイダとuidの組み合わせで取得したUser) が異なる場合はAuthAlreadyAssociatedのエラー
        if user and social.user != user:
            raise AuthAlreadyAssociated(backend)
        elif not user:
            user = social.user
    return {
        'social': social,
        'user': user,
        'is_new': user is None,
        'new_association': social is None
    }

def social_user(backend, uid, user=None, *args, **kwargs):
    """
    social_core.pipeline.social_auth.social_userのカスタム用関数.
    """
    provider = backend.name
    social = backend.strategy.storage.user.get_social_auth(provider, uid)

    if social:
        # 連携しようとしたSNSアカウントが、リクエストユーザーとは別のアカウントと既に紐付け済だった場合.
        if user and social.user != user:
            # SNSアカウント連携処理を中断し、エラーメッセージを表示する.
            provider_names = {
                "github": "GitHub",
                "google-oauth2": "Google",
                "twitter": "Twitter",
            }
            messages.add_message(
                kwargs['strategy'].request,
                messages.ERROR,
                f"この{ provider_names[backend.name] }アカウントは既に登録されているため連携できません。"
            )
            redirect("settings:account")
        elif not user:
            user = social.user
    return {'social': social,
            'user': user,
            'is_new': user is None,
            'new_association': social is None}

def social_user(backend, uid, user=None, *args, **kwargs):
    """
    social_core.pipeline.social_auth.social_userのカスタム用関数.
    """
    provider = backend.name
    social = backend.strategy.storage.user.get_social_auth(provider, uid)

    provider_names = {
        "github": "GitHub",
        "google-oauth2": "Google",
        "twitter": "Twitter",
    }

    # 同プロバイダのソーシャルアカウントを複数連携できないようにするバリデーション.
    if user and user.social_auth.filter(provider=provider).exclude(uid=uid).exists():
        messages.add_message(
            kwargs['strategy'].request,
            messages.ERROR,
            f"連携に失敗しました。既に他の{ provider_names[provider] }アカウントを連携済です。"
        )
        return redirect("settings:account")

# ユーザーに紐づく認証情報一覧
>>> user.social_auth.all()
<QuerySet [<UserSocialAuth: taro>, <UserSocialAuth: taro>]>

# プロバイダーを指定して認証情報を取得
>>> user.social_auth.filter(provider='google-oauth2')
<QuerySet [<UserSocialAuth: taro>, <UserSocialAuth: taro>]>

# 認証プロバイダーの取得
>>> user.social_auth.first().provider
'google-oauth2'

# アクセストークン等の取得
>>> user.social_auth.filter(provider='google-oauth2').first().extra_data
{'auth_time': 123456789, 'expires': 3599, 'token_type': 'Bearer', 'access_token': 'XXXXXXXXXX'}

# APIを利用する例
>>> social = user.social_auth.get(provider='google-oauth2')
>>> response = requests.get(
    'https://www.googleapis.com/plus/v1/people/me/people/visible',
    params={'access_token': social.extra_data['access_token']}
)
>>> response.json()['items']

{{ backends }}
<!--  {'associated': <QuerySet [<UserSocialAuth: username1>]>, 'not_associated': ['twitter', 'github'], 'backends': ['github', 'google-oauth2', 'twitter']} -->