Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3from collections.abc import ItemsView, Iterable, Iterator, KeysView, Mapping, MutableMapping, Sequence, ValuesView
4from shlex import shlex
5from typing import (
6 Any,
7 BinaryIO,
8 NamedTuple,
9 TypeVar,
10 Union,
11 cast,
12)
13from urllib.parse import SplitResult, parse_qsl, urlencode, urlsplit
15from starlette.concurrency import run_in_threadpool
16from starlette.types import Scope
19class Address(NamedTuple):
20 host: str
21 port: int
24_KeyType = TypeVar("_KeyType")
25# Mapping keys are invariant but their values are covariant since
26# you can only read them
27# that is, you can't do `Mapping[str, Animal]()["fido"] = Dog()`
28_CovariantValueType = TypeVar("_CovariantValueType", covariant=True)
31class URL:
32 def __init__(
33 self,
34 url: str = "",
35 scope: Scope | None = None,
36 **components: Any,
37 ) -> None:
38 if scope is not None:
39 assert not url, 'Cannot set both "url" and "scope".'
40 assert not components, 'Cannot set both "scope" and "**components".'
41 scheme = scope.get("scheme", "http")
42 server = scope.get("server", None)
43 path = scope["path"]
44 query_string = scope.get("query_string", b"")
46 host_header = None
47 for key, value in scope["headers"]:
48 if key == b"host":
49 host_header = value.decode("latin-1")
50 break
52 if host_header is not None:
53 url = f"{scheme}://{host_header}{path}"
54 elif server is None:
55 url = path
56 else:
57 host, port = server
58 default_port = {"http": 80, "https": 443, "ws": 80, "wss": 443}[scheme]
59 if port == default_port:
60 url = f"{scheme}://{host}{path}"
61 else:
62 url = f"{scheme}://{host}:{port}{path}"
64 if query_string:
65 url += "?" + query_string.decode()
66 elif components:
67 assert not url, 'Cannot set both "url" and "**components".'
68 url = URL("").replace(**components).components.geturl()
70 self._url = url
72 @property
73 def components(self) -> SplitResult:
74 if not hasattr(self, "_components"):
75 self._components = urlsplit(self._url)
76 return self._components
78 @property
79 def scheme(self) -> str:
80 return self.components.scheme
82 @property
83 def netloc(self) -> str:
84 return self.components.netloc
86 @property
87 def path(self) -> str:
88 return self.components.path
90 @property
91 def query(self) -> str:
92 return self.components.query
94 @property
95 def fragment(self) -> str:
96 return self.components.fragment
98 @property
99 def username(self) -> None | str:
100 return self.components.username
102 @property
103 def password(self) -> None | str:
104 return self.components.password
106 @property
107 def hostname(self) -> None | str:
108 return self.components.hostname
110 @property
111 def port(self) -> int | None:
112 return self.components.port
114 @property
115 def is_secure(self) -> bool:
116 return self.scheme in ("https", "wss")
118 def replace(self, **kwargs: Any) -> URL:
119 if "username" in kwargs or "password" in kwargs or "hostname" in kwargs or "port" in kwargs:
120 hostname = kwargs.pop("hostname", None)
121 port = kwargs.pop("port", self.port)
122 username = kwargs.pop("username", self.username)
123 password = kwargs.pop("password", self.password)
125 if hostname is None:
126 netloc = self.netloc
127 _, _, hostname = netloc.rpartition("@")
129 if hostname[-1] != "]":
130 hostname = hostname.rsplit(":", 1)[0]
132 netloc = hostname
133 if port is not None:
134 netloc += f":{port}"
135 if username is not None:
136 userpass = username
137 if password is not None:
138 userpass += f":{password}"
139 netloc = f"{userpass}@{netloc}"
141 kwargs["netloc"] = netloc
143 components = self.components._replace(**kwargs)
144 return self.__class__(components.geturl())
146 def include_query_params(self, **kwargs: Any) -> URL:
147 params = MultiDict(parse_qsl(self.query, keep_blank_values=True))
148 params.update({str(key): str(value) for key, value in kwargs.items()})
149 query = urlencode(params.multi_items())
150 return self.replace(query=query)
152 def replace_query_params(self, **kwargs: Any) -> URL:
153 query = urlencode([(str(key), str(value)) for key, value in kwargs.items()])
154 return self.replace(query=query)
156 def remove_query_params(self, keys: str | Sequence[str]) -> URL:
157 if isinstance(keys, str):
158 keys = [keys]
159 params = MultiDict(parse_qsl(self.query, keep_blank_values=True))
160 for key in keys:
161 params.pop(key, None)
162 query = urlencode(params.multi_items())
163 return self.replace(query=query)
165 def __eq__(self, other: Any) -> bool:
166 return str(self) == str(other)
168 def __str__(self) -> str:
169 return self._url
171 def __repr__(self) -> str:
172 url = str(self)
173 if self.password:
174 url = str(self.replace(password="********"))
175 return f"{self.__class__.__name__}({repr(url)})"
178class URLPath(str):
179 """
180 A URL path string that may also hold an associated protocol and/or host.
181 Used by the routing to return `url_path_for` matches.
182 """
184 def __new__(cls, path: str, protocol: str = "", host: str = "") -> URLPath:
185 assert protocol in ("http", "websocket", "")
186 return str.__new__(cls, path)
188 def __init__(self, path: str, protocol: str = "", host: str = "") -> None:
189 self.protocol = protocol
190 self.host = host
192 def make_absolute_url(self, base_url: str | URL) -> URL:
193 if isinstance(base_url, str):
194 base_url = URL(base_url)
195 if self.protocol:
196 scheme = {
197 "http": {True: "https", False: "http"},
198 "websocket": {True: "wss", False: "ws"},
199 }[self.protocol][base_url.is_secure]
200 else:
201 scheme = base_url.scheme
203 netloc = self.host or base_url.netloc
204 path = base_url.path.rstrip("/") + str(self)
205 return URL(scheme=scheme, netloc=netloc, path=path)
208class Secret:
209 """
210 Holds a string value that should not be revealed in tracebacks etc.
211 You should cast the value to `str` at the point it is required.
212 """
214 def __init__(self, value: str):
215 self._value = value
217 def __repr__(self) -> str:
218 class_name = self.__class__.__name__
219 return f"{class_name}('**********')"
221 def __str__(self) -> str:
222 return self._value
224 def __bool__(self) -> bool:
225 return bool(self._value)
228class CommaSeparatedStrings(Sequence[str]):
229 def __init__(self, value: str | Sequence[str]):
230 if isinstance(value, str):
231 splitter = shlex(value, posix=True)
232 splitter.whitespace = ","
233 splitter.whitespace_split = True
234 self._items = [item.strip() for item in splitter]
235 else:
236 self._items = list(value)
238 def __len__(self) -> int:
239 return len(self._items)
241 def __getitem__(self, index: int | slice) -> Any:
242 return self._items[index]
244 def __iter__(self) -> Iterator[str]:
245 return iter(self._items)
247 def __repr__(self) -> str:
248 class_name = self.__class__.__name__
249 items = [item for item in self]
250 return f"{class_name}({items!r})"
252 def __str__(self) -> str:
253 return ", ".join(repr(item) for item in self)
256class ImmutableMultiDict(Mapping[_KeyType, _CovariantValueType]):
257 _dict: dict[_KeyType, _CovariantValueType]
259 def __init__(
260 self,
261 *args: ImmutableMultiDict[_KeyType, _CovariantValueType]
262 | Mapping[_KeyType, _CovariantValueType]
263 | Iterable[tuple[_KeyType, _CovariantValueType]],
264 **kwargs: Any,
265 ) -> None:
266 assert len(args) < 2, "Too many arguments."
268 value: Any = args[0] if args else []
269 if kwargs:
270 value = ImmutableMultiDict(value).multi_items() + ImmutableMultiDict(kwargs).multi_items()
272 if not value:
273 _items: list[tuple[Any, Any]] = []
274 elif hasattr(value, "multi_items"):
275 value = cast(ImmutableMultiDict[_KeyType, _CovariantValueType], value)
276 _items = list(value.multi_items())
277 elif hasattr(value, "items"):
278 value = cast(Mapping[_KeyType, _CovariantValueType], value)
279 _items = list(value.items())
280 else:
281 value = cast("list[tuple[Any, Any]]", value)
282 _items = list(value)
284 self._dict = {k: v for k, v in _items}
285 self._list = _items
287 def getlist(self, key: Any) -> list[_CovariantValueType]:
288 return [item_value for item_key, item_value in self._list if item_key == key]
290 def keys(self) -> KeysView[_KeyType]:
291 return self._dict.keys()
293 def values(self) -> ValuesView[_CovariantValueType]:
294 return self._dict.values()
296 def items(self) -> ItemsView[_KeyType, _CovariantValueType]:
297 return self._dict.items()
299 def multi_items(self) -> list[tuple[_KeyType, _CovariantValueType]]:
300 return list(self._list)
302 def __getitem__(self, key: _KeyType) -> _CovariantValueType:
303 return self._dict[key]
305 def __contains__(self, key: Any) -> bool:
306 return key in self._dict
308 def __iter__(self) -> Iterator[_KeyType]:
309 return iter(self.keys())
311 def __len__(self) -> int:
312 return len(self._dict)
314 def __eq__(self, other: Any) -> bool:
315 if not isinstance(other, self.__class__):
316 return False
317 return sorted(self._list) == sorted(other._list)
319 def __repr__(self) -> str:
320 class_name = self.__class__.__name__
321 items = self.multi_items()
322 return f"{class_name}({items!r})"
325class MultiDict(ImmutableMultiDict[Any, Any]):
326 def __setitem__(self, key: Any, value: Any) -> None:
327 self.setlist(key, [value])
329 def __delitem__(self, key: Any) -> None:
330 self._list = [(k, v) for k, v in self._list if k != key]
331 del self._dict[key]
333 def pop(self, key: Any, default: Any = None) -> Any:
334 self._list = [(k, v) for k, v in self._list if k != key]
335 return self._dict.pop(key, default)
337 def popitem(self) -> tuple[Any, Any]:
338 key, value = self._dict.popitem()
339 self._list = [(k, v) for k, v in self._list if k != key]
340 return key, value
342 def poplist(self, key: Any) -> list[Any]:
343 values = [v for k, v in self._list if k == key]
344 self.pop(key)
345 return values
347 def clear(self) -> None:
348 self._dict.clear()
349 self._list.clear()
351 def setdefault(self, key: Any, default: Any = None) -> Any:
352 if key not in self:
353 self._dict[key] = default
354 self._list.append((key, default))
356 return self[key]
358 def setlist(self, key: Any, values: list[Any]) -> None:
359 if not values:
360 self.pop(key, None)
361 else:
362 existing_items = [(k, v) for (k, v) in self._list if k != key]
363 self._list = existing_items + [(key, value) for value in values]
364 self._dict[key] = values[-1]
366 def append(self, key: Any, value: Any) -> None:
367 self._list.append((key, value))
368 self._dict[key] = value
370 def update(
371 self,
372 *args: MultiDict | Mapping[Any, Any] | list[tuple[Any, Any]],
373 **kwargs: Any,
374 ) -> None:
375 value = MultiDict(*args, **kwargs)
376 existing_items = [(k, v) for (k, v) in self._list if k not in value.keys()]
377 self._list = existing_items + value.multi_items()
378 self._dict.update(value)
381class QueryParams(ImmutableMultiDict[str, str]):
382 """
383 An immutable multidict.
384 """
386 def __init__(
387 self,
388 *args: ImmutableMultiDict[Any, Any] | Mapping[Any, Any] | list[tuple[Any, Any]] | str | bytes,
389 **kwargs: Any,
390 ) -> None:
391 assert len(args) < 2, "Too many arguments."
393 value = args[0] if args else []
395 if isinstance(value, str):
396 super().__init__(parse_qsl(value, keep_blank_values=True), **kwargs)
397 elif isinstance(value, bytes):
398 super().__init__(parse_qsl(value.decode("latin-1"), keep_blank_values=True), **kwargs)
399 else:
400 super().__init__(*args, **kwargs) # type: ignore[arg-type]
401 self._list = [(str(k), str(v)) for k, v in self._list]
402 self._dict = {str(k): str(v) for k, v in self._dict.items()}
404 def __str__(self) -> str:
405 return urlencode(self._list)
407 def __repr__(self) -> str:
408 class_name = self.__class__.__name__
409 query_string = str(self)
410 return f"{class_name}({query_string!r})"
413class UploadFile:
414 """
415 An uploaded file included as part of the request data.
416 """
418 def __init__(
419 self,
420 file: BinaryIO,
421 *,
422 size: int | None = None,
423 filename: str | None = None,
424 headers: Headers | None = None,
425 ) -> None:
426 self.filename = filename
427 self.file = file
428 self.size = size
429 self.headers = headers or Headers()
431 @property
432 def content_type(self) -> str | None:
433 return self.headers.get("content-type", None)
435 @property
436 def _in_memory(self) -> bool:
437 # check for SpooledTemporaryFile._rolled
438 rolled_to_disk = getattr(self.file, "_rolled", True)
439 return not rolled_to_disk
441 async def write(self, data: bytes) -> None:
442 if self.size is not None:
443 self.size += len(data)
445 if self._in_memory:
446 self.file.write(data)
447 else:
448 await run_in_threadpool(self.file.write, data)
450 async def read(self, size: int = -1) -> bytes:
451 if self._in_memory:
452 return self.file.read(size)
453 return await run_in_threadpool(self.file.read, size)
455 async def seek(self, offset: int) -> None:
456 if self._in_memory:
457 self.file.seek(offset)
458 else:
459 await run_in_threadpool(self.file.seek, offset)
461 async def close(self) -> None:
462 if self._in_memory:
463 self.file.close()
464 else:
465 await run_in_threadpool(self.file.close)
467 def __repr__(self) -> str:
468 return f"{self.__class__.__name__}(filename={self.filename!r}, size={self.size!r}, headers={self.headers!r})"
471class FormData(ImmutableMultiDict[str, Union[UploadFile, str]]):
472 """
473 An immutable multidict, containing both file uploads and text input.
474 """
476 def __init__(
477 self,
478 *args: FormData | Mapping[str, str | UploadFile] | list[tuple[str, str | UploadFile]],
479 **kwargs: str | UploadFile,
480 ) -> None:
481 super().__init__(*args, **kwargs)
483 async def close(self) -> None:
484 for key, value in self.multi_items():
485 if isinstance(value, UploadFile):
486 await value.close()
489class Headers(Mapping[str, str]):
490 """
491 An immutable, case-insensitive multidict.
492 """
494 def __init__(
495 self,
496 headers: Mapping[str, str] | None = None,
497 raw: list[tuple[bytes, bytes]] | None = None,
498 scope: MutableMapping[str, Any] | None = None,
499 ) -> None:
500 self._list: list[tuple[bytes, bytes]] = []
501 if headers is not None:
502 assert raw is None, 'Cannot set both "headers" and "raw".'
503 assert scope is None, 'Cannot set both "headers" and "scope".'
504 self._list = [(key.lower().encode("latin-1"), value.encode("latin-1")) for key, value in headers.items()]
505 elif raw is not None:
506 assert scope is None, 'Cannot set both "raw" and "scope".'
507 self._list = raw
508 elif scope is not None:
509 # scope["headers"] isn't necessarily a list
510 # it might be a tuple or other iterable
511 self._list = scope["headers"] = list(scope["headers"])
513 @property
514 def raw(self) -> list[tuple[bytes, bytes]]:
515 return list(self._list)
517 def keys(self) -> list[str]: # type: ignore[override]
518 return [key.decode("latin-1") for key, value in self._list]
520 def values(self) -> list[str]: # type: ignore[override]
521 return [value.decode("latin-1") for key, value in self._list]
523 def items(self) -> list[tuple[str, str]]: # type: ignore[override]
524 return [(key.decode("latin-1"), value.decode("latin-1")) for key, value in self._list]
526 def getlist(self, key: str) -> list[str]:
527 get_header_key = key.lower().encode("latin-1")
528 return [item_value.decode("latin-1") for item_key, item_value in self._list if item_key == get_header_key]
530 def mutablecopy(self) -> MutableHeaders:
531 return MutableHeaders(raw=self._list[:])
533 def __getitem__(self, key: str) -> str:
534 get_header_key = key.lower().encode("latin-1")
535 for header_key, header_value in self._list:
536 if header_key == get_header_key:
537 return header_value.decode("latin-1")
538 raise KeyError(key)
540 def __contains__(self, key: Any) -> bool:
541 get_header_key = key.lower().encode("latin-1")
542 for header_key, header_value in self._list:
543 if header_key == get_header_key:
544 return True
545 return False
547 def __iter__(self) -> Iterator[Any]:
548 return iter(self.keys())
550 def __len__(self) -> int:
551 return len(self._list)
553 def __eq__(self, other: Any) -> bool:
554 if not isinstance(other, Headers):
555 return False
556 return sorted(self._list) == sorted(other._list)
558 def __repr__(self) -> str:
559 class_name = self.__class__.__name__
560 as_dict = dict(self.items())
561 if len(as_dict) == len(self):
562 return f"{class_name}({as_dict!r})"
563 return f"{class_name}(raw={self.raw!r})"
566class MutableHeaders(Headers):
567 def __setitem__(self, key: str, value: str) -> None:
568 """
569 Set the header `key` to `value`, removing any duplicate entries.
570 Retains insertion order.
571 """
572 set_key = key.lower().encode("latin-1")
573 set_value = value.encode("latin-1")
575 found_indexes: list[int] = []
576 for idx, (item_key, item_value) in enumerate(self._list):
577 if item_key == set_key:
578 found_indexes.append(idx)
580 for idx in reversed(found_indexes[1:]):
581 del self._list[idx]
583 if found_indexes:
584 idx = found_indexes[0]
585 self._list[idx] = (set_key, set_value)
586 else:
587 self._list.append((set_key, set_value))
589 def __delitem__(self, key: str) -> None:
590 """
591 Remove the header `key`.
592 """
593 del_key = key.lower().encode("latin-1")
595 pop_indexes: list[int] = []
596 for idx, (item_key, item_value) in enumerate(self._list):
597 if item_key == del_key:
598 pop_indexes.append(idx)
600 for idx in reversed(pop_indexes):
601 del self._list[idx]
603 def __ior__(self, other: Mapping[str, str]) -> MutableHeaders:
604 if not isinstance(other, Mapping):
605 raise TypeError(f"Expected a mapping but got {other.__class__.__name__}")
606 self.update(other)
607 return self
609 def __or__(self, other: Mapping[str, str]) -> MutableHeaders:
610 if not isinstance(other, Mapping):
611 raise TypeError(f"Expected a mapping but got {other.__class__.__name__}")
612 new = self.mutablecopy()
613 new.update(other)
614 return new
616 @property
617 def raw(self) -> list[tuple[bytes, bytes]]:
618 return self._list
620 def setdefault(self, key: str, value: str) -> str:
621 """
622 If the header `key` does not exist, then set it to `value`.
623 Returns the header value.
624 """
625 set_key = key.lower().encode("latin-1")
626 set_value = value.encode("latin-1")
628 for idx, (item_key, item_value) in enumerate(self._list):
629 if item_key == set_key:
630 return item_value.decode("latin-1")
631 self._list.append((set_key, set_value))
632 return value
634 def update(self, other: Mapping[str, str]) -> None:
635 for key, val in other.items():
636 self[key] = val
638 def append(self, key: str, value: str) -> None:
639 """
640 Append a header, preserving any duplicate entries.
641 """
642 append_key = key.lower().encode("latin-1")
643 append_value = value.encode("latin-1")
644 self._list.append((append_key, append_value))
646 def add_vary_header(self, vary: str) -> None:
647 existing = self.get("vary")
648 if existing is not None:
649 vary = ", ".join([existing, vary])
650 self["vary"] = vary
653class State:
654 """
655 An object that can be used to store arbitrary state.
657 Used for `request.state` and `app.state`.
658 """
660 _state: dict[str, Any]
662 def __init__(self, state: dict[str, Any] | None = None):
663 if state is None:
664 state = {}
665 super().__setattr__("_state", state)
667 def __setattr__(self, key: Any, value: Any) -> None:
668 self._state[key] = value
670 def __getattr__(self, key: Any) -> Any:
671 try:
672 return self._state[key]
673 except KeyError:
674 message = "'{}' object has no attribute '{}'"
675 raise AttributeError(message.format(self.__class__.__name__, key))
677 def __delattr__(self, key: Any) -> None:
678 del self._state[key]