Coverage for docs/docs_src/user_guide/adapters/fastapi/security/main_1_simple.py: 84%

1from typing import Annotated, Any, Optional, Union 9 ctx1aefghibcd

2 

3from fastapi import Depends, FastAPI, HTTPException, status 9 ctx1aefghibcd

4from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm 9 ctx1aefghibcd

5from pydantic import BaseModel 9 ctx1aefghibcd

6 

7from fastagency.adapters.fastapi import FastAPIAdapter 9 ctx1aefghibcd

8 

9from .workflows import wf 9 ctx1aefghibcd

10 

11app = FastAPI(title="FastAPI with FastAgency") 9 ctx1aefghibcd

12 

13################################################################################ 

14# 

15# Taken from https://fastapi.tiangolo.com/tutorial/security/simple-oauth2/ 

16# 

17 

18fake_users_db = { 9 ctx1aefghibcd

19 "johndoe": { 

20 "username": "johndoe", 

21 "full_name": "John Doe", 

22 "email": "johndoe@example.com", 

23 "hashed_password": "fakehashedsecret", # pragma: allowlist secret 

24 "disabled": False, 

25 }, 

26 "alice": { 

27 "username": "alice", 

28 "full_name": "Alice Wonderson", 

29 "email": "alice@example.com", 

30 "hashed_password": "fakehashedsecret2", # pragma: allowlist secret 

31 "disabled": True, 

32 }, 

33} 

34 

35 

36def fake_hash_password(password: str) -> str: 9 ctx1aefghibcd

37 return "fakehashed" + password 4 ctx1abcd

38 

39 

40oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") 9 ctx1aefghibcd

41 

42 

43class User(BaseModel): 9 ctx1aefghibcd

44 username: str 9 ctx1aefghibcd

45 email: Union[str, None] = None 9 ctx1aefghibcd

46 full_name: Union[str, None] = None 9 ctx1aefghibcd

47 disabled: Union[bool, None] = None 9 ctx1aefghibcd

48 

49 

50class UserInDB(User): 9 ctx1aefghibcd

51 hashed_password: str 9 ctx1aefghibcd

52 

53 

54def get_user(db: dict[str, Any], username: str) -> Optional[UserInDB]: 9 ctx1aefghibcd

55 if username in db: 55 ↛ 58line 55 didn't jump to line 58 because the condition on line 55 was always true4 ctx1abcd

56 user_dict = db[username] 4 ctx1abcd

57 return UserInDB(**user_dict) 4 ctx1abcd

58 return None 

59 

60 

61def fake_decode_token(token: str) -> Optional[UserInDB]: 9 ctx1aefghibcd

62 # This doesn't provide any security at all 

63 # Check the next version 

64 user = get_user(fake_users_db, token) 4 ctx1abcd

65 return user 4 ctx1abcd

66 

67 

68async def get_current_user( 9 ctx1aefghibcd

69 token: Annotated[str, Depends(oauth2_scheme)], 

70) -> Optional[User]: 

71 user = fake_decode_token(token) 4 ctx1abcd

72 if not user: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true4 ctx1abcd

73 raise HTTPException( 

74 status_code=status.HTTP_401_UNAUTHORIZED, 

75 detail="Invalid authentication credentials", 

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

77 ) 

78 return user 4 ctx1abcd

79 

80 

81async def get_current_active_user( 9 ctx1aefghibcd

82 current_user: Annotated[User, Depends(get_current_user)], 

83) -> Optional[User]: 

84 if current_user.disabled: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true4 ctx1abcd

85 raise HTTPException(status_code=400, detail="Inactive user") 

86 return current_user 4 ctx1abcd

87 

88 

89@app.post("/token") 9 ctx1aefghibcd

90async def login( 9 ctx1aefghibcd

91 form_data: Annotated[OAuth2PasswordRequestForm, Depends()], 

92) -> dict[str, str]: 

93 user_dict = fake_users_db.get(form_data.username) 4 ctx1abcd

94 if not user_dict: 4 ctx1abcd

95 raise HTTPException(status_code=400, detail="Incorrect username or password") 

96 user = UserInDB(**user_dict) 4 ctx1abcd

97 hashed_password = fake_hash_password(form_data.password) 4 ctx1abcd

98 if not hashed_password == user.hashed_password: 4 ctx1abcd

99 raise HTTPException(status_code=400, detail="Incorrect username or password") 

100 

101 return {"access_token": user.username, "token_type": "bearer"} 4 ctx1abcd

102 

103 

104# 

105# End of code from https://fastapi.tiangolo.com/tutorial/security/simple-oauth2/ 

106# 

107################################################################################ 

108 

109 

110def get_user_id( 9 ctx1aefghibcd

111 current_user: Annotated[User, Depends(get_current_active_user)], 

112) -> Optional[str]: 

113 return current_user.username 4 ctx1abcd

114 

115 

116adapter = FastAPIAdapter(provider=wf, get_user_id=get_user_id) 9 ctx1aefghibcd

117app.include_router(adapter.router) 9 ctx1aefghibcd

118 

119 

120# this is optional, but we would like to see the list of available workflows 

121@app.get("/") 9 ctx1aefghibcd

122def read_root() -> dict[str, dict[str, str]]: 9 ctx1aefghibcd

123 return {"Workflows": {name: wf.get_description(name) for name in wf.names}}