Coverage for /usr/lib/python3.10/site-packages/hyd/backend/user/models.py: 84%

1import datetime as dt 

2from typing import TypedDict 

3 

4from fastapi import status 

5from sqlalchemy import Boolean, Column, Integer, LargeBinary, String 

6from sqlalchemy.orm import Mapped, relationship 

7 

8from hyd.backend.db import EXTEND_EXISTING, DeclarativeMeta 

9from hyd.backend.exc import HTTPException_NO_PERMISSION 

10from hyd.backend.token.models import TokenEntry, TokenSchema 

11from hyd.backend.util.const import MAX_LENGTH_STR_ID 

12from hyd.backend.util.models import ( 

13 BASE_API_RESPONSE_SCHEMA, 

14 DETAIL_STR, 

15 NameStr, 

16 PrimaryKey, 

17 TimeStampMixin, 

18) 

19 

20#################################################################################################### 

21#### SQLAlchemy table definitions 

22#################################################################################################### 

23 

24 

25class UserEntry(DeclarativeMeta, TimeStampMixin): 

26 __tablename__ = "user_table" 

27 __table_args__ = {"extend_existing": EXTEND_EXISTING} 

28 id: Mapped[PrimaryKey] = Column(Integer, primary_key=True) 

29 username: Mapped[NameStr] = Column(String(length=MAX_LENGTH_STR_ID), unique=True) 

30 hashed_password: Mapped[bytes] = Column(LargeBinary) 

31 is_admin: Mapped[bool] = Column(Boolean) 

32 is_disabled: Mapped[bool] = Column(Boolean, default=False) 

33 

34 token_entries: Mapped[list[TokenEntry]] = relationship( 

35 "TokenEntry", back_populates="user_entry" 

36 ) 

37 login_token_entries: Mapped[list[TokenEntry]] = relationship( 

38 "TokenEntry", 

39 primaryjoin=("and_(TokenEntry.user_id==UserEntry.id, TokenEntry.is_login_token==True)"), 

40 viewonly=True, 

41 ) 

42 

43 _session_token_entry: TokenEntry | None = None 

44 _session_permitted_scopes: list[str] | None = None 

45 

46 @property 

47 def session_token_entry(self) -> TokenEntry: 

48 return self._session_token_entry 

49 

50 def check_scope_permission(self, *, scope: str) -> bool: # TODO raise if None 

51 """Check if the given scope is permitted for the given session.""" 

52 return scope in self._session_permitted_scopes 

53 

54 def check_token_project_permission(self, *, project_id: PrimaryKey) -> None: 

55 token_project_id = self._session_token_entry.project_id 

56 if token_project_id is None: 

57 return 

58 if token_project_id != project_id: 

59 raise HTTPException_NO_PERMISSION 

60 

61 

62#################################################################################################### 

63#### Response schema 

64#################################################################################################### 

65 

66# NOTE currently unused 

67class UserResponseSchema(TypedDict): 

68 id: PrimaryKey 

69 username: NameStr 

70 is_admin: bool 

71 is_disabled: bool 

72 created_at: dt.datetime 

73 

74 

75#################################################################################################### 

76#### OpenAPI definitions 

77#################################################################################################### 

78 

79 

80API_V1_LOGIN__POST = { 

81 **BASE_API_RESPONSE_SCHEMA, 

82 status.HTTP_200_OK: {"model": TokenSchema}, 

83 status.HTTP_400_BAD_REQUEST: DETAIL_STR, 

84} 

85 

86 

87API_V1_LOGOUT__PATCH = { 

88 **BASE_API_RESPONSE_SCHEMA, 

89 status.HTTP_200_OK: {"model": str}, 

90} 

91 

92API_V1_CHANGE_PASSWORD__PATCH = { 

93 **BASE_API_RESPONSE_SCHEMA, 

94 status.HTTP_200_OK: {"model": str}, 

95 status.HTTP_400_BAD_REQUEST: DETAIL_STR, 

96} 

97 

98 

99API_V1_GREET__GET = { 

100 **BASE_API_RESPONSE_SCHEMA, 

101 status.HTTP_200_OK: {"model": str}, 

102}