Coverage for fast_depends/pydantic/schema.py: 100%

1from collections.abc import Iterable 15 ctx1abcdefghijklmno

2from typing import Any 15 ctx1abcdefghijklmno

3 

4from fast_depends.core import CallModel 15 ctx1abcdefghijklmno

5from fast_depends.pydantic._compat import PYDANTIC_V2, create_model, model_schema 15 ctx1abcdefghijklmno

6 

7 

8def get_schema( 15 ctx1abcdefghijklmno

9 call: CallModel, 

10 *, 

11 embed: bool = False, 

12 resolve_refs: bool = False, 

13 exclude: Iterable[str] = (), 

14) -> dict[str, Any]: 

15 class_options: dict[str, Any] = { 15 ctx1abcdefghijklmno

16 i.field_name: (i.field_type, i.default_value) 

17 for i in call.flat_params 

18 if i.field_name not in exclude 

19 } 

20 

21 name = getattr(call.serializer, "name", "Undefined") 15 ctx1abcdefghijklmno

22 

23 if not class_options: 15 ctx1abcdefghijklmno

24 return {"title": name, "type": "null"} 15 ctx1abcdefghijklmno

25 

26 params_model = create_model(name, **class_options) 15 ctx1abcdefghijklmno

27 

28 body = model_schema(params_model) 15 ctx1abcdefghijklmno

29 

30 if resolve_refs: 15 ctx1abcdefghijklmno

31 pydantic_key = "$defs" if PYDANTIC_V2 else "definitions" 15 ctx1abcdefghijklmno

32 body = _move_pydantic_refs(body, pydantic_key) 15 ctx1abcdefghijklmno

33 body.pop(pydantic_key, None) 15 ctx1abcdefghijklmno

34 

35 if embed and len(body["properties"]) == 1: 15 ctx1abcdefghijklmno

36 body = list(body["properties"].values())[0] 15 ctx1abcdefghijklmno

37 

38 return body 15 ctx1abcdefghijklmno

39 

40 

41def _move_pydantic_refs( 15 ctx1abcdefghijklmno

42 original: Any, key: str, refs: dict[str, Any] | None = None 

43) -> Any: 

44 if not isinstance(original, dict): 15 ctx1abcdefghijklmno

45 return original 15 ctx1abcdefghijklmno

46 

47 data = original.copy() 15 ctx1abcdefghijklmno

48 

49 if refs is None: 15 ctx1abcdefghijklmno

50 raw_refs = data.get(key, {}) 15 ctx1abcdefghijklmno

51 refs = _move_pydantic_refs(raw_refs, key, raw_refs) 15 ctx1abcdefghijklmno

52 

53 name: str | None = None 15 ctx1abcdefghijklmno

54 for k in data: 15 ctx1abcdefghijklmno

55 if k == "$ref": 15 ctx1abcdefghijklmno

56 name = data[k].replace(f"#/{key}/", "") 15 ctx1abcdefghijklmno

57 

58 elif isinstance(data[k], dict): 15 ctx1abcdefghijklmno

59 data[k] = _move_pydantic_refs(data[k], key, refs) 15 ctx1abcdefghijklmno

60 

61 elif isinstance(data[k], list): 15 ctx1abcdefghijklmno

62 for i in range(len(data[k])): 15 ctx1abcdefghijklmno

63 data[k][i] = _move_pydantic_refs(data[k][i], key, refs) 15 ctx1abcdefghijklmno

64 

65 if name: 15 ctx1abcdefghijklmno

66 assert refs, "Smth wrong" 15 ctx1abcdefghijklmno

67 data = refs[name] 15 ctx1abcdefghijklmno

68 

69 return data 15 ctx1abcdefghijklmno