From 969dad4ca34e2e51cb1c6c2d4f71b8142f6af13e Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:54:45 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/uvicorn/protocols=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../uvicorn/protocols/__init__.py | Bin 0 -> 1024 bytes .../site-packages/uvicorn/protocols/utils.py | 62 ++++++++++++++++++ 2 files changed, 62 insertions(+) create mode 100644 venv/Lib/site-packages/uvicorn/protocols/__init__.py create mode 100644 venv/Lib/site-packages/uvicorn/protocols/utils.py diff --git a/venv/Lib/site-packages/uvicorn/protocols/__init__.py b/venv/Lib/site-packages/uvicorn/protocols/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..06d7405020018ddf3cacee90fd4af10487da3d20 GIT binary patch literal 1024 ScmZQz7zLvtFd70QH3R?z00031 literal 0 HcmV?d00001 diff --git a/venv/Lib/site-packages/uvicorn/protocols/utils.py b/venv/Lib/site-packages/uvicorn/protocols/utils.py new file mode 100644 index 0000000..a9a8906 --- /dev/null +++ b/venv/Lib/site-packages/uvicorn/protocols/utils.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import asyncio +import socket +import urllib.parse + +from uvicorn._types import WWWScope + + +class ClientDisconnected(OSError): ... + + +def get_remote_addr(transport: asyncio.Transport) -> tuple[str, int] | None: + socket_info: socket.socket | None = transport.get_extra_info("socket") + if socket_info is not None: + try: + info = socket_info.getpeername() + return (str(info[0]), int(info[1])) if isinstance(info, tuple) else None + except OSError: # pragma: no cover + # This case appears to inconsistently occur with uvloop + # bound to a unix domain socket. + return None + + info = transport.get_extra_info("peername") + if info is not None and isinstance(info, list | tuple) and len(info) == 2: + return (str(info[0]), int(info[1])) + return None + + +def get_local_addr(transport: asyncio.Transport) -> tuple[str, int | None] | None: + socket_info: socket.socket | None = transport.get_extra_info("socket") + if socket_info is not None: + info = socket_info.getsockname() + if isinstance(info, tuple): + return (str(info[0]), int(info[1])) + if isinstance(info, str): + return (info, None) + return None + info = transport.get_extra_info("sockname") + if info is not None and isinstance(info, list | tuple) and len(info) == 2: + return (str(info[0]), int(info[1])) + if isinstance(info, str): + return (info, None) + return None + + +def is_ssl(transport: asyncio.Transport) -> bool: + return bool(transport.get_extra_info("sslcontext")) + + +def get_client_addr(scope: WWWScope) -> str: + client = scope.get("client") + if not client: + return "" + return "%s:%d" % client + + +def get_path_with_query_string(scope: WWWScope) -> str: + path_with_query_string = urllib.parse.quote(scope["path"]) + if scope["query_string"]: + path_with_query_string = "{}?{}".format(path_with_query_string, scope["query_string"].decode("ascii")) + return path_with_query_string