X7ROOT File Manager
Current Path:
/opt/cloudlinux/venv/lib/python3.11/site-packages/aiohttp
opt
/
cloudlinux
/
venv
/
lib
/
python3.11
/
site-packages
/
aiohttp
/
ðŸ“
..
ðŸ“
.hash
📄
__init__.py
(7.58 KB)
ðŸ“
__pycache__
📄
_cparser.pxd
(4.22 KB)
📄
_find_header.pxd
(68 B)
📄
_headers.pxi
(1.96 KB)
📄
_helpers.cpython-311-x86_64-linux-gnu.so
(86.01 KB)
📄
_helpers.pyi
(202 B)
📄
_helpers.pyx
(1.02 KB)
📄
_http_parser.cpython-311-x86_64-linux-gnu.so
(569.59 KB)
📄
_http_parser.pyx
(27.4 KB)
📄
_http_writer.cpython-311-x86_64-linux-gnu.so
(75.01 KB)
📄
_http_writer.pyx
(4.47 KB)
📄
_websocket.cpython-311-x86_64-linux-gnu.so
(53.37 KB)
📄
_websocket.pyx
(1.52 KB)
📄
abc.py
(5.37 KB)
📄
base_protocol.py
(2.68 KB)
📄
client.py
(46.17 KB)
📄
client_exceptions.py
(9.19 KB)
📄
client_proto.py
(8.45 KB)
📄
client_reqrep.py
(38.75 KB)
📄
client_ws.py
(10.75 KB)
📄
compression_utils.py
(4.9 KB)
📄
connector.py
(51.56 KB)
📄
cookiejar.py
(13.69 KB)
📄
formdata.py
(5.96 KB)
📄
hdrs.py
(4.5 KB)
📄
helpers.py
(29.55 KB)
📄
http.py
(1.8 KB)
📄
http_exceptions.py
(2.65 KB)
📄
http_parser.py
(34.66 KB)
📄
http_websocket.py
(26.09 KB)
📄
http_writer.py
(5.79 KB)
📄
locks.py
(1.11 KB)
📄
log.py
(325 B)
📄
multipart.py
(31.71 KB)
📄
payload.py
(13.22 KB)
📄
payload_streamer.py
(2.04 KB)
📄
py.typed
(7 B)
📄
pytest_plugin.py
(11.33 KB)
📄
resolver.py
(4.95 KB)
📄
streams.py
(20.35 KB)
📄
tcp_helpers.py
(961 B)
📄
test_utils.py
(19.71 KB)
📄
tracing.py
(14.78 KB)
📄
typedefs.py
(1.44 KB)
📄
web.py
(18.81 KB)
📄
web_app.py
(17.88 KB)
📄
web_exceptions.py
(10.12 KB)
📄
web_fileresponse.py
(11.15 KB)
📄
web_log.py
(7.62 KB)
📄
web_middlewares.py
(3.94 KB)
📄
web_protocol.py
(22.5 KB)
📄
web_request.py
(28.08 KB)
📄
web_response.py
(27.08 KB)
📄
web_routedef.py
(5.99 KB)
📄
web_runner.py
(11.46 KB)
📄
web_server.py
(2.53 KB)
📄
web_urldispatcher.py
(39.12 KB)
📄
web_ws.py
(18.21 KB)
📄
worker.py
(7.78 KB)
Editing: 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()
Upload File
Create Folder