Coverage for src/lexigram/web/dependencies/functions.py: 50%

1from __future__ import annotations 

2 

3from typing import Any 

4 

5from starlette import status 

6from starlette.requests import Request 

7 

8from lexigram.web.dependencies.state import RequestState 

9from lexigram.web.exceptions import HTTPError 

10 

11 

12async def get_request_id(request: Request) -> str: 

13 """Get request ID from state.""" 

14 request_id = RequestState.require(request, "request_id") 

15 if not isinstance(request_id, str): 

16 raise HTTPError( 

17 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

18 detail="Request ID must be a string", 

19 ) 

20 return request_id 

21 

22 

23async def get_current_user_optional(request: Request) -> Any | None: 

24 """Get current user from state (optional).""" 

25 return RequestState.get(request, "user", object, default=None) 

26 

27 

28async def get_current_user_required(request: Request) -> Any: 

29 """Get current user from state (required).""" 

30 user = RequestState.get(request, "user", object, default=None) 

31 if user is None: 

32 raise HTTPError( 

33 status_code=status.HTTP_401_UNAUTHORIZED, 

34 detail="Authentication required", 

35 headers={"WWW-Authenticate": "Bearer"}, 

36 ) 

37 return user 

38 

39 

40async def get_database(request: Request) -> Any: 

41 """Get database connection from state.""" 

42 return RequestState.require(request, "db")