从多个数据库访问 Django 会话

好的，正如 comments 中所指出的，解决这个问题的方法是将默认的数据库会话存储子类化:https://github.com/django/django/blob/main/django/contrib/sessions/backends/db.py，这样我就可以根据需要添加using参数，以基于客户端的数据库更改和设置正确的会话.

首先定义您的新SessionStore(我的SessionStore非常类似，我只是在__init__和所需的方法中添加了Using变量):

import logging

from django.contrib.sessions.backends.base import CreateError, SessionBase, UpdateError
from django.core.exceptions import SuspiciousOperation
from django.db import DatabaseError, IntegrityError, router, transaction
from django.utils import timezone
from django.utils.functional import cached_property


class SessionStore(SessionBase):
    """
    Implement database session store.
    """

    def __init__(self, session_key=None, using=None):
        self.using = using # new, used for finding the right db to use
        super().__init__(session_key)


    def set_using(self, using):
        # not actually used because init covers it, but here if needed.
        self.using = using

    @classmethod
    def get_model_class(cls):
        # Avoids a circular import and allows importing SessionStore when
        # django.contrib.sessions is not in INSTALLED_APPS.
        from django.contrib.sessions.models import Session

        return Session

    @cached_property
    def model(self):
        return self.get_model_class()

    def _get_session_from_db(self):
        try:
            return self.model.objects.using(self.using).get(
                session_key=self.session_key, expire_date__gt=timezone.now()
            )
        except (self.model.DoesNotExist, SuspiciousOperation) as e:
            if isinstance(e, SuspiciousOperation):
                logger = logging.getLogger("django.security.%s" % e.__class__.__name__)
                logger.warning(str(e))
            self._session_key = None

    def load(self):
        s = self._get_session_from_db()
        return self.decode(s.session_data) if s else {}

    def exists(self, session_key):
        return self.model.objects.using(self.using).filter(session_key=session_key).exists()

    def create(self):
        while True:
            self._session_key = self._get_new_session_key()
            try:
                # Save immediately to ensure we have a unique entry in the
                # database.
                self.save(must_create=True)
            except CreateError:
                # Key wasn't unique. Try again.
                continue
            self.modified = True
            return

    def create_model_instance(self, data):
        """
        Return a new instance of the session model object, which represents the
        current session state. Intended to be used for saving the session data
        to the database.
        """
        return self.model(
            session_key=self._get_or_create_session_key(),
            session_data=self.encode(data),
            expire_date=self.get_expiry_date(),
        )

    def save(self, must_create=False):
        """
        Save the current session data to the database. If 'must_create' is
        True, raise a database error if the saving operation doesn't create a
        new entry (as opposed to possibly updating an existing entry).
        """
        if self.session_key is None:
            return self.create()
        data = self._get_session(no_load=must_create)
        obj = self.create_model_instance(data)
        # use the default using based on router if not directly passed in
        if not self.using:
            self.using = router.db_for_write(self.model, instance=obj)
        # print(using)
        try:
            with transaction.atomic(using=self.using):
                obj.save(
                    force_insert=must_create, force_update=not must_create, using=self.using
                )
        except IntegrityError:
            if must_create:
                raise CreateError
            raise
        except DatabaseError:
            if not must_create:
                raise UpdateError
            raise

    def delete(self, session_key=None):
        if session_key is None:
            if self.session_key is None:
                return
            session_key = self.session_key
        try:
            self.model.objects.using(self.using).get(session_key=session_key).delete()
        except self.model.DoesNotExist:
            pass

    @classmethod
    def clear_expired(cls):
        cls.get_model_class().objects.filter(expire_date__lt=timezone.now()).delete()

然后在settings.py或任何文件中的某个位置，使用您的定制SessionStore作为引擎:

SESSION_ENGINE = "utils.session_store"个

这就是我修改后的逻辑.

从我用于身份验证的Angular 来看，其中没有客户端URL，因此我无法从路由直接访问会话数据库:

def quickbooks_authentication(request):
    # ensure session key is there...we need it here.
    session_key = request.COOKIES.get('session_key')
    client_url = request.COOKIES.get('client_url')
    if not session_key:
        messages.error(request, 'Error, the session_key is not saved. Try again.')
        return redirect(request.META.get('HTTP_REFERER', f"{reverse('index', args=[])}"))
    # now ensure the session store exists, if not we have errors.
    store = SessionStore(session_key=session_key, using=client_url)
    if not store.exists(session_key=session_key):
        messages.error(request, 'Error, the SessionStore did not exist. Try again.')

    # now set the store to request.session,so it persists.
    request.session = store
    ...

...从那里，您可以正常访问会话，当在具有如下定义的下一个视图上恢复会话时，您从store中编辑的任何内容都将保留:

def closed_quickbooks_batch_detailed(request, client_url):
    print(request.session.values())  # good to go...