- uses: "actions/checkout@v3"
- uses: "actions/setup-python@v4"
with:
- python-version: 3.7
+ python-version: 3.10
- name: "Install dependencies"
run: "scripts/install"
- name: "Build package & docs"
strategy:
matrix:
- python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
+ python-version: ["3.8", "3.9", "3.10", "3.11"]
steps:
- uses: "actions/checkout@v3"
## Requirements
-Python 3.7+ (For Python 3.6 support, install version 0.19.1)
+Python 3.8+
## Installation
## Requirements
-Python 3.7+ (For Python 3.6 support, install version 0.19.1)
+Python 3.8+
## Installation
description = "The little ASGI library that shines."
readme = "README.md"
license = "BSD-3-Clause"
-requires-python = ">=3.7"
+requires-python = ">=3.8"
authors = [
{ name = "Tom Christie", email = "tom@tomchristie.com" },
]
"License :: OSI Approved :: BSD License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
- "Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
import asyncio
import functools
-import sys
import typing
-from types import TracebackType
-
-if sys.version_info < (3, 8): # pragma: no cover
- from typing_extensions import Protocol
-else: # pragma: no cover
- from typing import Protocol
def is_async_callable(obj: typing.Any) -> bool:
T_co = typing.TypeVar("T_co", covariant=True)
-# TODO: once 3.8 is the minimum supported version (27 Jun 2023)
-# this can just become
-# class AwaitableOrContextManager(
-# typing.Awaitable[T_co],
-# typing.AsyncContextManager[T_co],
-# typing.Protocol[T_co],
-# ):
-# pass
-class AwaitableOrContextManager(Protocol[T_co]):
- def __await__(self) -> typing.Generator[typing.Any, None, T_co]:
- ... # pragma: no cover
-
- async def __aenter__(self) -> T_co:
- ... # pragma: no cover
-
- async def __aexit__(
- self,
- __exc_type: typing.Optional[typing.Type[BaseException]],
- __exc_value: typing.Optional[BaseException],
- __traceback: typing.Optional[TracebackType],
- ) -> typing.Union[bool, None]:
- ... # pragma: no cover
+class AwaitableOrContextManager(
+ typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]
+):
+ ...
-class SupportsAsyncClose(Protocol):
+class SupportsAsyncClose(typing.Protocol):
async def close(self) -> None:
... # pragma: no cover
def routes(self) -> typing.List[BaseRoute]:
return self.router.routes
- # TODO: Make `__name` a positional-only argument when we drop Python 3.7 support.
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
- return self.router.url_path_for(__name, **path_params)
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
+ return self.router.url_path_for(name, **path_params)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
scope["app"] = self
import json
-import sys
import typing
from base64 import b64decode, b64encode
from starlette.requests import HTTPConnection
from starlette.types import ASGIApp, Message, Receive, Scope, Send
-if sys.version_info >= (3, 8): # pragma: no cover
- from typing import Literal
-else: # pragma: no cover
- from typing_extensions import Literal
-
class SessionMiddleware:
def __init__(
session_cookie: str = "session",
max_age: typing.Optional[int] = 14 * 24 * 60 * 60, # 14 days, in seconds
path: str = "/",
- same_site: Literal["lax", "strict", "none"] = "lax",
+ same_site: typing.Literal["lax", "strict", "none"] = "lax",
https_only: bool = False,
) -> None:
self.app = app
self._state = State(self.scope["state"])
return self._state
- def url_for(self, __name: str, **path_params: typing.Any) -> URL:
+ def url_for(self, name: str, /, **path_params: typing.Any) -> URL:
router: Router = self.scope["router"]
- url_path = router.url_path_for(__name, **path_params)
+ url_path = router.url_path_for(name, **path_params)
return url_path.make_absolute_url(base_url=self.base_url)
import json
import os
import stat
-import sys
import typing
from datetime import datetime
from email.utils import format_datetime, formatdate
from functools import partial
-from mimetypes import guess_type as mimetypes_guess_type
+from mimetypes import guess_type
from urllib.parse import quote
import anyio
from starlette.datastructures import URL, MutableHeaders
from starlette.types import Receive, Scope, Send
-if sys.version_info >= (3, 8): # pragma: no cover
- from typing import Literal
-else: # pragma: no cover
- from typing_extensions import Literal
-
-# Workaround for adding samesite support to pre 3.8 python
-http.cookies.Morsel._reserved["samesite"] = "SameSite" # type: ignore[attr-defined]
-
-
-# Compatibility wrapper for `mimetypes.guess_type` to support `os.PathLike` on <py3.8
-def guess_type(
- url: typing.Union[str, "os.PathLike[str]"], strict: bool = True
-) -> typing.Tuple[typing.Optional[str], typing.Optional[str]]:
- if sys.version_info < (3, 8): # pragma: no cover
- url = os.fspath(url)
- return mimetypes_guess_type(url, strict)
-
class Response:
media_type = None
domain: typing.Optional[str] = None,
secure: bool = False,
httponly: bool = False,
- samesite: typing.Optional[Literal["lax", "strict", "none"]] = "lax",
+ samesite: typing.Optional[typing.Literal["lax", "strict", "none"]] = "lax",
) -> None:
cookie: "http.cookies.BaseCookie[str]" = http.cookies.SimpleCookie()
cookie[key] = value
domain: typing.Optional[str] = None,
secure: bool = False,
httponly: bool = False,
- samesite: typing.Optional[Literal["lax", "strict", "none"]] = "lax",
+ samesite: typing.Optional[typing.Literal["lax", "strict", "none"]] = "lax",
) -> None:
self.set_cookie(
key,
def matches(self, scope: Scope) -> typing.Tuple[Match, Scope]:
raise NotImplementedError() # pragma: no cover
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
raise NotImplementedError() # pragma: no cover
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
return Match.FULL, child_scope
return Match.NONE, {}
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
seen_params = set(path_params.keys())
expected_params = set(self.param_convertors.keys())
- if __name != self.name or seen_params != expected_params:
- raise NoMatchFound(__name, path_params)
+ if name != self.name or seen_params != expected_params:
+ raise NoMatchFound(name, path_params)
path, remaining_params = replace_params(
self.path_format, self.param_convertors, path_params
return Match.FULL, child_scope
return Match.NONE, {}
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
seen_params = set(path_params.keys())
expected_params = set(self.param_convertors.keys())
- if __name != self.name or seen_params != expected_params:
- raise NoMatchFound(__name, path_params)
+ if name != self.name or seen_params != expected_params:
+ raise NoMatchFound(name, path_params)
path, remaining_params = replace_params(
self.path_format, self.param_convertors, path_params
return Match.FULL, child_scope
return Match.NONE, {}
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
- if self.name is not None and __name == self.name and "path" in path_params:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
+ if self.name is not None and name == self.name and "path" in path_params:
# 'name' matches "<mount_name>".
path_params["path"] = path_params["path"].lstrip("/")
path, remaining_params = replace_params(
)
if not remaining_params:
return URLPath(path=path)
- elif self.name is None or __name.startswith(self.name + ":"):
+ elif self.name is None or name.startswith(self.name + ":"):
if self.name is None:
# No mount name.
- remaining_name = __name
+ remaining_name = name
else:
# 'name' matches "<mount_name>:<child_name>".
- remaining_name = __name[len(self.name) + 1 :]
+ remaining_name = name[len(self.name) + 1 :]
path_kwarg = path_params.get("path")
path_params["path"] = ""
path_prefix, remaining_params = replace_params(
)
except NoMatchFound:
pass
- raise NoMatchFound(__name, path_params)
+ raise NoMatchFound(name, path_params)
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
return Match.FULL, child_scope
return Match.NONE, {}
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
- if self.name is not None and __name == self.name and "path" in path_params:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
+ if self.name is not None and name == self.name and "path" in path_params:
# 'name' matches "<mount_name>".
path = path_params.pop("path")
host, remaining_params = replace_params(
)
if not remaining_params:
return URLPath(path=path, host=host)
- elif self.name is None or __name.startswith(self.name + ":"):
+ elif self.name is None or name.startswith(self.name + ":"):
if self.name is None:
# No mount name.
- remaining_name = __name
+ remaining_name = name
else:
# 'name' matches "<mount_name>:<child_name>".
- remaining_name = __name[len(self.name) + 1 :]
+ remaining_name = name[len(self.name) + 1 :]
host, remaining_params = replace_params(
self.host_format, self.param_convertors, path_params
)
return URLPath(path=str(url), protocol=url.protocol, host=host)
except NoMatchFound:
pass
- raise NoMatchFound(__name, path_params)
+ raise NoMatchFound(name, path_params)
async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
await self.app(scope, receive, send)
response = PlainTextResponse("Not Found", status_code=404)
await response(scope, receive, send)
- def url_path_for(self, __name: str, **path_params: typing.Any) -> URLPath:
+ def url_path_for(self, name: str, /, **path_params: typing.Any) -> URLPath:
for route in self.routes:
try:
- return route.url_path_for(__name, **path_params)
+ return route.url_path_for(name, **path_params)
except NoMatchFound:
pass
- raise NoMatchFound(__name, path_params)
+ raise NoMatchFound(name, path_params)
async def startup(self) -> None:
"""
**env_options: typing.Any,
) -> "jinja2.Environment":
@pass_context
- # TODO: Make `__name` a positional-only argument when we drop Python 3.7
- # support.
- def url_for(context: dict, __name: str, **path_params: typing.Any) -> URL:
+ def url_for(context: dict, name: str, /, **path_params: typing.Any) -> URL:
request = context["request"]
- return request.url_for(__name, **path_params)
+ return request.url_for(name, **path_params)
loader = jinja2.FileSystemLoader(directory)
env_options.setdefault("loader", loader)
import json
import math
import queue
-import sys
import typing
import warnings
from concurrent.futures import Future
"You can install this with:\n"
" $ pip install httpx\n"
)
-
-if sys.version_info >= (3, 8): # pragma: no cover
- from typing import TypedDict
-else: # pragma: no cover
- from typing_extensions import TypedDict
-
_PortalFactoryType = typing.Callable[
[], typing.ContextManager[anyio.abc.BlockingPortal]
]
await instance(receive, send)
-class _AsyncBackend(TypedDict):
+class _AsyncBackend(typing.TypedDict):
backend: str
backend_options: typing.Dict[str, typing.Any]
import contextlib
import functools
-import sys
import typing
import uuid
-if sys.version_info < (3, 8):
- from typing_extensions import TypedDict # pragma: no cover
-else:
- from typing import TypedDict # pragma: no cover
-
import pytest
from starlette.applications import Starlette
startup_complete = False
shutdown_complete = False
- class State(TypedDict):
+ class State(typing.TypedDict):
count: int
items: typing.List[int]