Zareian
d25ecab0b3
TadrisinoTokenAuthentication ran one SELECT on accounts_user for every
authenticated request. Off by default (AUTH_USER_CACHE_TTL=0) so behaviour
is unchanged until a deployment opts in.
accounts/authentication.py
- _get_request_user(): when AUTH_USER_CACHE_TTL>0, cache all concrete
field values under auth:user:<id> and rebuild a full User instance
(so save(update_fields=[...]) stays safe); otherwise load from DB as
before. invalidate_user_cache() / _load_user_or_fail() helpers.
accounts/receivers.py (new)
- post_save/post_delete on User -> invalidate_user_cache(pk); covers
every in-app write path (SetUserGrade, login, refresh, last_activity).
accounts/apps.py
- AccountsConfig.ready() imports receivers.
accounts/tests.py
- AuthUserRowCacheTests: disabled -> loads every request; enabled ->
loads once then cache; User.save() invalidates; SetUserGradeView still
persists and the next request sees the new grade.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
from django.conf import settings
|
||
from django.core import signing
|
||
from django.core.cache import cache
|
||
from rest_framework.authentication import BaseAuthentication
|
||
from rest_framework.exceptions import AuthenticationFailed
|
||
|
||
from .models import User
|
||
|
||
ACCESS_TOKEN_SALT = 'tadrisino-access-token'
|
||
|
||
# فیلدهای غیرِ‌رابطه‌ایِ User که کش می‌شن؛ همه‌ی concrete fieldها تا بازسازیِ
|
||
# instance برای save(update_fields=...) امن بمونه.
|
||
_USER_CACHE_FIELDS = [f.attname for f in User._meta.concrete_fields]
|
||
|
||
|
||
def _auth_user_cache_key(user_id: int) -> str:
|
||
return f'auth:user:{user_id}'
|
||
|
||
|
||
def invalidate_user_cache(user_id: int) -> None:
|
||
cache.delete(_auth_user_cache_key(user_id))
|
||
|
||
|
||
def _load_user_or_fail(user_id: int) -> User:
|
||
try:
|
||
return User.objects.get(pk=user_id)
|
||
except User.DoesNotExist:
|
||
raise AuthenticationFailed('کاربر پیدا نشد.')
|
||
|
||
|
||
def _get_request_user(user_id: int) -> User:
|
||
"""
|
||
ردیفِ کاربر برای request جاری. با AUTH_USER_CACHE_TTL>0 از کش خونده
|
||
می‌شه (یک SELECT کمتر در هر request authenticated). instanceِ بازسازی‌شده
|
||
کاملِ همه‌ی فیلدهاست، پس save(update_fields=[...]) روش امنه.
|
||
"""
|
||
ttl = getattr(settings, 'AUTH_USER_CACHE_TTL', 0)
|
||
if not ttl:
|
||
return _load_user_or_fail(user_id)
|
||
|
||
key = _auth_user_cache_key(user_id)
|
||
data = cache.get(key)
|
||
if data is None:
|
||
user = _load_user_or_fail(user_id)
|
||
cache.set(key, {f: getattr(user, f) for f in _USER_CACHE_FIELDS}, ttl)
|
||
return user
|
||
|
||
user = User(**data)
|
||
user._state.adding = False
|
||
user._state.db = 'default'
|
||
return user
|
||
|
||
|
||
def issue_access_token(user_id: int) -> str:
|
||
"""
|
||
ساخت access token امضاشده (stateless) — دیتابیس ذخیره نمی‌شه، فقط
|
||
امضا و expiry داره. با django.core.signing (بدون نیاز به کتابخونه‌ی JWT).
|
||
"""
|
||
signer = signing.TimestampSigner(salt=ACCESS_TOKEN_SALT)
|
||
return signer.sign(str(user_id))
|
||
|
||
|
||
def decode_access_token(token: str) -> int:
|
||
signer = signing.TimestampSigner(salt=ACCESS_TOKEN_SALT)
|
||
try:
|
||
value = signer.unsign(token, max_age=settings.ACCESS_TOKEN_TTL_SECONDS)
|
||
except signing.SignatureExpired:
|
||
raise AuthenticationFailed('توکن منقضی شده است.')
|
||
except signing.BadSignature:
|
||
raise AuthenticationFailed('توکن نامعتبر است.')
|
||
return int(value)
|
||
|
||
|
||
class TadrisinoTokenAuthentication(BaseAuthentication):
|
||
"""
|
||
هدر مورد انتظار: Authorization: Bearer <access_token>
|
||
اگه هدر نباشه، درخواست به‌صورت anonymous ادامه پیدا می‌کنه (None, None)
|
||
و View خودش تصمیم می‌گیره که آیا auth اجباریه (IsAuthenticated) یا نه.
|
||
"""
|
||
|
||
keyword = 'Bearer'
|
||
|
||
def authenticate(self, request):
|
||
auth_header = request.headers.get('Authorization', '')
|
||
if not auth_header:
|
||
return None
|
||
|
||
parts = auth_header.split()
|
||
if len(parts) != 2 or parts[0] != self.keyword:
|
||
return None
|
||
|
||
token = parts[1]
|
||
user_id = decode_access_token(token)
|
||
return (_get_request_user(user_id), token)
|
||
|
||
def authenticate_header(self, request):
|
||
return self.keyword
|