(
self,
handler: Callable[[ServerConnection], Awaitable[None]],
host: str | None = None,
port: int | None = None,
*,
# WebSocket
origins: Sequence[Origin | re.Pattern[str] | None] | None = None,
extensions: Sequence[ServerExtensionFactory] | None = None,
subprotocols: Sequence[Subprotocol] | None = None,
select_subprotocol: (
Callable[
[ServerConnection, Sequence[Subprotocol]],
Subprotocol | None,
]
| None
) = None,
compression: str | None = "deflate",
# HTTP
process_request: (
Callable[
[ServerConnection, Request],
Awaitable[Response | None] | Response | None,
]
| None
) = None,
process_response: (
Callable[
[ServerConnection, Request, Response],
Awaitable[Response | None] | Response | None,
]
| None
) = None,
server_header: str | None = SERVER,
# Timeouts
open_timeout: float | None = 10,
ping_interval: float | None = 20,
ping_timeout: float | None = 20,
close_timeout: float | None = 10,
# Limits
max_size: int | None | tuple[int | None, int | None] = 2**20,
max_queue: int | None | tuple[int | None, int | None] = 16,
write_limit: int | tuple[int, int | None] = 2**15,
# Logging
logger: LoggerLike | None = None,
# Escape hatch for advanced customization
create_connection: type[ServerConnection] | None = None,
# Other keyword arguments are passed to loop.create_server
**kwargs: Any,
)
| 696 | """ |
| 697 | |
| 698 | def __init__( |
| 699 | self, |
| 700 | handler: Callable[[ServerConnection], Awaitable[None]], |
| 701 | host: str | None = None, |
| 702 | port: int | None = None, |
| 703 | *, |
| 704 | # WebSocket |
| 705 | origins: Sequence[Origin | re.Pattern[str] | None] | None = None, |
| 706 | extensions: Sequence[ServerExtensionFactory] | None = None, |
| 707 | subprotocols: Sequence[Subprotocol] | None = None, |
| 708 | select_subprotocol: ( |
| 709 | Callable[ |
| 710 | [ServerConnection, Sequence[Subprotocol]], |
| 711 | Subprotocol | None, |
| 712 | ] |
| 713 | | None |
| 714 | ) = None, |
| 715 | compression: str | None = "deflate", |
| 716 | # HTTP |
| 717 | process_request: ( |
| 718 | Callable[ |
| 719 | [ServerConnection, Request], |
| 720 | Awaitable[Response | None] | Response | None, |
| 721 | ] |
| 722 | | None |
| 723 | ) = None, |
| 724 | process_response: ( |
| 725 | Callable[ |
| 726 | [ServerConnection, Request, Response], |
| 727 | Awaitable[Response | None] | Response | None, |
| 728 | ] |
| 729 | | None |
| 730 | ) = None, |
| 731 | server_header: str | None = SERVER, |
| 732 | # Timeouts |
| 733 | open_timeout: float | None = 10, |
| 734 | ping_interval: float | None = 20, |
| 735 | ping_timeout: float | None = 20, |
| 736 | close_timeout: float | None = 10, |
| 737 | # Limits |
| 738 | max_size: int | None | tuple[int | None, int | None] = 2**20, |
| 739 | max_queue: int | None | tuple[int | None, int | None] = 16, |
| 740 | write_limit: int | tuple[int, int | None] = 2**15, |
| 741 | # Logging |
| 742 | logger: LoggerLike | None = None, |
| 743 | # Escape hatch for advanced customization |
| 744 | create_connection: type[ServerConnection] | None = None, |
| 745 | # Other keyword arguments are passed to loop.create_server |
| 746 | **kwargs: Any, |
| 747 | ) -> None: |
| 748 | if subprotocols is not None: |
| 749 | validate_subprotocols(subprotocols) |
| 750 | |
| 751 | if compression == "deflate": |
| 752 | extensions = enable_server_permessage_deflate(extensions) |
| 753 | elif compression is not None: |
| 754 | raise ValueError(f"unsupported compression: {compression}") |
| 755 |
no test coverage detected