0
Self Samples
8
Cumulative
1
Lines Hit
0.03%
% of Total
0
Max Self
8
Max Total
Intensity:
Cold
Hot
Self Time
Total Time
Show All
Hot Only
Heat
Specialization
Line
Self
Total
Code
1
from __future__ import annotations
2
3
import functools
4
import re
5
import sys
6
from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
7
from contextlib import AbstractAsyncContextManager, asynccontextmanager
8
from dataclasses import dataclass
9
from ipaddress import AddressValueError, IPv6Address
10
from typing import Any, Generic, Protocol, TypeVar, overload
11
12
import anyio.abc
13
14
from starlette.types import Scope
15
16
if sys.version_info >= (3, 13): # pragma: no cover
17
from inspect import iscoroutinefunction
18
from typing import TypeIs
19
else: # pragma: no cover
20
from asyncio import iscoroutinefunction
21
22
from typing_extensions import TypeIs
23
24
if sys.version_info < (3, 11): # pragma: no cover
25
from exceptiongroup import BaseExceptionGroup
26
27
28
T = TypeVar("T")
29
AwaitableCallable = Callable[..., Awaitable[T]]
30
31
# Reject characters that could make a Host header change the URL path or authority.
32
_HOST_RE = re.compile(
33
r"^(?P<host>[a-z0-9._~%!$&'()*+,;=-]+|\[(?:(?P<ipv6>[a-f0-9]*:[a-f0-9.:]+)|"
34
r"(?-i:v)[a-f0-9]+\.[a-z0-9._~!$&'()*+,;=:-]+)\])(?::(?P<port>[0-9]+))?$",
35
re.IGNORECASE,
36
)
37
38
39
@dataclass(frozen=True, slots=True)
40
class ParsedHost:
41
host: str
42
port: str | None
43
44
@property
45
def authority(self) -> str:
46
return self.host if self.port is None else f"{self.host}:{self.port}"
47
48
@property
49
def is_valid_port(self) -> bool:
50
if self.port is None:
51
return True
52
port = self.port.lstrip("0")
53
return len(port) <= 5 and int(port or "0") <= 65535
54
55
56
@overload
57
def is_async_callable(obj: AwaitableCallable[T]) -> TypeIs[AwaitableCallable[T]]: ...
58
59
60
@overload
61
def is_async_callable(obj: Any) -> TypeIs[AwaitableCallable[Any]]: ...
62
63
64
def is_async_callable(obj: Any) -> Any:
65
while isinstance(obj, functools.partial):
66
obj = obj.func
67
68
return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))
69
70
71
T_co = TypeVar("T_co", covariant=True)
72
73
74
class AwaitableOrContextManager(
75
Awaitable[T_co], AbstractAsyncContextManager[T_co], Protocol[T_co]
76
): ... # pragma: no branch
77
78
79
class SupportsAsyncClose(Protocol):
80
async def close(self) -> None: ... # pragma: no cover
81
82
83
SupportsAsyncCloseType = TypeVar("SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False)
84
85
86
class AwaitableOrContextManagerWrapper(Generic[SupportsAsyncCloseType]):
87
__slots__ = ("aw", "entered")
88
89
def __init__(self, aw: Awaitable[SupportsAsyncCloseType]) -> None:
90
self.aw = aw
91
92
def __await__(self) -> Generator[Any, None, SupportsAsyncCloseType]:
93
return self.aw.__await__()
94
95
async def __aenter__(self) -> SupportsAsyncCloseType:
96
8
self.entered = await self.aw
97
return self.entered
98
99
async def __aexit__(self, *args: Any) -> None | bool:
100
await self.entered.close()
101
return None
102
103
104
@asynccontextmanager
105
async def create_collapsing_task_group() -> AsyncGenerator[anyio.abc.TaskGroup, None]:
106
try:
107
async with anyio.create_task_group() as tg:
108
yield tg
109
except BaseExceptionGroup as excs:
110
if len(excs.exceptions) != 1:
111
raise
112
113
exc = excs.exceptions[0]
114
context = None if exc.__suppress_context__ else exc.__context__
115
raise exc from exc.__cause__ or context
116
117
118
def parse_host_header(host_header: str | None) -> ParsedHost | None:
119
"""Parse `host_header` into its host and port components.
120
121
The host preserves brackets around IP literals. Invalid headers produce `None`.
122
"""
123
if host_header is None:
124
return None
125
126
match = _HOST_RE.fullmatch(host_header)
127
if match is None:
128
return None
129
130
ipv6 = match["ipv6"]
131
if ipv6 is not None:
132
try:
133
IPv6Address(ipv6)
134
except AddressValueError:
135
return None
136
137
return ParsedHost(match["host"], match["port"])
138
139
140
def get_route_path(scope: Scope) -> str:
141
path: str = scope["path"]
142
root_path = scope.get("root_path", "")
143
if not root_path:
144
return path
145
146
if not path.startswith(root_path):
147
return path
148
149
if path == root_path:
150
return ""
151
152
if path[len(root_path)] == "/":
153
return path[len(root_path) :]
154
155
return path