Coverage for packages/core/src/integrify/schemas.py: 83%

1import json 

2from typing import Any, ClassVar, Generic 

3 

4from integrify.utils import _ResponseT 

5from pydantic import BaseModel, Field, field_validator 

6from typing_extensions import TypedDict 

7 

8 

9class APIResponse(BaseModel, Generic[_ResponseT]): 

10 """Cavab sorğu base payload tipi. Generic tip-i qeyd etmıəklə 

11 sorğu cavabını validate edə bilərsiniz. 

12 """ 

13 

14 ok: bool = Field(validation_alias='is_success') 

15 """Cavab sorğusunun statusu 400dən kiçikdirsə""" 

16 

17 status_code: int 

18 """Cavab sorğusunun status kodu""" 

19 

20 headers: dict 

21 """Cavab sorğusunun header-i""" 

22 

23 body: _ResponseT = Field(validation_alias='content') 

24 """Cavab sorğusunun body-si""" 

25 

26 @field_validator('body', mode='before') 

27 @classmethod 

28 def convert_to_dict(cls, v: str | bytes): 

29 """Binary content-i dict-ə çevirərək, validation-a hazır vəziyyətə gətirir. 

30 

31 Cavab JSON deyilsə (məs., gateway xətası zamanı HTML səhifə və ya boş body), 

32 `json.loads` xətası atmaq əvəzinə boş dict qaytarılır ki, sorğu axını crash olmasın. 

33 Bu halda, status kodu və `ok` field-ləri vasitəsilə xətanı öyrənmək olar. 

34 """ 

35 if isinstance(v, (bytes, bytearray, str)): 35 ↛ 44line 35 didn't jump to line 44 because the condition on line 35 was always true

36 if not v: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 return {} 

38 

39 try: 

40 return json.loads(v) 

41 except (json.JSONDecodeError, ValueError, UnicodeDecodeError): 

42 return {} 

43 

44 return v 

45 

46 

47class DryResponse(TypedDict): 

48 """Dry-run sorğularının `return` tipi""" 

49 

50 url: str 

51 """Sorğu göndəriləcək url""" 

52 

53 verb: str 

54 """Sorğu metodu (GET, POST və s.)""" 

55 

56 headers: dict[str, str] 

57 """Sorğu header-ləri""" 

58 

59 data: dict[str, Any] 

60 """Sorğu data-sı (body-si)""" 

61 

62 request_args: dict[str, Any] 

63 """httpx.request funksiyasına ötürülən parametrlər""" 

64 

65 

66class PayloadBaseModel(BaseModel): 

67 URL_PARAM_FIELDS: ClassVar[set[str]] = set() 

68 

69 @classmethod 

70 def get_input_fields(cls) -> list[str]: 

71 """Modelin field-lərinin listini almaq""" 

72 return list(cls.model_fields.keys()) 

73 

74 @classmethod 

75 def from_args(cls, *args, **kwds): 

76 """Verilən `*args` və `**kwds` (və ya `**kwargs`) parametrlərini birləşdirərək 

77 Pydantic validasiyası edən funksiya. Positional arqumentlər üçün (`*args`) Pydantic 

78 modelindəki field-lərin ardıcıllığı və çağırılan funksiyada parametrlərinin ardıcıllığı 

79 EYNİ OLMALIDIR, əks halda, bu method yararsızdır. 

80 """ 

81 fields = cls.get_input_fields() 

82 

83 if len(args) > len(fields): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 raise TypeError( 

85 f'{cls.__name__}.from_args() got {len(args)} positional arguments ' 

86 f'but only {len(fields)} are expected' 

87 ) 

88 

89 positional = dict(zip(fields, args)) 

90 

91 # Eyni field həm positional, həm keyword kimi verilibsə, xəta qaldırırıq 

92 duplicates = positional.keys() & kwds.keys() 

93 if duplicates: 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true

94 raise TypeError( 

95 f'{cls.__name__}.from_args() got multiple values for ' 

96 f'argument(s): {", ".join(sorted(duplicates))}' 

97 ) 

98 

99 return cls.model_validate({**positional, **kwds})