Accessing session data when using custom store

[AakashGfude]

I have made a custom store that stores user information decoded from a token in the database.

Baiscally the code looks like:

class InDatabaseStore(SessionStore):
    """
    InDatabaseStore is a session store that stores session data in a database.
    It is used in the Session Middleware of the FastAPI app.
    """
    def __init__(self):
        self._storage = {}

    async def read(self, session_id: str, lifetime: int) -> Dict:
        """Read session data from a data source using session_id."""
        async for db in get_db_session():
            try:
                token = await get_token(db, session_id)
                if token:
                    self._storage[session_id] = {
                                "email": token.email,
                                "name": token.name,
                                "family_name": token.family_name,
                                "exp": token.exp,
                                "role_names": token.roles,
                            }
                    return self._storage
                else:
                    self._storage[session_id] = {}
                    return self._storage
            except Exception as e:
                # Return an error message if there's an issue with the database connection
                return {"error": f"Database connection error: {str(e)}"}

        # Return an error message if the loop is not entered (no database sessions)
        return {"error": "No database session established"}

    async def write(self, session_id: str, data: bytes, lifetime: int, ttl: int) -> str:
        """Write session data into data source and return session id."""
        async for db in get_db_session():
            if not await get_token(db, session_id):
                json_data = json.loads(data.decode("utf-8"))
                if "userinfo" not in json_data:
                    json_data["userinfo"] = {}
                json_data["userinfo"]["id"] = session_id
                await create_token(db, json_data["userinfo"])
                self._storage[session_id] = data
            return session_id
        return "No database session established"

    async def remove(self, session_id: str):
        """Remove session data."""
        # del self._storage[session_id]
        async for db in get_db_session():
            await delete_token(db, session_id)
            del self._storage[session_id]

    async def exists(self, session_id: str) -> bool:
        print("existing mode in database")
        return True if session_id else False

Now, I would have imagined that whatever is returned from read function will be stored in request.session? But I only get an empty object when accessing request.session in subsequent middleware. How to store the data in the session? Let me know if you have any ideas.

Thank you,

[alex-oleshkevich]

Did you load session using load_session or use SessionAutoloadMiddleware?

[alex-oleshkevich]

Also, session backend has to return bytes, not dict. Sorry, docs are outdated.

async def read(self, session_id: str, lifetime: int) -> bytes: ...

[AakashGfude]

It's working now. Thanks. I was loading the middleware before SessionAutoloadMiddleware. I can get the session data in request.session now.