bt-source/panel/class/flask_session/sessions.py

import dataclasses
import random
import secrets
import time
from abc import ABC
try:
    import cPickle as pickle
except ImportError:
    import pickle

from datetime import datetime, timezone

from flask.sessions import SessionInterface as FlaskSessionInterface
from flask.sessions import SessionMixin
from itsdangerous import BadSignature, Signer, want_bytes
from werkzeug.datastructures import CallbackDict
from .utils import check_flask_sqlalchemy_version


def total_seconds(td):
    return td.days * 60 * 60 * 24 + td.seconds


class ServerSideSession(CallbackDict, SessionMixin):
    """Baseclass for server-side based sessions."""

    def __bool__(self) -> bool:
        return bool(dict(self)) and self.keys() != {"_permanent"}

    def __init__(self, initial=None, sid=None, permanent=None):
        def on_update(self):
            self.modified = True

        CallbackDict.__init__(self, initial, on_update)
        self.sid = sid
        if permanent:
            self.permanent = permanent
        self.modified = False


class RedisSession(ServerSideSession):
    pass


class MemcachedSession(ServerSideSession):
    pass


class FileSystemSession(ServerSideSession):
    pass


class MongoDBSession(ServerSideSession):
    pass


class SqlAlchemySession(ServerSideSession):
    pass


class SessionInterface(FlaskSessionInterface):
    def _generate_sid(self, session_id_length):
        return secrets.token_urlsafe(session_id_length)

    def __get_signer(self, app):
        if not hasattr(app, "secret_key") or not app.secret_key:
            raise KeyError("SECRET_KEY must be set when SESSION_USE_SIGNER=True")
        return Signer(app.secret_key, salt="flask-session", key_derivation="hmac")

    def _unsign(self, app, sid):
        signer = self.__get_signer(app)
        sid_as_bytes = signer.unsign(sid)
        sid = sid_as_bytes.decode()
        return sid

    def _sign(self, app, sid):
        signer = self.__get_signer(app)
        sid_as_bytes = want_bytes(sid)
        return signer.sign(sid_as_bytes).decode("utf-8")


class NullSessionInterface(SessionInterface):
    """Used to open a :class:`flask.sessions.NullSession` instance.

    If you do not configure a different ``SESSION_TYPE``, this will be used to
    generate nicer error messages.  Will allow read-only access to the empty
    session but fail on setting.
    """

    def open_session(self, app, request):
        return None


class ServerSideSessionInterface(SessionInterface, ABC):
    """Used to open a :class:`flask.sessions.ServerSideSessionInterface` instance."""

    def __init__(self, db, key_prefix, use_signer=False, permanent=True, sid_length=32):
        self.db = db
        self.key_prefix = key_prefix
        self.use_signer = use_signer
        self.permanent = permanent
        self.sid_length = sid_length
        self.has_same_site_capability = hasattr(self, "get_cookie_samesite")

    def set_cookie_to_response(self, app, session, response, expires):
        session_id = self._sign(app, session.sid) if self.use_signer else session.sid
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)
        httponly = self.get_cookie_httponly(app)
        secure = self.get_cookie_secure(app)
        samesite = None
        if self.has_same_site_capability:
            samesite = self.get_cookie_samesite(app)

        response.set_cookie(
            app.config["SESSION_COOKIE_NAME"],
            session_id,
            expires=expires,
            httponly=httponly,
            domain=domain,
            path=path,
            secure=secure,
            samesite=samesite,
        )

    def open_session(self, app, request):
        sid = request.cookies.get(app.config["SESSION_COOKIE_NAME"])
        if not sid:
            sid = self._generate_sid(self.sid_length)
            return self.session_class(sid=sid, permanent=self.permanent)
        if self.use_signer:
            try:
                sid = self._unsign(app, sid)
            except BadSignature:
                sid = self._generate_sid(self.sid_length)
                return self.session_class(sid=sid, permanent=self.permanent)
        return self.fetch_session(sid)

    def fetch_session(self, sid):
        raise NotImplementedError()


class RedisSessionInterface(ServerSideSessionInterface):
    """Uses the Redis key-value store as a session backend. (`redis-py` required)

    :param redis: A ``redis.Redis`` instance.
    :param key_prefix: A prefix that is added to all Redis store keys.
    :param use_signer: Whether to sign the session id cookie or not.
    :param permanent: Whether to use permanent session or not.
    :param sid_length: The length of the generated session id in bytes.

    .. versionadded:: 0.6
        The `sid_length` parameter was added.

    .. versionadded:: 0.2
        The `use_signer` parameter was added.
    """

    serializer = pickle
    session_class = RedisSession

    def __init__(self, redis, key_prefix, use_signer, permanent, sid_length):
        if redis is None:
            from redis import Redis

            redis = Redis()
        self.redis = redis
        super().__init__(redis, key_prefix, use_signer, permanent, sid_length)

    def fetch_session(self, sid):
        # Get the saved session (value) from the database
        prefixed_session_id = self.key_prefix + sid
        value = self.redis.get(prefixed_session_id)

        # If the saved session still exists and hasn't auto-expired, load the session data from the document
        if value is not None:
            try:
                session_data = self.serializer.loads(value)
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)

        # If the saved session  does not exist, create a new session
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app config
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                self.redis.delete(self.key_prefix + session.sid)
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        # Serialize the session data
        serialized_session_data = self.serializer.dumps(dict(session))

        # Update existing or create new session in the database
        self.redis.set(
            name=self.key_prefix + session.sid,
            value=serialized_session_data,
            ex=total_seconds(app.permanent_session_lifetime),
        )

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


class MemcachedSessionInterface(ServerSideSessionInterface):
    """A Session interface that uses memcached as backend. (`pylibmc` or `python-memcached` or `pymemcache` required)

    :param client: A ``memcache.Client`` instance.
    :param key_prefix: A prefix that is added to all Memcached store keys.
    :param use_signer: Whether to sign the session id cookie or not.
    :param permanent: Whether to use permanent session or not.
    :param sid_length: The length of the generated session id in bytes.

    .. versionadded:: 0.6
        The `sid_length` parameter was added.

    .. versionadded:: 0.2
        The `use_signer` parameter was added.

    """

    serializer = pickle
    session_class = MemcachedSession

    def __init__(self, client, key_prefix, use_signer, permanent, sid_length):
        if client is None:
            client = self._get_preferred_memcache_client()
        self.client = client
        super().__init__(client, key_prefix, use_signer, permanent, sid_length)

    def _get_preferred_memcache_client(self):
        clients = [
            ("pylibmc", ["127.0.0.1:11211"]),
            ("memcache", ["127.0.0.1:11211"]),
            ("pymemcache.client.base", "127.0.0.1:11211"),
        ]

        for module_name, server in clients:
            try:
                module = __import__(module_name)
                ClientClass = module.Client
                return ClientClass(server)
            except ImportError:
                continue

        raise ImportError("No memcache module found")

    def _get_memcache_timeout(self, timeout):
        """
        Memcached deals with long (> 30 days) timeouts in a special
        way. Call this function to obtain a safe value for your timeout.
        """
        if timeout > 2592000:  # 60*60*24*30, 30 days
            # Switch to absolute timestamps.
            timeout += int(time.time())
        return timeout

    def fetch_session(self, sid):
        # Get the saved session (item) from the database
        prefixed_session_id = self.key_prefix + sid
        item = self.client.get(prefixed_session_id)

        # If the saved session still exists and hasn't auto-expired, load the session data from the document
        if item is not None:
            try:
                session_data = self.serializer.loads(want_bytes(item))
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)

        # If the saved session  does not exist, create a new session
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app config
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # Generate a prefixed session id from the session id as a storage key
        prefixed_session_id = self.key_prefix + session.sid

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                self.client.delete(prefixed_session_id)
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        # Serialize the session data
        serialized_session_data = self.serializer.dumps(dict(session))

        # Update existing or create new session in the database
        self.client.set(
            prefixed_session_id,
            serialized_session_data,
            self._get_memcache_timeout(total_seconds(app.permanent_session_lifetime)),
        )

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


class FileSystemSessionInterface(ServerSideSessionInterface):
    """Uses the :class:`cachelib.file.FileSystemCache` as a session backend.

    :param cache_dir: the directory where session files are stored.
    :param threshold: the maximum number of items the session stores before it
                      starts deleting some.
    :param mode: the file mode wanted for the session files, default 0600
    :param key_prefix: A prefix that is added to FileSystemCache store keys.
    :param use_signer: Whether to sign the session id cookie or not.
    :param permanent: Whether to use permanent session or not.
    :param sid_length: The length of the generated session id in bytes.

    .. versionadded:: 0.6
        The `sid_length` parameter was added.

    .. versionadded:: 0.2
        The `use_signer` parameter was added.
    """

    session_class = FileSystemSession
    _save_check_keys = ("login", "tmp_login", "admin_auth", "api_request_tip", "down")

    def __init__(
        self,
        cache_dir,
        threshold,
        mode,
        key_prefix,
        use_signer,
        permanent,
        sid_length,
    ):
        from cachelib.file import FileSystemCache

        self.cache = FileSystemCache(cache_dir, threshold=threshold, mode=mode)
        super().__init__(self.cache, key_prefix, use_signer, permanent, sid_length)


    def fetch_session(self, sid):
        # Get the saved session (item) from the database
        prefixed_session_id = self.key_prefix + sid
        item = self.cache.get(prefixed_session_id)

        # If the saved session exists and has not auto-expired, load the session data from the item
        if item is not None:
            return self.session_class(item, sid=sid)

        # If the saved session  does not exist, create a new session
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app config
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # Generate a prefixed session id from the session id as a storage key
        prefixed_session_id = self.key_prefix + session.sid

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                self.cache.delete(prefixed_session_id)
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # 如果没能正确输入安全入口 且不是api请求 且不是临时登录
        if not any((i in session for i in self._save_check_keys)):
            self.cache.delete(prefixed_session_id)
            response.delete_cookie(
                app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
            )
            return

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        # Serialize the session data (or just cast into dictionary in this case)
        session_data = dict(session)

        # Update existing or create new session in the database
        self.cache.set(
            prefixed_session_id,
            session_data,
            total_seconds(app.permanent_session_lifetime),
        )

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


class MongoDBSessionInterface(ServerSideSessionInterface):
    """A Session interface that uses mongodb as backend. (`pymongo` required)

    :param client: A ``pymongo.MongoClient`` instance.
    :param db: The database you want to use.
    :param collection: The collection you want to use.
    :param key_prefix: A prefix that is added to all MongoDB store keys.
    :param use_signer: Whether to sign the session id cookie or not.
    :param permanent: Whether to use permanent session or not.
    :param sid_length: The length of the generated session id in bytes.

    .. versionadded:: 0.6
        The `sid_length` parameter was added.

    .. versionadded:: 0.2
        The `use_signer` parameter was added.
    """

    serializer = pickle
    session_class = MongoDBSession

    def __init__(
        self,
        client,
        db,
        collection,
        key_prefix,
        use_signer,
        permanent,
        sid_length,
    ):
        import pymongo

        if client is None:
            client = pymongo.MongoClient()

        self.client = client
        self.store = client[db][collection]
        self.use_deprecated_method = int(pymongo.version.split(".")[0]) < 4
        super().__init__(self.store, key_prefix, use_signer, permanent, sid_length)

    def fetch_session(self, sid):
        # Get the saved session (document) from the database
        prefixed_session_id = self.key_prefix + sid
        document = self.store.find_one({"id": prefixed_session_id})

        # If the expiration time is less than or equal to the current time (expired), delete the document
        if document is not None:
            expiration_datetime = document.get("expiration")
            # tz_aware mongodb fix
            expiration_datetime_tz_aware = expiration_datetime.replace(
                tzinfo=timezone.utc
            )
            now_datetime_tz_aware = datetime.utcnow().replace(tzinfo=timezone.utc)
            if expiration_datetime is None or (
                expiration_datetime_tz_aware <= now_datetime_tz_aware
            ):
                if self.use_deprecated_method:
                    self.store.remove({"id": prefixed_session_id})
                else:
                    self.store.delete_one({"id": prefixed_session_id})
                document = None

        # If the saved session still exists after checking for expiration, load the session data from the document
        if document is not None:
            try:
                session_data = self.serializer.loads(want_bytes(document["val"]))
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)

        # If the saved session does not exist, create a new session
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app config
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # Generate a prefixed session id from the session id as a storage key
        prefixed_session_id = self.key_prefix + session.sid

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                if self.use_deprecated_method:
                    self.store.remove({"id": prefixed_session_id})
                else:
                    self.store.delete_one({"id": prefixed_session_id})
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        # Serialize the session data
        serialized_session_data = self.serializer.dumps(dict(session))

        # Update existing or create new session in the database
        if self.use_deprecated_method:
            self.store.update(
                {"id": prefixed_session_id},
                {
                    "id": prefixed_session_id,
                    "val": serialized_session_data,
                    "expiration": expiration_datetime,
                },
                True,
            )
        else:
            self.store.update_one(
                {"id": prefixed_session_id},
                {
                    "$set": {
                        "id": prefixed_session_id,
                        "val": serialized_session_data,
                        "expiration": expiration_datetime,
                    }
                },
                True,
            )

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


class SqlAlchemySessionInterface(ServerSideSessionInterface):
    """Uses the Flask-SQLAlchemy from a flask app as a session backend.

    :param app: A Flask app instance.
    :param db: A Flask-SQLAlchemy instance.
    :param table: The table name you want to use.
    :param key_prefix: A prefix that is added to all store keys.
    :param use_signer: Whether to sign the session id cookie or not.
    :param permanent: Whether to use permanent session or not.
    :param sid_length: The length of the generated session id in bytes.
    :param sequence: The sequence to use for the primary key if needed.
    :param schema: The db schema to use
    :param bind_key: The db bind key to use

    .. versionadded:: 0.6
        The `sid_length`, `sequence`, `schema` and `bind_key` parameters were added.

    .. versionadded:: 0.2
        The `use_signer` parameter was added.
    """

    serializer = pickle
    session_class = SqlAlchemySession

    def __init__(
        self,
        app,
        db,
        table,
        sequence,
        schema,
        bind_key,
        key_prefix,
        use_signer,
        permanent,
        sid_length,
    ):
        if db is None:
            check_flask_sqlalchemy_version()
            from flask_sqlalchemy import SQLAlchemy
            db = SQLAlchemy(app)

        self.db = db
        self.sequence = sequence
        self.schema = schema
        self.bind_key = bind_key
        super().__init__(self.db, key_prefix, use_signer, permanent, sid_length)
        app.before_request(self._cleanup_n_requests)
        self.cleanup_n_requests = 100

        # Create the Session database model
        class Session(self.db.Model):
            __tablename__ = table

            if self.schema is not None:
                __table_args__ = {"schema": self.schema, "keep_existing": True}
            else:
                __table_args__ = {"keep_existing": True}

            if self.bind_key is not None:
                __bind_key__ = self.bind_key

            # Set the database columns, support for id sequences
            if sequence:
                id = self.db.Column(self.db.Integer, self.db.Sequence(sequence), primary_key=True)
            else:
                id = self.db.Column(self.db.Integer, primary_key=True)
            session_id = self.db.Column(self.db.String(255), unique=True)
            data = self.db.Column(self.db.LargeBinary)
            expiry = self.db.Column(self.db.DateTime, index=True)

            def __init__(self, session_id, data, expiry):
                self.session_id = session_id
                self.data = data
                self.expiry = expiry

            def __repr__(self):
                return "<Session data %s>" % self.data

        with app.app_context():
            self.db.create_all()

        self.sql_session_model = Session

    def fetch_session(self, sid):
        # Get the saved session (record) from the database
        store_id = self.key_prefix + sid
        record = self.sql_session_model.query.filter_by(session_id=store_id).first()

        # If the expiration time is less than or equal to the current time (expired), delete the document
        if record is not None:
            expiration_datetime = record.expiry
            if expiration_datetime is None or expiration_datetime <= datetime.utcnow():
                self.db.session.delete(record)
                self.db.session.commit()
                record = None

        # If the saved session still exists after checking for expiration, load the session data from the document
        if record:
            try:
                session_data = self.serializer.loads(want_bytes(record.data))
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # Generate a prefixed session id
        prefixed_session_id = self.key_prefix + session.sid

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                self.sql_session_model.query.filter_by(
                    session_id=prefixed_session_id
                ).delete()
                self.db.session.commit()
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # Serialize session data
        serialized_session_data = self.serializer.dumps(dict(session))

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        # Update existing or create new session in the database
        record = self.sql_session_model.query.filter_by(
            session_id=prefixed_session_id
        ).first()
        if record:
            record.data = serialized_session_data
            record.expiry = expiration_datetime
        else:
            record = self.sql_session_model(
                session_id=prefixed_session_id,
                data=serialized_session_data,
                expiry=expiration_datetime,
            )
            self.db.session.add(record)
        self.db.session.commit()

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


    def _cleanup_n_requests(self) -> None:
        """
        Delete expired sessions on average every N requests.

        This is less desirable than using the scheduled app command cleanup as it may
        slow down some requests but may be useful for rapid development.
        """
        if self.cleanup_n_requests and random.randint(0, self.cleanup_n_requests) == 0:
            try:
                self.db.session.query(self.sql_session_model).filter(
                    self.sql_session_model.expiry <= datetime.utcnow()
                ).delete(synchronize_session=False)
                self.db.session.commit()
            except Exception:
                self.db.session.rollback()



class SqliteSessionInterface(ServerSideSessionInterface):
    """
    sqlite 专用session实现，使用自定义的数据库连接，不在依赖SqlAlchemy
    """
    serializer = pickle
    session_class = SqlAlchemySession

    def __init__(
        self,
        app,
        db,
        table,
        sequence,
        schema,
        bind_key,
        key_prefix,
        use_signer,
        permanent,
        sid_length,
    ):
        from .sqlite_pool import FlaskSQLitePool, SQLiteConnection
        if db is None:
            app.config['SQLITE_DATABASE'] = app.config["SQLALCHEMY_DATABASE_URI"]
            db = FlaskSQLitePool(app)

        self.db: FlaskSQLitePool = db
        super().__init__(self.db, key_prefix, use_signer, permanent, sid_length)
        app.before_request(self._cleanup_n_requests)
        self.cleanup_n_requests = 100
        with app.app_context():
            with db.get_connection() as conn:
                conn.executescript(
                    """CREATE TABLE IF NOT EXISTS `sessions` (
        `id` INTEGER NOT NULL, 
        `session_id` VARCHAR(255), 
        `data` BLOB, 
        `expiry` DATETIME, 
        PRIMARY KEY (`id`), 
        UNIQUE (`session_id`)
);
CREATE INDEX IF NOT EXISTS `ix_sessions_expiry` ON sessions (`expiry`);
""")
                conn.commit()

        @dataclasses.dataclass
        class Record:
            id: int
            session_id: str
            data: bytes
            expiry: datetime

            @classmethod
            def from_row(cls, row) -> "Record":
                r_id = row[0]
                r_expiry = datetime.fromisoformat(row[3])
                r_expiry = r_expiry.replace(tzinfo=timezone.utc)
                return cls(r_id, row[1], row[2], r_expiry)

        self.record_class = Record


    def fetch_session(self, sid):
        # Get the saved session (record) from the database
        store_id = self.key_prefix + sid
        with self.db.get_connection() as conn:
            cursor = conn.cursor()
            res = cursor.execute("select * from sessions where session_id=? limit 1", (store_id,))
            row = res.fetchone()
            if not row:
                record = None
            else:
                record = self.record_class.from_row(row)

            # If the expiration time is less than or equal to the current time (expired), delete the document
            if record is not None:
                expiration_datetime = record.expiry
                now = datetime.now(tz=timezone.utc)
                if expiration_datetime is None or expiration_datetime <= now:
                    cursor.execute("DELETE FROM sessions WHERE id = ?", (record.id,))
                    cursor.close()
                    conn.commit()
                    conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
                    record = None
            else:
                cursor.close()


        # If the saved session still exists after checking for expiration, load the session data from the document
        if record:
            try:
                session_data = self.serializer.loads(want_bytes(record.data))
                return self.session_class(session_data, sid=sid)
            except pickle.UnpicklingError:
                return self.session_class(sid=sid, permanent=self.permanent)
        return self.session_class(sid=sid, permanent=self.permanent)

    def save_session(self, app, session, response):
        if not self.should_set_cookie(app, session):
            return

        # Get the domain and path for the cookie from the app
        domain = self.get_cookie_domain(app)
        path = self.get_cookie_path(app)

        # Generate a prefixed session id
        prefixed_session_id = self.key_prefix + session.sid

        # If the session is empty, do not save it to the database or set a cookie
        if not session:
            # If the session was deleted (empty and modified), delete the saved session  from the database and tell the client to delete the cookie
            if session.modified:
                with self.db.get_connection() as conn:
                    cursor = conn.cursor()
                    cursor.execute("DELETE FROM sessions WHERE id = ?", (prefixed_session_id,))
                    cursor.close()
                    conn.commit()
                    conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
                response.delete_cookie(
                    app.config["SESSION_COOKIE_NAME"], domain=domain, path=path
                )
            return

        # Serialize session data
        serialized_session_data = self.serializer.dumps(dict(session))

        # Get the new expiration time for the session
        expiration_datetime = self.get_expiration_time(app, session)

        with self.db.get_connection() as conn:
            # Update existing or create new session in the database
            cursor = conn.cursor()
            res = cursor.execute("select * from sessions where session_id=? limit 1", (prefixed_session_id,))
            row = res.fetchone()
            if not row:
                record = None
            else:
                record = self.record_class.from_row(row)

            if record:
                record.data = serialized_session_data
                record.expiry = expiration_datetime
                cursor.execute(
                    "update sessions set data = ?, expiry = ? where session_id = ?",
                    (serialized_session_data, expiration_datetime.isoformat(), prefixed_session_id),
                )
                cursor.close()
                conn.commit()
                conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
            else:
                cursor.execute(
                    "insert into sessions (session_id, data, expiry) values (?, ?, ?)",
                    (prefixed_session_id, serialized_session_data, expiration_datetime.isoformat()))
                cursor.close()
                conn.commit()
                conn.execute("PRAGMA wal_checkpoint(PASSIVE)")

        # Set the browser cookie
        self.set_cookie_to_response(app, session, response, expiration_datetime)


    def _cleanup_n_requests(self) -> None:
        """
        Delete expired sessions on average every N requests.

        This is less desirable than using the scheduled app command cleanup as it may
        slow down some requests but may be useful for rapid development.
        """
        if self.cleanup_n_requests and random.randint(0, self.cleanup_n_requests) == 0:
            try:
                with self.db.get_connection() as conn:
                    now = datetime.now(tz=timezone.utc)
                    cursor = conn.cursor()
                    cursor.execute("DELETE FROM sessions WHERE expiry  < ?", (now,))
                    cursor.close()
                    conn.commit()
                    conn.execute("PRAGMA wal_checkpoint(FULL)")
            except Exception:
                import traceback
                print("Error: Failed to delete expired sessions", traceback.format_exc(), flush=True)
                pass