(
security: Any, allowed_fields: Optional[List[str]] = None
)
| 20 | |
| 21 | |
| 22 | def get_security( |
| 23 | security: Any, allowed_fields: Optional[List[str]] = None |
| 24 | ) -> Tuple[Dict[str, str], Dict[str, List[str]]]: |
| 25 | headers: Dict[str, str] = {} |
| 26 | query_params: Dict[str, List[str]] = {} |
| 27 | |
| 28 | if security is None: |
| 29 | return headers, query_params |
| 30 | |
| 31 | if not isinstance(security, BaseModel): |
| 32 | raise TypeError("security must be a pydantic model") |
| 33 | |
| 34 | sec_fields: Dict[str, FieldInfo] = security.__class__.model_fields |
| 35 | sec_field_names = ( |
| 36 | list(sec_fields.keys()) if allowed_fields is None else allowed_fields |
| 37 | ) |
| 38 | |
| 39 | for name in sec_field_names: |
| 40 | if name not in sec_fields: |
| 41 | continue |
| 42 | |
| 43 | sec_field = sec_fields[name] |
| 44 | |
| 45 | value = getattr(security, name) |
| 46 | if value is None: |
| 47 | continue |
| 48 | |
| 49 | metadata = find_field_metadata(sec_field, SecurityMetadata) |
| 50 | if metadata is None: |
| 51 | continue |
| 52 | if metadata.option: |
| 53 | _parse_security_option(headers, query_params, value) |
| 54 | return headers, query_params |
| 55 | if metadata.scheme: |
| 56 | # Special case for basic auth or custom auth which could be a flattened model |
| 57 | if metadata.sub_type in ["basic", "custom"] and not isinstance( |
| 58 | value, BaseModel |
| 59 | ): |
| 60 | _parse_security_scheme(headers, query_params, metadata, name, security) |
| 61 | else: |
| 62 | _parse_security_scheme(headers, query_params, metadata, name, value) |
| 63 | |
| 64 | if not metadata.composite: |
| 65 | return headers, query_params |
| 66 | |
| 67 | return headers, query_params |
| 68 | |
| 69 | |
| 70 | def get_security_from_env(security: Any, security_class: Any) -> Optional[BaseModel]: |
nothing calls this directly
no test coverage detected