golden hour
/opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp
⬆️ Go Up
Upload
File/Folder
Size
Actions
.hash
-
Del
OK
__init__.py
7.58 KB
Del
OK
__pycache__
-
Del
OK
_cparser.pxd
4.22 KB
Del
OK
_find_header.pxd
68 B
Del
OK
_headers.pxi
1.96 KB
Del
OK
_helpers.cpython-311-x86_64-linux-gnu.so
563.5 KB
Del
OK
_helpers.pyi
202 B
Del
OK
_helpers.pyx
1.02 KB
Del
OK
_http_parser.cpython-311-x86_64-linux-gnu.so
2.66 MB
Del
OK
_http_parser.pyx
27.4 KB
Del
OK
_http_writer.cpython-311-x86_64-linux-gnu.so
477.98 KB
Del
OK
_http_writer.pyx
4.47 KB
Del
OK
_websocket.cpython-311-x86_64-linux-gnu.so
249.84 KB
Del
OK
_websocket.pyx
1.52 KB
Del
OK
abc.py
5.37 KB
Del
OK
base_protocol.py
2.68 KB
Del
OK
client.py
46.17 KB
Del
OK
client_exceptions.py
9.19 KB
Del
OK
client_proto.py
8.45 KB
Del
OK
client_reqrep.py
38.75 KB
Del
OK
client_ws.py
10.75 KB
Del
OK
compression_utils.py
4.9 KB
Del
OK
connector.py
51.56 KB
Del
OK
cookiejar.py
13.69 KB
Del
OK
formdata.py
5.96 KB
Del
OK
hdrs.py
4.5 KB
Del
OK
helpers.py
29.55 KB
Del
OK
http.py
1.8 KB
Del
OK
http_exceptions.py
2.65 KB
Del
OK
http_parser.py
34.66 KB
Del
OK
http_websocket.py
26.09 KB
Del
OK
http_writer.py
5.79 KB
Del
OK
locks.py
1.11 KB
Del
OK
log.py
325 B
Del
OK
multipart.py
31.71 KB
Del
OK
payload.py
13.22 KB
Del
OK
payload_streamer.py
2.04 KB
Del
OK
py.typed
7 B
Del
OK
pytest_plugin.py
11.33 KB
Del
OK
resolver.py
4.95 KB
Del
OK
streams.py
20.35 KB
Del
OK
tcp_helpers.py
961 B
Del
OK
test_utils.py
19.71 KB
Del
OK
tracing.py
14.78 KB
Del
OK
typedefs.py
1.44 KB
Del
OK
web.py
18.81 KB
Del
OK
web_app.py
17.88 KB
Del
OK
web_exceptions.py
10.12 KB
Del
OK
web_fileresponse.py
11.15 KB
Del
OK
web_log.py
7.62 KB
Del
OK
web_middlewares.py
3.94 KB
Del
OK
web_protocol.py
22.5 KB
Del
OK
web_request.py
28.08 KB
Del
OK
web_response.py
27.08 KB
Del
OK
web_routedef.py
5.99 KB
Del
OK
web_runner.py
11.46 KB
Del
OK
web_server.py
2.53 KB
Del
OK
web_urldispatcher.py
39.12 KB
Del
OK
web_ws.py
18.21 KB
Del
OK
worker.py
7.78 KB
Del
OK
Edit: formdata.py
import io from typing import Any, Iterable, List, Optional from urllib.parse import urlencode from multidict import MultiDict, MultiDictProxy from . import hdrs, multipart, payload from .helpers import guess_filename from .payload import Payload __all__ = ("FormData",) class FormData: """Helper class for form body generation. Supports multipart/form-data and application/x-www-form-urlencoded. """ def __init__( self, fields: Iterable[Any] = (), quote_fields: bool = True, charset: Optional[str] = None, ) -> None: self._writer = multipart.MultipartWriter("form-data") self._fields: List[Any] = [] self._is_multipart = False self._is_processed = False self._quote_fields = quote_fields self._charset = charset if isinstance(fields, dict): fields = list(fields.items()) elif not isinstance(fields, (list, tuple)): fields = (fields,) self.add_fields(*fields) @property def is_multipart(self) -> bool: return self._is_multipart def add_field( self, name: str, value: Any, *, content_type: Optional[str] = None, filename: Optional[str] = None, content_transfer_encoding: Optional[str] = None, ) -> None: if isinstance(value, io.IOBase): self._is_multipart = True elif isinstance(value, (bytes, bytearray, memoryview)): if filename is None and content_transfer_encoding is None: filename = name type_options: MultiDict[str] = MultiDict({"name": name}) if filename is not None and not isinstance(filename, str): raise TypeError( "filename must be an instance of str. " "Got: %s" % filename ) if filename is None and isinstance(value, io.IOBase): filename = guess_filename(value, name) if filename is not None: type_options["filename"] = filename self._is_multipart = True headers = {} if content_type is not None: if not isinstance(content_type, str): raise TypeError( "content_type must be an instance of str. " "Got: %s" % content_type ) headers[hdrs.CONTENT_TYPE] = content_type self._is_multipart = True if content_transfer_encoding is not None: if not isinstance(content_transfer_encoding, str): raise TypeError( "content_transfer_encoding must be an instance" " of str. Got: %s" % content_transfer_encoding ) headers[hdrs.CONTENT_TRANSFER_ENCODING] = content_transfer_encoding self._is_multipart = True self._fields.append((type_options, headers, value)) def add_fields(self, *fields: Any) -> None: to_add = list(fields) while to_add: rec = to_add.pop(0) if isinstance(rec, io.IOBase): k = guess_filename(rec, "unknown") self.add_field(k, rec) # type: ignore[arg-type] elif isinstance(rec, (MultiDictProxy, MultiDict)): to_add.extend(rec.items()) elif isinstance(rec, (list, tuple)) and len(rec) == 2: k, fp = rec self.add_field(k, fp) # type: ignore[arg-type] else: raise TypeError( "Only io.IOBase, multidict and (name, file) " "pairs allowed, use .add_field() for passing " "more complex parameters, got {!r}".format(rec) ) def _gen_form_urlencoded(self) -> payload.BytesPayload: # form data (x-www-form-urlencoded) data = [] for type_options, _, value in self._fields: data.append((type_options["name"], value)) charset = self._charset if self._charset is not None else "utf-8" if charset == "utf-8": content_type = "application/x-www-form-urlencoded" else: content_type = "application/x-www-form-urlencoded; " "charset=%s" % charset return payload.BytesPayload( urlencode(data, doseq=True, encoding=charset).encode(), content_type=content_type, ) def _gen_form_data(self) -> multipart.MultipartWriter: """Encode a list of fields using the multipart/form-data MIME format""" if self._is_processed: raise RuntimeError("Form data has been processed already") for dispparams, headers, value in self._fields: try: if hdrs.CONTENT_TYPE in headers: part = payload.get_payload( value, content_type=headers[hdrs.CONTENT_TYPE], headers=headers, encoding=self._charset, ) else: part = payload.get_payload( value, headers=headers, encoding=self._charset ) except Exception as exc: raise TypeError( "Can not serialize value type: %r\n " "headers: %r\n value: %r" % (type(value), headers, value) ) from exc if dispparams: part.set_content_disposition( "form-data", quote_fields=self._quote_fields, **dispparams ) # FIXME cgi.FieldStorage doesn't likes body parts with # Content-Length which were sent via chunked transfer encoding assert part.headers is not None part.headers.popall(hdrs.CONTENT_LENGTH, None) self._writer.append_payload(part) self._is_processed = True return self._writer def __call__(self) -> Payload: if self._is_multipart: return self._gen_form_data() else: return self._gen_form_urlencoded()
Save