import asyncio
import signal
import socket
from abc import ABC, abstractmethod
from collections.abc import Callable
from typing import Any, Generic, TypeVar
from yarl import URL
from .abc import AbstractAccessLogger, AbstractStreamWriter
from .http_parser import RawRequestMessage
from .streams import StreamReader
from .typedefs import PathLike
from .web_app import Application
from .web_exceptions import HTTPBadRequest
from .web_log import AccessLogger
from .web_protocol import RequestHandler
from .web_request import BaseRequest, Request
from .web_server import Server
try:
from ssl import SSLContext
except ImportError: # pragma: no cover
SSLContext = object # type: ignore[misc,assignment]
try:
import aiofastnet
except ImportError:
aiofastnet = None # type: ignore[assignment]
async def create_server(
loop: asyncio.AbstractEventLoop,
protocol_factory: Callable[[], asyncio.Protocol],
host: str | None = None,
port: int | None = None,
*,
sock: socket.socket | None = None,
ssl: SSLContext | None = None,
backlog: int = 100,
reuse_address: bool | None = None,
reuse_port: bool | None = None,
) -> asyncio.Server:
if aiofastnet is not None:
return await aiofastnet.create_server(
loop,
protocol_factory,
host,
port,
sock=sock,
ssl=ssl,
backlog=backlog,
reuse_address=reuse_address,
reuse_port=reuse_port,
)
else:
return await loop.create_server( # type: ignore[unreachable]
protocol_factory,
host,
port,
sock=sock,
ssl=ssl,
backlog=backlog,
reuse_address=reuse_address,
reuse_port=reuse_port,
)
__all__ = (
"BaseSite",
"TCPSite",
"UnixSite",
"NamedPipeSite",
"SockSite",
"BaseRunner",
"AppRunner",
"ServerRunner",
"GracefulExit",
)
_Request = TypeVar("_Request", bound=BaseRequest)
[docs]
class GracefulExit(SystemExit):
code = 1
def _raise_graceful_exit() -> None:
raise GracefulExit()
[docs]
class BaseSite(ABC):
__slots__ = ("_runner", "_ssl_context", "_backlog", "_server")
def __init__(
self,
runner: "BaseRunner[Any]",
*,
ssl_context: SSLContext | None = None,
backlog: int = 128,
) -> None:
if runner.server is None:
raise RuntimeError("Call runner.setup() before making a site")
self._runner = runner
self._ssl_context = ssl_context
self._backlog = backlog
self._server: asyncio.Server | None = None
@property
@abstractmethod
def name(self) -> str:
"""Return the name of the site (e.g. a URL)."""
[docs]
@abstractmethod
async def start(self) -> None:
self._runner._reg_site(self)
[docs]
async def stop(self) -> None:
self._runner._check_site(self)
if self._server is not None: # Maybe not started yet
self._server.close()
self._runner._unreg_site(self)
[docs]
class TCPSite(BaseSite):
__slots__ = ("_host", "_port", "_bound_port", "_reuse_address", "_reuse_port")
def __init__(
self,
runner: "BaseRunner[Any]",
host: str | None = None,
port: int | None = None,
*,
ssl_context: SSLContext | None = None,
backlog: int = 128,
reuse_address: bool | None = None,
reuse_port: bool | None = None,
) -> None:
super().__init__(
runner,
ssl_context=ssl_context,
backlog=backlog,
)
self._host = host
if port is None:
port = 8443 if self._ssl_context else 8080
self._port = port
self._bound_port: int | None = None
self._reuse_address = reuse_address
self._reuse_port = reuse_port
@property
def port(self) -> int:
"""The port the server is listening on.
If the server hasn't been started yet, this returns the requested port
(which might be 0 for a dynamic port).
After the server starts, it returns the actual bound port. This is
especially useful when port=0 was requested, as it allows retrieving the
dynamically assigned port after the site has started.
"""
if self._bound_port is not None:
return self._bound_port
return self._port
@property
def name(self) -> str:
scheme = "https" if self._ssl_context else "http"
host = "0.0.0.0" if not self._host else self._host
return str(URL.build(scheme=scheme, host=host, port=self.port))
async def start(self) -> None:
await super().start()
loop = asyncio.get_running_loop()
server = self._runner.server
assert server is not None
self._server = await create_server(
loop,
server,
self._host,
self._port,
ssl=self._ssl_context,
backlog=self._backlog,
reuse_address=self._reuse_address,
reuse_port=self._reuse_port,
)
if self._server.sockets:
self._bound_port = self._server.sockets[0].getsockname()[1]
else:
self._bound_port = self._port
[docs]
class UnixSite(BaseSite):
__slots__ = ("_path",)
def __init__(
self,
runner: "BaseRunner[Any]",
path: PathLike,
*,
ssl_context: SSLContext | None = None,
backlog: int = 128,
) -> None:
super().__init__(
runner,
ssl_context=ssl_context,
backlog=backlog,
)
self._path = path
@property
def name(self) -> str:
scheme = "https" if self._ssl_context else "http"
return f"{scheme}://unix:{self._path}:"
async def start(self) -> None:
await super().start()
loop = asyncio.get_running_loop()
server = self._runner.server
assert server is not None
self._server = await loop.create_unix_server(
server,
self._path,
ssl=self._ssl_context,
backlog=self._backlog,
)
[docs]
class NamedPipeSite(BaseSite):
__slots__ = ("_path",)
def __init__(self, runner: "BaseRunner[Any]", path: str) -> None:
loop = asyncio.get_running_loop()
if not isinstance(
loop, asyncio.ProactorEventLoop # type: ignore[attr-defined]
):
raise RuntimeError(
"Named Pipes only available in proactor loop under windows"
)
super().__init__(runner)
self._path = path
@property
def name(self) -> str:
return self._path
async def start(self) -> None:
await super().start()
loop = asyncio.get_running_loop()
server = self._runner.server
assert server is not None
_server = await loop.start_serving_pipe( # type: ignore[attr-defined]
server, self._path
)
self._server = _server[0]
[docs]
class SockSite(BaseSite):
__slots__ = ("_sock", "_name")
def __init__(
self,
runner: "BaseRunner[Any]",
sock: socket.socket,
*,
ssl_context: SSLContext | None = None,
backlog: int = 128,
) -> None:
super().__init__(
runner,
ssl_context=ssl_context,
backlog=backlog,
)
self._sock = sock
scheme = "https" if self._ssl_context else "http"
if hasattr(socket, "AF_UNIX") and sock.family == socket.AF_UNIX:
name = f"{scheme}://unix:{sock.getsockname()}:"
else:
host, port = sock.getsockname()[:2]
name = str(URL.build(scheme=scheme, host=host, port=port))
self._name = name
@property
def name(self) -> str:
return self._name
async def start(self) -> None:
await super().start()
loop = asyncio.get_running_loop()
server = self._runner.server
assert server is not None
self._server = await create_server(
loop, server, sock=self._sock, ssl=self._ssl_context, backlog=self._backlog
)
[docs]
class BaseRunner(ABC, Generic[_Request]):
__slots__ = ("_handle_signals", "_kwargs", "_server", "_sites", "_shutdown_timeout")
def __init__(
self,
*,
handle_signals: bool = False,
shutdown_timeout: float = 60.0,
**kwargs: Any,
) -> None:
self._handle_signals = handle_signals
self._kwargs = kwargs
self._server: Server[_Request] | None = None
self._sites: list[BaseSite] = []
self._shutdown_timeout = shutdown_timeout
@property
def server(self) -> Server[_Request] | None:
return self._server
@property
def addresses(self) -> list[Any]:
ret: list[Any] = []
for site in self._sites:
server = site._server
if server is not None:
sockets = server.sockets
if sockets is not None:
for sock in sockets:
ret.append(sock.getsockname())
return ret
@property
def sites(self) -> set[BaseSite]:
return set(self._sites)
[docs]
async def setup(self) -> None:
loop = asyncio.get_running_loop()
if self._handle_signals:
try:
loop.add_signal_handler(signal.SIGINT, _raise_graceful_exit)
loop.add_signal_handler(signal.SIGTERM, _raise_graceful_exit)
except NotImplementedError:
# add_signal_handler is not implemented on Windows
pass
self._server = await self._make_server()
@abstractmethod
async def shutdown(self) -> None:
"""Call any shutdown hooks to help server close gracefully."""
[docs]
async def cleanup(self) -> None:
# The loop over sites is intentional, an exception on gather()
# leaves self._sites in unpredictable state.
# The loop guarantees that a site is either deleted on success or
# still present on failure
for site in list(self._sites):
await site.stop()
if self._server: # If setup succeeded
# Yield to event loop to ensure incoming requests prior to stopping the sites
# have all started to be handled before we proceed to close idle connections.
await asyncio.sleep(0)
self._server.pre_shutdown()
await self.shutdown()
await self._server.shutdown(self._shutdown_timeout)
await self._cleanup_server()
self._server = None
if self._handle_signals:
loop = asyncio.get_running_loop()
try:
loop.remove_signal_handler(signal.SIGINT)
loop.remove_signal_handler(signal.SIGTERM)
except NotImplementedError:
# remove_signal_handler is not implemented on Windows
pass
@abstractmethod
async def _make_server(self) -> Server[_Request]:
"""Return a new server for the runner to serve requests."""
@abstractmethod
async def _cleanup_server(self) -> None:
"""Run any cleanup steps after the server is shutdown."""
def _reg_site(self, site: BaseSite) -> None:
if site in self._sites:
raise RuntimeError(f"Site {site} is already registered in runner {self}")
self._sites.append(site)
def _check_site(self, site: BaseSite) -> None:
if site not in self._sites:
raise RuntimeError(f"Site {site} is not registered in runner {self}")
def _unreg_site(self, site: BaseSite) -> None:
if site not in self._sites:
raise RuntimeError(f"Site {site} is not registered in runner {self}")
self._sites.remove(site)
[docs]
class ServerRunner(BaseRunner[BaseRequest]):
"""Low-level web server runner"""
__slots__ = ("_web_server",)
def __init__(
self,
web_server: Server[BaseRequest],
*,
handle_signals: bool = False,
**kwargs: Any,
) -> None:
super().__init__(handle_signals=handle_signals, **kwargs)
self._web_server = web_server
async def shutdown(self) -> None:
pass
async def _make_server(self) -> Server[BaseRequest]:
return self._web_server
async def _cleanup_server(self) -> None:
pass
[docs]
class AppRunner(BaseRunner[Request]):
"""Web Application runner"""
__slots__ = ("_app",)
def __init__(
self,
app: Application,
*,
handle_signals: bool = False,
access_log_class: type[AbstractAccessLogger] = AccessLogger,
**kwargs: Any,
) -> None:
if not isinstance(app, Application):
raise TypeError(
f"The first argument should be web.Application instance, got {app!r}"
)
kwargs["access_log_class"] = access_log_class
if app._handler_args:
for k, v in app._handler_args.items():
kwargs[k] = v
if not issubclass(kwargs["access_log_class"], AbstractAccessLogger):
raise TypeError(
"access_log_class must be subclass of "
"aiohttp.abc.AbstractAccessLogger, got {}".format(
kwargs["access_log_class"]
)
)
super().__init__(handle_signals=handle_signals, **kwargs)
self._app = app
@property
def app(self) -> Application:
return self._app
async def shutdown(self) -> None:
await self._app.shutdown()
async def _make_server(self) -> Server[Request]:
self._app.on_startup.freeze()
await self._app.startup()
self._app.freeze()
return Server(
self._app._handle,
request_factory=self._make_request,
**self._kwargs,
)
def _make_request(
self,
message: RawRequestMessage,
payload: StreamReader,
protocol: RequestHandler[Request],
writer: AbstractStreamWriter,
task: "asyncio.Task[None]",
pre_handler_error: HTTPBadRequest | None,
_cls: type[Request] = Request,
) -> Request:
loop = asyncio.get_running_loop()
return _cls(
message,
payload,
protocol,
writer,
task,
loop,
client_max_size=self.app._client_max_size,
pre_handler_error=pre_handler_error,
)
async def _cleanup_server(self) -> None:
await self._app.cleanup()
