Skip to content
Snippets Groups Projects
Commit ba4c8f72 authored by Krzysztof Mochocki's avatar Krzysztof Mochocki
Browse files

Move url to communication and add sub-package interface

parent 9c97dcca
No related branches found
No related tags found
1 merge request!80Draft: Remove notifications
from __future__ import annotations
from beekeepy._communication import rules
from beekeepy._communication.abc.communicator import AbstractCommunicator
from beekeepy._communication.abc.overseer import AbstractOverseer
from beekeepy._communication.aiohttp_communicator import AioHttpCommunicator
from beekeepy._communication.httpx_communicator import HttpxCommunicator
from beekeepy._communication.overseers import CommonOverseer, StrictOverseer
from beekeepy._communication.request_communicator import RequestCommunicator
from beekeepy._communication.settings import CommunicationSettings
from beekeepy._communication.url import AnyUrl, HttpUrl, P2PUrl, Url, WsUrl
__all__ = [
"CommunicationSettings",
"CommonOverseer",
"StrictOverseer",
"AioHttpCommunicator",
"HttpxCommunicator",
"RequestCommunicator",
"AnyUrl",
"HttpUrl",
"P2PUrl",
"Url",
"WsUrl",
"AbstractCommunicator",
"AbstractOverseer",
"rules",
]
......@@ -3,6 +3,8 @@ from __future__ import annotations
from typing import Generic, Literal, TypeVar, get_args
from urllib.parse import urlparse
from typing_extensions import Self
P2PProtocolT = Literal[""]
HttpProtocolT = Literal["http", "https"]
WsProtocolT = Literal["ws", "wss"]
......@@ -18,6 +20,7 @@ class Url(Generic[ProtocolT]):
if protocol is not None and protocol not in allowed_proto:
raise ValueError(f"Unknown protocol: `{protocol}`, allowed: {allowed_proto}")
target_protocol: str = protocol or self._default_protocol()
if isinstance(url, Url):
self.__protocol: str = url.__protocol
self.__address: str = url.__address
......@@ -78,6 +81,14 @@ class Url(Generic[ProtocolT]):
"""
return [""]
@classmethod
def factory(cls, *, port: int = 0, address: str = "127.0.0.1") -> Self:
return cls((f"{cls._default_protocol()}://" if cls._default_protocol() else "") + f"{address}:{port}")
@classmethod
def _default_protocol(cls) -> str:
return cls._allowed_protocols()[0]
class HttpUrl(Url[HttpProtocolT]):
@classmethod
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment