dlt.sources.rest_api.config_setup
create_response_hooks
def create_response_hooks(
response_actions: Optional[List[ResponseAction]]
) -> Optional[Dict[str, Any]]
Create response hooks based on the provided response actions. Note that if the error status code is not handled by the response actions, the default behavior is to raise an HTTP error.
Example:
def set_encoding(response, *args, **kwargs): response.encoding = 'windows-1252' return response
def remove_field(response: Response, *args, **kwargs) -> Response: payload = response.json() for record in payload: record.pop("email", None)
-
modified_content
- bytes = json.dumps(payload).encode("utf-8") response._content = modified_content return responseresponse_actions = [ set_encoding,
-
{"status_code"
- 404, "action": "ignore"}, -
{"content"
- "Not found", "action": "ignore"}, -
{"status_code"
- 200, "content": "some text", "action": "ignore"}, -
{"status_code"
- 200, "action": remove_field}, ] hooks = create_response_hooks(response_actions)
collect_resolved_values
def collect_resolved_values(
item: Dict[str, Any], resolved_params: List[ResolvedParam],
incremental: Optional[Incremental[Any]],
incremental_value_convert: Optional[Callable[...,
Any]]) -> Dict[str, Any]
Collects field values from the parent item based on resolved_params
and sets up incremental if present. Returns the resulting placeholders
(params_values) and a ResourcesContext that may store resources.<name>.<field>
.
expand_placeholders
def expand_placeholders(obj: Any, placeholders: Dict[str, Any]) -> Any
Recursively expand str.format placeholders in obj
using placeholders
.
build_parent_record
def build_parent_record(
item: Dict[str, Any], parent_resource_name: str,
include_from_parent: Optional[List[str]]) -> Dict[str, Any]
Builds a dictionary of the include_from_parent
fields from the parent,
renaming them using make_parent_key_name
.