From 098b328b7a3040df17885e1af1c176472a3e6e34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=87a=C4=9F=C4=B1l=20Ulu=C5=9Fahin?= Date: Fri, 23 Oct 2026 14:49:32 +0100 Subject: [PATCH] Move form related content classes into _forms.py --- src/hip/__init__.py | 3 +- src/hip/_content.py | 338 +------------------------------------------- src/hip/_forms.py | 338 ++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 341 insertions(+), 338 deletions(-) create mode 100644 src/hip/_forms.py diff --git a/src/hip/__init__.py b/src/hip/__init__.py index 9e4d7bf..54b2af2 100644 --- a/src/hip/__init__.py +++ b/src/hip/__init__.py @@ -1,6 +1,7 @@ from . import backends, parsers, transports from ._client import Client -from ._content import HTML, JSON, Binary, Content, File, Files, Form, MultiPart, Text +from ._content import HTML, JSON, Binary, Content, Text +from ._forms import File, Files, Form, MultiPart from ._headers import Headers from ._request import Method, Request from ._response import Response, StatusCode diff --git a/src/hip/_content.py b/src/hip/_content.py index 94457b6..852b7d0 100644 --- a/src/hip/_content.py +++ b/src/hip/_content.py @@ -1,11 +1,9 @@ import copy import json -import os import typing from ._headers import Headers, parse_opts_header -from ._streams import ByteStream, FileStream, MultiPartStream, Stream -from ._urlencode import urldecode, urlencode +from ._streams import ByteStream, Stream __all__ = [ "HTML", @@ -13,28 +11,10 @@ "Binary", "Content", "Empty", - "File", - "Files", - # Request Content Types - "Form", - "MultiPart", # Response Content Types "Text", ] -# https://github.com/nginx/nginx/blob/master/conf/mime.types -_content_types = { - ".json": "application/json", - ".js": "application/javascript", - ".txt": "text/plain", - ".html": "text/html", - ".css": "text/css", - ".png": "image/png", - ".jpeg": "image/jpeg", - ".jpg": "image/jpeg", - ".gif": "image/gif", -} - class Content: def open(self) -> Stream: @@ -53,275 +33,6 @@ def __bytes__(self): ) -class Form(typing.Mapping[str, str], Content): - """ - HTML form data, as an immutable multi-dict. - Form parameters, as a multi-dict. - """ - - def __init__( - self, - form: ( - typing.Mapping[str, str | typing.Sequence[str]] - | typing.Sequence[tuple[str, str]] - | str - | None - ) = None, - ) -> None: - d: dict[str, list[str]] = {} - - if form is None: - d = {} - elif isinstance(form, str): - d = urldecode(form) - elif isinstance(form, typing.Mapping): - # Convert dict inputs like: - # {"a": "123", "b": ["456", "789"]} - # To dict inputs where values are always lists, like: - # {"a": ["123"], "b": ["456", "789"]} - d = {k: [v] if isinstance(v, str) else list(v) for k, v in form.items()} - else: - # Convert list inputs like: - # [("a", "123"), ("a", "456"), ("b", "789")] - # To a dict representation, like: - # {"a": ["123", "456"], "b": ["789"]} - for k, v in form: - d.setdefault(k, []).append(v) - - self._dict = d - self._content = str(self).encode("ascii") - - # Content API - - def open(self) -> Stream: - return ByteStream(self._content) - - def headers(self) -> Headers: - content_length = len(self._content) - return Headers( - { - "Content-Type": "application/x-www-form-urlencoded", - "Content-Length": f"{content_length}", - } - ) - - def __bytes__(self) -> bytes: - return self._content - - # Dict operations - - def keys(self) -> typing.KeysView[str]: - return self._dict.keys() - - def values(self) -> typing.ValuesView[str]: - return {k: v[0] for k, v in self._dict.items()}.values() - - def items(self) -> typing.ItemsView[str, str]: - return {k: v[0] for k, v in self._dict.items()}.items() - - def get(self, key: str, default: typing.Any = None) -> typing.Any: - if key in self._dict: - return self._dict[key][0] - return default - - # Multi-dict operations - - def multi_items(self) -> list[tuple[str, str]]: - multi_items: list[tuple[str, str]] = [] - for k, v in self._dict.items(): - multi_items.extend([(k, i) for i in v]) - return multi_items - - def multi_dict(self) -> dict[str, list[str]]: - return {k: list(v) for k, v in self._dict.items()} - - def get_list(self, key: str) -> list[str]: - return list(self._dict.get(key, [])) - - # Update operations - - def copy_set(self, key: str, value: str) -> "Form": - d = self.multi_dict() - d[key] = [value] - return Form(d) - - def copy_append(self, key: str, value: str) -> "Form": - d = self.multi_dict() - d[key] = d.get(key, []) + [value] - return Form(d) - - def copy_remove(self, key: str) -> "Form": - d = self.multi_dict() - d.pop(key, None) - return Form(d) - - # Accessors & built-ins - - def __getitem__(self, key: str) -> str: - return self._dict[key][0] - - def __contains__(self, key: typing.Any) -> bool: - return key in self._dict - - def __iter__(self) -> typing.Iterator[str]: - return iter(self.keys()) - - def __len__(self) -> int: - return len(self._dict) - - def __bool__(self) -> bool: - return bool(self._dict) - - def __hash__(self) -> int: - return hash(str(self)) - - def __eq__(self, other: object) -> bool: - return isinstance(other, Form) and sorted(self.multi_items()) == sorted( - other.multi_items() - ) - - def __str__(self) -> str: - return urlencode(self.multi_dict()) - - def __repr__(self) -> str: - return f"
" - - -class File(Content): - """ - Wrapper class used for files in uploads and multipart requests. - """ - - def __init__(self, path: str): - self._path = path - - @property - def name(self) -> str: - return os.path.basename(self._path) - - def open(self) -> Stream: - fileobj = open(self._path, "rb") # noqa: SIM115 - return FileStream(fileobj) - - def headers(self) -> Headers: - content_length = os.path.getsize(self._path) - - _, ext = os.path.splitext(self._path) - content_type = _content_types.get(ext, "application/octet-stream") - if content_type.startswith("text/"): - content_type += "; charset='utf-8'" - - return Headers( - { - "Content-Type": f"{content_type}", - "Content-Length": f"{content_length}", - } - ) - - def __lt__(self, other: typing.Any) -> bool: - return isinstance(other, File) and other._path < self._path - - def __eq__(self, other: object) -> bool: - return isinstance(other, File) and other._path == self._path - - def __repr__(self) -> str: - return f"" - - -class Files(typing.Mapping[str, File], Content): - """ - File parameters, as a multi-dict. - """ - - def __init__( - self, - files: ( - typing.Mapping[str, File | typing.Sequence[File]] - | typing.Sequence[tuple[str, File]] - | None - ) = None, - boundary: str = "", - ) -> None: - d: dict[str, list[File]] = {} - - if files is None: - d = {} - elif isinstance(files, typing.Mapping): - d = {k: [v] if isinstance(v, File) else list(v) for k, v in files.items()} - else: - d = {} - for k, v in files: - d.setdefault(k, []).append(v) - - self._dict = d - self._boundary = boundary or os.urandom(16).hex() - - # Standard dict interface - def keys(self) -> typing.KeysView[str]: - return self._dict.keys() - - def values(self) -> typing.ValuesView[File]: - return {k: v[0] for k, v in self._dict.items()}.values() - - def items(self) -> typing.ItemsView[str, File]: - return {k: v[0] for k, v in self._dict.items()}.items() - - def get(self, key: str, default: typing.Any = None) -> typing.Any: - if key in self._dict: - return self._dict[key][0] - return None - - # Multi dict interface - def multi_items(self) -> list[tuple[str, File]]: - multi_items: list[tuple[str, File]] = [] - for k, v in self._dict.items(): - multi_items.extend([(k, i) for i in v]) - return multi_items - - def multi_dict(self) -> dict[str, list[File]]: - return {k: list(v) for k, v in self._dict.items()} - - def get_list(self, key: str) -> list[File]: - return list(self._dict.get(key, [])) - - # Content interface - def open(self) -> Stream: - return MultiPart(files=self, boundary=self._boundary).open() - - def headers(self) -> Headers: - content_type = f"multipart/form-data; boundary={self._boundary}" - return Headers( - { - "Content-Type": f"{content_type}", - "Transfer-Encoding": "chunked", - } - ) - - # Builtins - def __getitem__(self, key: str) -> File: - return self._dict[key][0] - - def __contains__(self, key: typing.Any) -> bool: - return key in self._dict - - def __iter__(self) -> typing.Iterator[str]: - return iter(self.keys()) - - def __len__(self) -> int: - return len(self._dict) - - def __bool__(self) -> bool: - return bool(self._dict) - - def __eq__(self, other: object) -> bool: - return isinstance(other, Files) and sorted(self.multi_items()) == sorted( - other.multi_items() - ) - - def __repr__(self) -> str: - return f"" - - class JSON(Content): def __init__(self, data: typing.Any, source: bytes | None = None) -> None: self._data = data @@ -491,53 +202,6 @@ def __repr__(self) -> str: return "" -class MultiPart(Content): - def __init__( - self, - form: ( - Form - | typing.Mapping[str, str | typing.Sequence[str]] - | typing.Sequence[tuple[str, str]] - | str - | None - ) = None, - files: ( - Files - | typing.Mapping[str, File | typing.Sequence[File]] - | typing.Sequence[tuple[str, File]] - | None - ) = None, - boundary: str | None = None, - ): - self._form = form if isinstance(form, Form) else Form(form) - self._files = files if isinstance(files, Files) else Files(files) - self._boundary = os.urandom(16).hex() if boundary is None else boundary - - @property - def form(self) -> Form: - return self._form - - @property - def files(self) -> Files: - return self._files - - def open(self) -> Stream: - form = [(key, value) for key, value in self._form.items()] - files = [(key, file._path) for key, file in self._files.items()] - return MultiPartStream(form, files, boundary=self._boundary) - - def headers(self) -> Headers: - return Headers( - { - "Content-Type": f"multipart/form-data; boundary={self._boundary}", - "Transfer-Encoding": "chunked", - } - ) - - def __repr__(self) -> str: - return f"" - - def select_media(choices: dict[str, type[Content]], content_type: str) -> type[Content]: # Eg. multipart/form-data; boundary=398hjiun98jhhi87g98h76g8 ct, _ = parse_opts_header(content_type) diff --git a/src/hip/_forms.py b/src/hip/_forms.py new file mode 100644 index 0000000..c444fe9 --- /dev/null +++ b/src/hip/_forms.py @@ -0,0 +1,338 @@ +import os +import typing + +from ._content import Content +from ._headers import Headers +from ._streams import ByteStream, FileStream, MultiPartStream, Stream +from ._urlencode import urldecode, urlencode + +__all__ = ["File", "Files", "Form", "MultiPart"] + +# https://github.com/nginx/nginx/blob/master/conf/mime.types +_content_types = { + ".json": "application/json", + ".js": "application/javascript", + ".txt": "text/plain", + ".html": "text/html", + ".css": "text/css", + ".png": "image/png", + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".gif": "image/gif", +} + + +class Form(typing.Mapping[str, str], Content): + """ + HTML form data, as an immutable multi-dict. + Form parameters, as a multi-dict. + """ + + def __init__( + self, + form: ( + typing.Mapping[str, str | typing.Sequence[str]] + | typing.Sequence[tuple[str, str]] + | str + | None + ) = None, + ) -> None: + d: dict[str, list[str]] = {} + + if form is None: + d = {} + elif isinstance(form, str): + d = urldecode(form) + elif isinstance(form, typing.Mapping): + # Convert dict inputs like: + # {"a": "123", "b": ["456", "789"]} + # To dict inputs where values are always lists, like: + # {"a": ["123"], "b": ["456", "789"]} + d = {k: [v] if isinstance(v, str) else list(v) for k, v in form.items()} + else: + # Convert list inputs like: + # [("a", "123"), ("a", "456"), ("b", "789")] + # To a dict representation, like: + # {"a": ["123", "456"], "b": ["789"]} + for k, v in form: + d.setdefault(k, []).append(v) + + self._dict = d + self._content = str(self).encode("ascii") + + # Content API + + def open(self) -> Stream: + return ByteStream(self._content) + + def headers(self) -> Headers: + content_length = len(self._content) + return Headers( + { + "Content-Type": "application/x-www-form-urlencoded", + "Content-Length": f"{content_length}", + } + ) + + def __bytes__(self) -> bytes: + return self._content + + # Dict operations + + def keys(self) -> typing.KeysView[str]: + return self._dict.keys() + + def values(self) -> typing.ValuesView[str]: + return {k: v[0] for k, v in self._dict.items()}.values() + + def items(self) -> typing.ItemsView[str, str]: + return {k: v[0] for k, v in self._dict.items()}.items() + + def get(self, key: str, default: typing.Any = None) -> typing.Any: + if key in self._dict: + return self._dict[key][0] + return default + + # Multi-dict operations + + def multi_items(self) -> list[tuple[str, str]]: + multi_items: list[tuple[str, str]] = [] + for k, v in self._dict.items(): + multi_items.extend([(k, i) for i in v]) + return multi_items + + def multi_dict(self) -> dict[str, list[str]]: + return {k: list(v) for k, v in self._dict.items()} + + def get_list(self, key: str) -> list[str]: + return list(self._dict.get(key, [])) + + # Update operations + + def copy_set(self, key: str, value: str) -> "Form": + d = self.multi_dict() + d[key] = [value] + return Form(d) + + def copy_append(self, key: str, value: str) -> "Form": + d = self.multi_dict() + d[key] = d.get(key, []) + [value] + return Form(d) + + def copy_remove(self, key: str) -> "Form": + d = self.multi_dict() + d.pop(key, None) + return Form(d) + + # Accessors & built-ins + + def __getitem__(self, key: str) -> str: + return self._dict[key][0] + + def __contains__(self, key: typing.Any) -> bool: + return key in self._dict + + def __iter__(self) -> typing.Iterator[str]: + return iter(self.keys()) + + def __len__(self) -> int: + return len(self._dict) + + def __bool__(self) -> bool: + return bool(self._dict) + + def __hash__(self) -> int: + return hash(str(self)) + + def __eq__(self, other: object) -> bool: + return isinstance(other, Form) and sorted(self.multi_items()) == sorted( + other.multi_items() + ) + + def __str__(self) -> str: + return urlencode(self.multi_dict()) + + def __repr__(self) -> str: + return f"" + + +class File(Content): + """ + Wrapper class used for files in uploads and multipart requests. + """ + + def __init__(self, path: str): + self._path = path + + @property + def name(self) -> str: + return os.path.basename(self._path) + + def open(self) -> Stream: + fileobj = open(self._path, "rb") # noqa: SIM115 + return FileStream(fileobj) + + def headers(self) -> Headers: + content_length = os.path.getsize(self._path) + + _, ext = os.path.splitext(self._path) + content_type = _content_types.get(ext, "application/octet-stream") + if content_type.startswith("text/"): + content_type += "; charset='utf-8'" + + return Headers( + { + "Content-Type": f"{content_type}", + "Content-Length": f"{content_length}", + } + ) + + def __lt__(self, other: typing.Any) -> bool: + return isinstance(other, File) and other._path < self._path + + def __eq__(self, other: object) -> bool: + return isinstance(other, File) and other._path == self._path + + def __repr__(self) -> str: + return f"" + + +class Files(typing.Mapping[str, File], Content): + """ + File parameters, as a multi-dict. + """ + + def __init__( + self, + files: ( + typing.Mapping[str, File | typing.Sequence[File]] + | typing.Sequence[tuple[str, File]] + | None + ) = None, + boundary: str = "", + ) -> None: + d: dict[str, list[File]] = {} + + if files is None: + d = {} + elif isinstance(files, typing.Mapping): + d = {k: [v] if isinstance(v, File) else list(v) for k, v in files.items()} + else: + d = {} + for k, v in files: + d.setdefault(k, []).append(v) + + self._dict = d + self._boundary = boundary or os.urandom(16).hex() + + # Standard dict interface + def keys(self) -> typing.KeysView[str]: + return self._dict.keys() + + def values(self) -> typing.ValuesView[File]: + return {k: v[0] for k, v in self._dict.items()}.values() + + def items(self) -> typing.ItemsView[str, File]: + return {k: v[0] for k, v in self._dict.items()}.items() + + def get(self, key: str, default: typing.Any = None) -> typing.Any: + if key in self._dict: + return self._dict[key][0] + return None + + # Multi dict interface + def multi_items(self) -> list[tuple[str, File]]: + multi_items: list[tuple[str, File]] = [] + for k, v in self._dict.items(): + multi_items.extend([(k, i) for i in v]) + return multi_items + + def multi_dict(self) -> dict[str, list[File]]: + return {k: list(v) for k, v in self._dict.items()} + + def get_list(self, key: str) -> list[File]: + return list(self._dict.get(key, [])) + + # Content interface + def open(self) -> Stream: + return MultiPart(files=self, boundary=self._boundary).open() + + def headers(self) -> Headers: + content_type = f"multipart/form-data; boundary={self._boundary}" + return Headers( + { + "Content-Type": f"{content_type}", + "Transfer-Encoding": "chunked", + } + ) + + # Builtins + def __getitem__(self, key: str) -> File: + return self._dict[key][0] + + def __contains__(self, key: typing.Any) -> bool: + return key in self._dict + + def __iter__(self) -> typing.Iterator[str]: + return iter(self.keys()) + + def __len__(self) -> int: + return len(self._dict) + + def __bool__(self) -> bool: + return bool(self._dict) + + def __eq__(self, other: object) -> bool: + return isinstance(other, Files) and sorted(self.multi_items()) == sorted( + other.multi_items() + ) + + def __repr__(self) -> str: + return f"" + + +class MultiPart(Content): + def __init__( + self, + form: ( + Form + | typing.Mapping[str, str | typing.Sequence[str]] + | typing.Sequence[tuple[str, str]] + | str + | None + ) = None, + files: ( + Files + | typing.Mapping[str, File | typing.Sequence[File]] + | typing.Sequence[tuple[str, File]] + | None + ) = None, + boundary: str | None = None, + ): + self._form = form if isinstance(form, Form) else Form(form) + self._files = files if isinstance(files, Files) else Files(files) + self._boundary = os.urandom(16).hex() if boundary is None else boundary + + @property + def form(self) -> Form: + return self._form + + @property + def files(self) -> Files: + return self._files + + def open(self) -> Stream: + form = [(key, value) for key, value in self._form.items()] + files = [(key, file._path) for key, file in self._files.items()] + return MultiPartStream(form, files, boundary=self._boundary) + + def headers(self) -> Headers: + return Headers( + { + "Content-Type": f"multipart/form-data; boundary={self._boundary}", + "Transfer-Encoding": "chunked", + } + ) + + def __repr__(self) -> str: + return f""