Coverage for whole_app/views.py: 100%

1import typing 

2 

3import fastapi 

4import structlog 

5from anyio import to_thread 

6 

7from . import dictionaries, misc_helpers, models, spell 

8from .auth import auth_via_api_key 

9from .dictionaries.protocol import UserDictProtocol 

10from .settings import SETTINGS 

11 

12 

13LOGGER_OBJ: typing.Final = structlog.get_logger() 

14SPELL_APP: typing.Final = fastapi.FastAPI( 

15 title=SETTINGS.app_title, 

16 version=SETTINGS.current_version, 

17 docs_url=SETTINGS.docs_url, 

18 openapi_url=f"{SETTINGS.api_prefix}/openapi.json", 

19) 

20if SETTINGS.enable_cors: 

21 from fastapi.middleware.cors import CORSMiddleware 

22 

23 SPELL_APP.add_middleware( 

24 CORSMiddleware, 

25 allow_origins=("*",), 

26 allow_credentials=True, 

27 allow_methods=["*"], 

28 allow_headers=["*"], 

29 ) 

30if SETTINGS.sentry_dsn: 

31 import sentry_sdk 

32 from sentry_sdk.integrations.asgi import SentryAsgiMiddleware 

33 

34 sentry_sdk.init(dsn=SETTINGS.sentry_dsn) 

35 SPELL_APP.add_middleware(SentryAsgiMiddleware) 

36 

37 

38@SPELL_APP.on_event("startup") 

39def startup() -> None: 

40 dictionaries.init_storage() 

41 misc_helpers.init_logger() 

42 LOGGER_OBJ.info("Current settings: %s", SETTINGS) 

43 

44 

45@SPELL_APP.post(f"{SETTINGS.api_prefix}/check/", summary="Check spelling") 

46async def spell_check_main_endpoint( 

47 request_payload: models.SpellCheckRequest, 

48 spell_service: typing.Annotated[ 

49 spell.SpellCheckService, 

50 fastapi.Depends(spell.SpellCheckService), 

51 ], 

52 storage_engine: typing.Annotated[ 

53 UserDictProtocol, 

54 fastapi.Depends(dictionaries.prepare_storage_engine), 

55 ], 

56) -> models.SpellCheckResponse: 

57 """Check spelling of text for exact language.""" 

58 exclusion_words: list[str] = [] 

59 if request_payload.user_name and not SETTINGS.dictionaries_disabled: 

60 exclusion_words = await storage_engine.prepare( 

61 request_payload.user_name, 

62 ).fetch_records() 

63 return models.SpellCheckResponse( 

64 **request_payload.model_dump(), 

65 corrections=await to_thread.run_sync( 

66 spell_service.prepare(request_payload, exclusion_words).run_check, 

67 ), 

68 ) 

69 

70 

71@SPELL_APP.get(f"{SETTINGS.api_prefix}/health/", summary="Regular healthcheck api") 

72async def check_health_of_service() -> models.HealthCheckResponse: 

73 """Check health of service.""" 

74 return models.HealthCheckResponse() 

75 

76 

77if not SETTINGS.dictionaries_disabled: 

78 

79 @SPELL_APP.post( 

80 f"{SETTINGS.api_prefix}/dictionaries/", 

81 summary="Add word to user dictionary", 

82 status_code=201, 

83 ) 

84 async def save_word( 

85 request_model: models.UserDictionaryRequestWithWord, 

86 storage_engine: typing.Annotated[ 

87 UserDictProtocol, 

88 fastapi.Depends(dictionaries.prepare_storage_engine), 

89 ], 

90 _: typing.Annotated[str, fastapi.Depends(auth_via_api_key)], 

91 ) -> bool: 

92 """Save word to user dictionary.""" 

93 await storage_engine.prepare(request_model.user_name).save_record( 

94 request_model.exception_word, 

95 ) 

96 return True 

97 

98 @SPELL_APP.delete( 

99 f"{SETTINGS.api_prefix}/dictionaries/", 

100 summary="Remove word from user dictionary", 

101 ) 

102 async def delete_word( 

103 request_model: models.UserDictionaryRequestWithWord, 

104 storage_engine: typing.Annotated[ 

105 UserDictProtocol, 

106 fastapi.Depends(dictionaries.prepare_storage_engine), 

107 ], 

108 _: typing.Annotated[str, fastapi.Depends(auth_via_api_key)], 

109 ) -> bool: 

110 """Save word to user dictionary.""" 

111 await storage_engine.prepare(request_model.user_name).remove_record( 

112 request_model.exception_word, 

113 ) 

114 return True