Coverage for nexios\session\middleware.py: 95%

1from nexios.middlewares.base import BaseMiddleware 

2from .signed_cookies import SignedSessionManager 

3from .base import BaseSessionInterface 

4from nexios.http import Request, Response 

5from nexios.config import get_config 

6import warnings, typing 

7 

8 

9class SessionMiddleware(BaseMiddleware): 

10 def get_manager(self): 

11 if not self.config: 

12 return SignedSessionManager 

13 else: 

14 return self.config.manager or SignedSessionManager 

15 

16 async def process_request( 

17 self, 

18 request: Request, 

19 response: Response, 

20 call_next: typing.Callable[..., typing.Awaitable[typing.Any]], 

21 ): 

22 self.secret = get_config().secret_key 

23 

24 self.config = get_config().session 

25 if not self.secret: 

26 warnings.warn( 

27 "`secret_key` is not set, `secret_key` is required to use session", 

28 RuntimeWarning, 

29 ) 

30 return await call_next() 

31 

32 if self.config: 

33 session_cookie_name = self.config.session_cookie_name or "session_id" 

34 else: 

35 session_cookie_name = "session_id" 

36 

37 self.session_cookie_name = session_cookie_name 

38 manager = self.get_manager() 

39 request.scope["session"] = manager 

40 

41 session: type[BaseSessionInterface] = manager( 

42 session_key=request.cookies.get(session_cookie_name) # type:ignore 

43 ) 

44 await session.load() # type: ignore 

45 request.scope["session"] = session 

46 await call_next() 

47 

48 async def process_response(self, request: Request, response: Response): 

49 if not self.secret: 

50 return 

51 if request.session.is_empty() and request.session.accessed: 

52 response.delete_cookie(key=self.session_cookie_name) 

53 return 

54 

55 if request.session.should_set_cookie: 

56 await request.session.save() 

57 

58 session_key = request.session.get_session_key() 

59 response.set_cookie( 

60 key=self.session_cookie_name, 

61 value=session_key, 

62 domain=request.session.get_cookie_domain(), 

63 path=request.session.get_cookie_path(), 

64 httponly=request.session.get_cookie_httponly(), 

65 secure=request.session.get_cookie_secure(), 

66 samesite=request.session.get_cookie_samesite(), 

67 expires=request.session.get_expiration_time(), 

68 )