codekingpro/portable-devtools

1import os2import functools3from typing import Callable, Dict, IO, List, Optional, Tuple, Union4 5from .argtypes import KeyValueArg6from .constants import (7    SEPARATORS_GROUP_MULTIPART, SEPARATOR_DATA_EMBED_FILE_CONTENTS,8    SEPARATOR_DATA_EMBED_RAW_JSON_FILE, SEPARATOR_GROUP_NESTED_JSON_ITEMS,9    SEPARATOR_DATA_RAW_JSON, SEPARATOR_DATA_STRING, SEPARATOR_FILE_UPLOAD,10    SEPARATOR_FILE_UPLOAD_TYPE, SEPARATOR_HEADER, SEPARATOR_HEADER_EMPTY,11    SEPARATOR_HEADER_EMBED, SEPARATOR_QUERY_PARAM,12    SEPARATOR_QUERY_EMBED_FILE, RequestType13)14from .dicts import (15    BaseMultiDict, MultipartRequestDataDict, RequestDataDict,16    RequestFilesDict, HTTPHeadersDict, RequestJSONDataDict,17    RequestQueryParamsDict,18)19from .exceptions import ParseError20from .nested_json import interpret_nested_json21from ..utils import get_content_type, load_json_preserve_order_and_dupe_keys, split_iterable22 23 24class RequestItems:25 26    def __init__(self, request_type: Optional[RequestType] = None):27        self.headers = HTTPHeadersDict()28        self.request_type = request_type29        self.is_json = request_type is None or request_type is RequestType.JSON30        self.data = RequestJSONDataDict() if self.is_json else RequestDataDict()31        self.files = RequestFilesDict()32        self.params = RequestQueryParamsDict()33        # To preserve the order of fields in file upload multipart requests.34        self.multipart_data = MultipartRequestDataDict()35 36    @classmethod37    def from_args(38        cls,39        request_item_args: List[KeyValueArg],40        request_type: Optional[RequestType] = None,41    ) -> 'RequestItems':42        instance = cls(request_type=request_type)43        rules: Dict[str, Tuple[Callable, dict]] = {44            SEPARATOR_HEADER: (45                process_header_arg,46                instance.headers,47            ),48            SEPARATOR_HEADER_EMPTY: (49                process_empty_header_arg,50                instance.headers,51            ),52            SEPARATOR_HEADER_EMBED: (53                process_embed_header_arg,54                instance.headers,55            ),56            SEPARATOR_QUERY_PARAM: (57                process_query_param_arg,58                instance.params,59            ),60            SEPARATOR_QUERY_EMBED_FILE: (61                process_embed_query_param_arg,62                instance.params,63            ),64            SEPARATOR_FILE_UPLOAD: (65                process_file_upload_arg,66                instance.files,67            ),68            SEPARATOR_DATA_STRING: (69                process_data_item_arg,70                instance.data,71            ),72            SEPARATOR_DATA_EMBED_FILE_CONTENTS: (73                process_data_embed_file_contents_arg,74                instance.data,75            ),76            SEPARATOR_GROUP_NESTED_JSON_ITEMS: (77                process_data_nested_json_embed_args,78                instance.data,79            ),80            SEPARATOR_DATA_RAW_JSON: (81                convert_json_value_to_form_if_needed(82                    in_json_mode=instance.is_json,83                    processor=process_data_raw_json_embed_arg84                ),85                instance.data,86            ),87            SEPARATOR_DATA_EMBED_RAW_JSON_FILE: (88                convert_json_value_to_form_if_needed(89                    in_json_mode=instance.is_json,90                    processor=process_data_embed_raw_json_file_arg,91                ),92                instance.data,93            ),94        }95 96        if instance.is_json:97            json_item_args, request_item_args = split_iterable(98                iterable=request_item_args,99                key=lambda arg: arg.sep in SEPARATOR_GROUP_NESTED_JSON_ITEMS100            )101            if json_item_args:102                pairs = [(arg.key, rules[arg.sep][0](arg)) for arg in json_item_args]103                processor_func, target_dict = rules[SEPARATOR_GROUP_NESTED_JSON_ITEMS]104                value = processor_func(pairs)105                target_dict.update(value)106 107        # Then handle all other items.108        for arg in request_item_args:109            processor_func, target_dict = rules[arg.sep]110            value = processor_func(arg)111 112            if arg.sep in SEPARATORS_GROUP_MULTIPART:113                instance.multipart_data[arg.key] = value114 115            if isinstance(target_dict, BaseMultiDict):116                target_dict.add(arg.key, value)117            else:118                target_dict[arg.key] = value119 120        return instance121 122 123JSONType = Union[str, bool, int, list, dict]124 125 126def process_header_arg(arg: KeyValueArg) -> Optional[str]:127    return arg.value or None128 129 130def process_embed_header_arg(arg: KeyValueArg) -> str:131    return load_text_file(arg).rstrip('\n')132 133 134def process_empty_header_arg(arg: KeyValueArg) -> str:135    if not arg.value:136        return arg.value137    raise ParseError(138        f'Invalid item {arg.orig!r} (to specify an empty header use `Header;`)'139    )140 141 142def process_query_param_arg(arg: KeyValueArg) -> str:143    return arg.value144 145 146def process_embed_query_param_arg(arg: KeyValueArg) -> str:147    return load_text_file(arg).rstrip('\n')148 149 150def process_file_upload_arg(arg: KeyValueArg) -> Tuple[str, IO, str]:151    parts = arg.value.split(SEPARATOR_FILE_UPLOAD_TYPE)152    filename = parts[0]153    mime_type = parts[1] if len(parts) > 1 else None154    try:155        f = open(os.path.expanduser(filename), 'rb')156    except OSError as e:157        raise ParseError(f'{arg.orig!r}: {e}')158    return (159        os.path.basename(filename),160        f,161        mime_type or get_content_type(filename),162    )163 164 165def convert_json_value_to_form_if_needed(in_json_mode: bool, processor: Callable[[KeyValueArg], JSONType]) -> Callable[[], str]:166    """167    We allow primitive values to be passed to forms via JSON key/value syntax.168 169    But complex values lead to an error because there’s no clear way to serialize them.170 171    """172    if in_json_mode:173        return processor174 175    @functools.wraps(processor)176    def wrapper(*args, **kwargs) -> str:177        try:178            output = processor(*args, **kwargs)179        except ParseError:180            output = None181        if isinstance(output, (str, int, float)):182            return str(output)183        else:184            raise ParseError('Cannot use complex JSON value types with --form/--multipart.')185 186    return wrapper187 188 189def process_data_item_arg(arg: KeyValueArg) -> str:190    return arg.value191 192 193def process_data_embed_file_contents_arg(arg: KeyValueArg) -> str:194    return load_text_file(arg)195 196 197def process_data_embed_raw_json_file_arg(arg: KeyValueArg) -> JSONType:198    contents = load_text_file(arg)199    value = load_json(arg, contents)200    return value201 202 203def process_data_raw_json_embed_arg(arg: KeyValueArg) -> JSONType:204    value = load_json(arg, arg.value)205    return value206 207 208def process_data_nested_json_embed_args(pairs) -> Dict[str, JSONType]:209    return interpret_nested_json(pairs)210 211 212def load_text_file(item: KeyValueArg) -> str:213    path = item.value214    try:215        with open(os.path.expanduser(path), 'rb') as f:216            return f.read().decode()217    except OSError as e:218        raise ParseError(f'{item.orig!r}: {e}')219    except UnicodeDecodeError:220        raise ParseError(221            f'{item.orig!r}: cannot embed the content of {item.value!r},'222            ' not a UTF-8 or ASCII-encoded text file'223        )224 225 226def load_json(arg: KeyValueArg, contents: str) -> JSONType:227    try:228        return load_json_preserve_order_and_dupe_keys(contents)229    except ValueError as e:230        raise ParseError(f'{arg.orig!r}: {e}')231