import logging

from paddington import (
    Context, ErrorTypeSwitch, TypeSwitch, RouteNotFound, SequentialSwitch,
)
from paddington.switches.tiejoint import make_tie

error_type_switch = ErrorTypeSwitch()


@error_type_switch.track(ValueError)
def handle_value_error(event, context):
    print("!!! handler_value_error", event, context)


suppress_route_not_found = ErrorTypeSwitch()


@suppress_route_not_found.track(RouteNotFound)
def handle_no_route(event, context):
    print("!!! handle_no_route", event, context)


event_type_switch = TypeSwitch(error_track=error_type_switch)


@event_type_switch.track(int)
def handle_int(event, context):
    print(">>> handle_int", event, context)


@event_type_switch.track(str)
def handle_str(event, context):
    print(">>> handle_str", event, context)
    raise ValueError


@make_tie(event_type_switch)
def tie(track, event, context):
    print("    inner_middleware", event, context)
    track(event, context)


root_switch = SequentialSwitch(error_track=suppress_route_not_found)


@root_switch.track(lambda event, context: event == "test")
def handle_test(event, context):
    print(">>> handle_test", event, context)


root_switch.track(lambda event, context: True, tie)


def main():
    logging.basicConfig(level=logging.INFO)
    root_switch("test", Context())
    print()
    root_switch("hello_world", Context())
    print()
    root_switch(1, Context())
    print()
    root_switch(None, Context())
    print()


if __name__ == '__main__':
    main()

examples/base.py
Tishka17/paddington
from dataclasses import dataclass
from typing import Any, Callable

from paddington import Context, TypeSwitch, TieJoint


@dataclass
class CommandOne:
    value: str


class CommandOneHandler:
    def __init__(self, dependency: str):
        self.dependency = dependency

    def __call__(self, command: CommandOne):
        print(self.dependency, command)


def inject(track: Callable, event: Any, context: Context):
    return track(context.data["dependency"])(event)


mediator = TypeSwitch()
mediator.track(CommandOne, CommandOneHandler)

root = TieJoint(mediator, inject)

context = Context()
context.data["dependency"] = "dependency_value"
root(CommandOne("value"), context)

examples/didiator_mediator.py
Tishka17/paddington
import logging
import os

from magic_filter import F

from bot_framework.client import TgClient
from bot_framework.dispatch import polling
from bot_framework.routers import UpdateSwitch
from bot_framework.types import Message, CallbackQuery

router = UpdateSwitch()


@router.message(F.text == "/start")
def process_message(event: Message, bot: TgClient):
    bot.send_message(text="Started with bot", chat_id=event.chat.id)


@router.callback_query()
def process_callback_query(event: CallbackQuery, bot: TgClient):
    bot.answer_callback(callback_query_id=event.id, text="Click found")


subrouter = UpdateSwitch()


@subrouter.message()
def process_message(event: Message, bot: TgClient):
    bot.send_message(chat_id=event.chat.id, text="Your text: " + event.text)


router.include_router(subrouter)


def main():
    # real main
    logging.basicConfig(level=logging.DEBUG)
    bot = TgClient(token=os.getenv("BOT_TOKEN"))
    polling(router, bot)


if __name__ == '__main__':
    main()

examples/tg_bot/bot.py
Tishka17/paddington
import json
from datetime import datetime
from typing import List, Any

from adaptix import Retort, dumper, loader
from dataclass_rest import get
from dataclass_rest.client_protocol import FactoryProtocol
from dataclass_rest.http.requests import RequestsClient, RequestsMethod
from requests import Session, Response

from .types import Update, Message


class TgError(Exception):
    pass


class TgMethod(RequestsMethod):

    def _response_body(self, response: Response) -> Any:
        json_body = super()._response_body(response)
        if not (ok := json_body.get("ok")):
            raise TgError(
                (json_body.get("description"), json_body.get("error_code")),
            )
        return json_body.get("result")


class TgClient(RequestsClient):
    method_class = TgMethod

    def __init__(self, token):
        super().__init__(f"https://api.telegram.org/bot{token}/", Session())

    def _init_request_args_factory(self) -> Retort:
        return Retort(recipe=[
            dumper(List, json.dumps)
        ])

    def _init_response_body_factory(self) -> FactoryProtocol:
        return Retort(recipe=[
            loader(datetime, datetime.fromtimestamp),
        ])

    @get("getUpdates")
    def get_updates(
            self, offset: int, limit: int, timeout: int,
            allowed_updates: List[str],
    ) -> List[Update]:
        pass

    @get("sendMessage")
    def send_message(
            self, chat_id: int, text: str,
    ) -> Message:
        pass

    @get("answerCallbackQuery")
    def answer_callback(
            self, callback_query_id: str, text: str, show_alert: bool = False,
    ) -> bool:
        pass

examples/tg_bot/bot_framework/client.py
Tishka17/paddington
import inspect
from typing import Dict, Any

from paddington import Context, Track, WheelSet
from .types import Update


class UnpackWheelSet(WheelSet):
    def __init__(self, track: Track):
        super().__init__(track, self.tie)

    def _prepare_kwargs(self, spec, kwargs: Dict[str, Any]) -> Dict[str, Any]:
        if spec.varkw:
            return kwargs

        return {
            k: v for k, v in kwargs.items() if
            k in spec.args or k in spec.kwonlyargs
        }

    def _patch_track(self, track: Track):
        callback = inspect.unwrap(track)
        spec = inspect.getfullargspec(callback)

        def patched_track(event: Update, context: Context):
            kwargs = self._prepare_kwargs(spec, context.data)
            return callback(event.event, **kwargs)

        return patched_track


def polling(dispatcher, bot):
    offset = None
    dispatcher = UnpackWheelSet(dispatcher)

    while True:
        updates = bot.get_updates(
            offset=offset,
            limit=10,
            timeout=10,
            allowed_updates=["message", "callback_query"],
        )
        for event in updates:
            context = Context()
            context.data["bot"] = bot
            dispatcher(event, context)
            offset = event.update_id + 1

examples/tg_bot/bot_framework/dispatch.py
Tishka17/paddington
from typing import Any

from magic_filter import MagicFilter

from paddington import SequentialSwitch, MapSwitch, Context, Track
from .types import Update


class FSwitch(SequentialSwitch):

    def _prepare_predicate(self, predicate: Any):
        if isinstance(predicate, MagicFilter):
            def real_predicate(event, context):
                return predicate.resolve(event.event)

            return real_predicate
        return predicate


def update_type(event: Update, context: Context) -> str:
    return event.event_type


class UpdateSwitch(MapSwitch):
    default: FSwitch

    def __init__(self):
        super().__init__(update_type, FSwitch())
        self.message = self.track("message", FSwitch()).track
        self.callback_query = self.track("callback_query", FSwitch()).track

    def include_router(self, track: Track):
        self.default.track(track=track)

examples/tg_bot/bot_framework/routers.py
Tishka17/paddington
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List


@dataclass
class User:
    id: int
    is_bot: bool
    first_name: str
    last_name: Optional[str] = None
    username: Optional[str] = None
    language_code: Optional[str] = None
    is_premium: Optional[bool] = None
    added_to_attachment_menu: Optional[bool] = None
    can_join_groups: Optional[bool] = None
    can_read_all_group_messages: Optional[bool] = None
    supports_inline_queries: Optional[bool] = None


@dataclass
class Chat:
    id: int
    type: str
    title: Optional[str] = None
    username: Optional[str] = None
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    is_forum: Optional[bool] = None


@dataclass
class MessageEntity:
    type: str
    offset: int
    length: int
    url: Optional[str] = None
    user: Optional[User] = None
    language: Optional[str] = None
    custom_emoji_id: Optional[str] = None


@dataclass
class Message:
    message_id: int
    date: datetime
    chat: Chat
    message_thread_id: Optional[int] = None
    from_: Optional[User] = None
    text: Optional[str] = None
    entities: Optional[List[MessageEntity]] = None


@dataclass
class CallbackQuery:
    id: str
    from_: User
    chat_instance: str
    message: Optional[Message] = None
    inline_message_id: Optional[str] = None
    data: Optional[str] = None
    game_short_name: Optional[str] = None


@dataclass
class Update:
    update_id: int
    message: Optional[Message] = None
    callback_query: Optional[CallbackQuery] = None

    @property
    def event_type(self):
        attr_names = (
            key
            for key, value in self.__dict__.items()
            if key != "update_id" and value is not None
        )
        return next(attr_names, None)

    @property
    def event(self):
        attr_values = (
            value
            for key, value in self.__dict__.items()
            if key != "update_id" and value is not None
        )
        return next(attr_values, None)

examples/tg_bot/bot_framework/types.py
Tishka17/paddington
import logging
from dataclasses import dataclass
from http import HTTPStatus

from paddington import (
    Joint, Track, ErrorEvent, ErrorTypeSwitch, RouteNotFound, SequentialSwitch,
)
from web_framework.app import App, WsgiContext
from web_framework.rest_view import RestWheelSet, HttpResponse
from web_framework.wsgi_switch import WsgiSwitch

error_router = ErrorTypeSwitch(default=SequentialSwitch())
router = WsgiSwitch(error_track=error_router)


@dataclass
class User:
    id: int
    name: str


@dataclass
class Error:
    error: str


class UserManager:
    def __init__(self):
        self.users = []


class ManagerJoint(Joint):
    def __init__(self, track: Track):
        super().__init__(track)
        self.user_manager = UserManager()

    def __call__(self, event, context: WsgiContext):
        context.data["user_manager"] = self.user_manager
        return self.track(event, context)


@router.track("/", methods=["GET"])
def index(environ, context: WsgiContext) -> HttpResponse[dict[str, str]]:
    return HttpResponse(
        body={"ok": "Index"}
    )


@router.track("/users", methods=["GET"])
def get_users(environ, context: WsgiContext) -> HttpResponse[list[User]]:
    user_manager: UserManager = context.data["user_manager"]
    return HttpResponse(body=user_manager.users)


@router.track("/users", methods=["POST"])
def add_user(environ, context: WsgiContext) -> HttpResponse[User]:
    user_manager: UserManager = context.data["user_manager"]

    user = User(
        id=len(user_manager.users),
        name=f"User {len(user_manager.users)}",
    )

    user_manager.users.append(user)
    return HttpResponse(body=user)


@error_router.track(RouteNotFound)
def handle_not_found_error(
        environ: ErrorEvent, context: WsgiContext,
) -> HttpResponse[Error]:
    logging.error(f"Resource {environ.event['PATH_INFO']} not found")
    return HttpResponse(
        status=HTTPStatus.NOT_FOUND,
        body=Error(error=f"Resource {environ.event['PATH_INFO']} not found"),
    )


@error_router.default.track()
def handle_any_error(
        environ: ErrorEvent, context: WsgiContext,
) -> HttpResponse[Error]:
    logging.error("Unhandled error in HTTP")
    return HttpResponse(
        status=HTTPStatus.INTERNAL_SERVER_ERROR,
        body=Error(error=str(environ.exception)),
    )


router = ManagerJoint(router)
app = App(RestWheelSet(router))

examples/web_app/app.py
Tishka17/paddington
from typing import Callable

from paddington import Track, Context


class WsgiContext(Context):
    def __init__(self, start_response: Callable):
        super().__init__()
        self.start_response = start_response


class App:
    def __init__(self, switch: Track):
        self.switch = switch

    def __call__(self, environ, start_response):
        context = WsgiContext(start_response)
        return self.switch(environ, context)

examples/web_app/web_framework/app.py
Tishka17/paddington
import functools
import json
from dataclasses import dataclass
from http import HTTPStatus
from operator import itemgetter
from typing import Any, TypeVar, Generic, get_type_hints

from adaptix import Retort, dumper, name_mapping, Chain

from paddington import WheelSet, Track
from .app import WsgiContext

BodyT = TypeVar("BodyT")


@dataclass
class HttpResponse(Generic[BodyT]):
    body: BodyT
    status: HTTPStatus = HTTPStatus.OK


class RestWheelSet(WheelSet):
    def __init__(self, track: Track):
        super().__init__(track, self.tie)
        self.retort = Retort(recipe=[
            name_mapping(HttpResponse, only=["body"]),
            dumper(HttpResponse, itemgetter("body"), chain=Chain.LAST)
        ])

    def _patch_track(self, track: Track):
        response_type = get_type_hints(track).get("return", HttpResponse[Any])

        @functools.wraps(track)
        def patched_track(event: any, context: WsgiContext):
            response = track(event, context)
            body = self.retort.dump(response, response_type)

            response_headers = [('Content-type', 'application/json')]
            context.start_response(
                f"{response.status.value} {response.status.phrase}",
                response_headers,
            )
            return [
                json.dumps(body).encode("utf-8"),
            ]

        return patched_track

examples/web_app/web_framework/rest_view.py
Tishka17/paddington
import re
from typing import Callable, Any

from paddington import SequentialSwitch


def url(regex_str):
    regex = re.compile(regex_str)

    def url_match(environ, context):
        return regex.fullmatch(environ["PATH_INFO"])

    return url_match


def method(*methods):
    def method_match(environ, context):
        return environ["REQUEST_METHOD"] in methods

    return method_match


class WsgiSwitch(SequentialSwitch):
    def track(
            self, *predicates: Callable | str,
            methods: list[str] | None = None,
            track: Callable | None = None,
    ):
        if methods:
            predicates = predicates + (method(*methods),)
        return super().track(*predicates, track=track)

    def _prepare_predicate(self, predicate: Any) -> Callable:
        if isinstance(predicate, str):
            return url(predicate)
        return predicate

examples/web_app/web_framework/wsgi_switch.py
Tishka17/paddington
from .context import Context
from .errors import PaddingtonError, ErrorEvent, RouteNotFound
from .protocols import Track
from .switches import (
    SequentialSwitch, Joint, make_joint,
    MapSwitch, TypeSwitch, ErrorTypeSwitch, TieJoint, WheelSet,
)

src/paddington/__init__.py
Tishka17/paddington
class Context:
    def __init__(self):
        self.data = {}
        self.ties = []
        self.output = None

src/paddington/context.py
Tishka17/paddington
class PaddingtonError(Exception):
    pass


class RouteNotFound(PaddingtonError):
    pass


class ErrorEvent:
    def __init__(self, exception, router, event):
        self.exception = exception
        self.router = router
        self.event = event

    def __repr__(self):
        return (
            f"{self.__class__.__name__}"
            f"({self.exception!r}, {self.router!r}, {self.event!r})"
        )

src/paddington/errors.py
Tishka17/paddington
from typing import Protocol, Any

from .context import Context


class Track(Protocol):
    def __call__(self, event: Any, context: Context):
        raise NotImplementedError


class Tie(Protocol):
    def __call__(self, track: Track, event: Any, context: Context):
        raise NotImplementedError

src/paddington/protocols.py
Tishka17/paddington
from .base import BaseSwitch
from .joint import Joint, make_joint
from .map import ErrorTypeSwitch, MapSwitch, TypeSwitch
from .sequential import SequentialSwitch
from .tiejoint import TieJoint
from .wheelset import WheelSet

src/paddington/switches/__init__.py
Tishka17/paddington
from abc import ABC, abstractmethod
from logging import getLogger
from typing import Any, Optional

from ..context import Context
from ..errors import ErrorEvent, RouteNotFound
from ..protocols import Track, Tie

logger = getLogger(__name__)


class InternalTrack(ABC):
    @abstractmethod
    def __call__(self, event: Any, context: Context):
        raise NotImplementedError


class OutputTrack(InternalTrack):
    def __init__(self, track: Track):
        self.track = track
        self.flags = {}

    def _call_track(self, event, context: Context):
        return self.track(event, context)

    def __call__(self, event: Any, context: Context):
        track = self._call_track
        context.output = self
        for tie in reversed(context.ties):
            track = make_internal_joint(track, tie)
        return track(event, context)


def wrap_output(track: Track) -> InternalTrack:
    if isinstance(track, InternalTrack):
        return track
    return OutputTrack(track)


class InternalJoint(InternalTrack):
    def __init__(
            self, track: Track,
    ) -> None:
        super().__init__()
        self.track = track

    def __call__(self, event: Any, context: Context):
        return self.track(event, context)


def make_internal_joint(
        track: Track, handler: Tie,
) -> InternalJoint:
    def joint_track(event: Any, context: Context):
        return handler(track, event, context)

    return InternalJoint(joint_track)


class BaseSwitch(InternalTrack):
    def __init__(self, error_track: Optional[Track] = None) -> None:
        self.error_track = error_track

    def __call__(self, event: Any, context: Context):
        if not self.error_track:
            return self._dispatch(event, context)

        try:
            return self._dispatch(event, context)
        except Exception as e:
            error_event = ErrorEvent(exception=e, event=event, router=self)
            try:
                return self.error_track(error_event, context)
            except RouteNotFound:
                raise e

    def _dispatch(self, event: Any, context: Context):
        raise NotImplementedError

src/paddington/switches/base.py
Tishka17/paddington
import functools
from typing import Any, Callable

from .base import InternalJoint, wrap_output
from ..context import Context
from ..protocols import Track, Tie


class Joint(InternalJoint):
    def __init__(
            self, track: Track,
    ) -> None:
        super().__init__(wrap_output(track))


def make_joint(track: Track) -> Callable[[Tie], Joint]:
    def make_joint_decorator(handler: Tie) -> Joint:
        @functools.wraps(handler)
        def joint_track(event: Any, context: Context):
            return handler(track, event, context)

        return Joint(joint_track)

    return make_joint_decorator

src/paddington/switches/joint.py
Tishka17/paddington
from logging import getLogger
from typing import Any, Callable, Optional

from .base import BaseSwitch, wrap_output
from ..context import Context
from ..errors import ErrorEvent, RouteNotFound
from ..protocols import Track

logger = getLogger(__name__)


class MapSwitch(BaseSwitch):
    def __init__(
            self,
            getter: Callable,
            default: Optional[Track] = None,
            error_track: Optional[Track] = None,
    ) -> None:
        super().__init__(error_track)
        self.tracks: dict[Any, Track] = {}
        self.default = default
        self.getter = getter

    def track(self, value: Any, track: Optional[Callable] = None):
        if track:
            track = wrap_output(track)
            self.tracks[value] = track
            return track
        else:
            def decorator(track: Callable):
                self.track(value, track)

            return decorator

    def __getitem__(self, item) -> Track:
        return self.tracks[item]

    def __setitem__(self, value: Any, track: Track):
        self.track(value, track)

    def _dispatch(self, event: Any, context: Context):
        key = self.getter(event, context)
        logger.debug("MapSwitch key retrieved: %s", key)
        try:
            route = self.tracks[key]
        except KeyError as e:
            if self.default:
                return self.default(event, context)
            raise RouteNotFound from e
        else:
            try:
                return route(event, context)
            except RouteNotFound:
                if self.default:
                    return self.default(event, context)
                raise


def get_event_type(event, context: Context):
    return type(event)


def get_error_type(event: ErrorEvent, context: Context):
    return type(event.exception)


class TypeSwitch(MapSwitch):
    def __init__(
            self, error_track: Optional[Track] = None,
            default: Optional[Track] = None,
    ) -> None:
        super().__init__(
            get_error_type, error_track=error_track, default=default,
        )


class ErrorTypeSwitch(MapSwitch):
    def __init__(
            self, error_track: Optional[Track] = None,
            default: Optional[Track] = None,
    ) -> None:
        super().__init__(
            get_error_type, error_track=error_track, default=default,
        )

src/paddington/switches/map.py
Tishka17/paddington
from logging import getLogger
from typing import Any, Callable, Optional, List

from .base import BaseSwitch, wrap_output
from ..context import Context
from ..errors import RouteNotFound
from ..protocols import Track

logger = getLogger(__name__)


class SequentialSwitch(BaseSwitch):
    def __init__(self, error_track: Optional[Track] = None) -> None:
        super().__init__(error_track)
        self.tracks: list[tuple[List[Callable], Callable]] = []

    def _prepare_predicate(self, predicate: Any) -> Callable:
        return predicate

    def track(self, *predicates: Callable, track: Optional[Callable] = None):
        predicates = [
            self._prepare_predicate(predicate)
            for predicate in predicates
            if predicate is not None
        ]
        if track:
            track = wrap_output(track)
            self.tracks.append((predicates, track))
            return track
        else:
            def decorator(track: Callable):
                return self.track(*predicates, track=track)

            return decorator

    def _validate_predicates(
            self, predicates, event: Any, context: Context,
    ) -> bool:
        for predicate in predicates:
            if not predicate(event, context):
                return False
        return True

    def _dispatch(self, event: Any, context: Context):
        for predicates, track in self.tracks:
            logger.debug(
                "SequentialSwitch try predicates %s for track %s",
                predicates, track,
            )
            if not self._validate_predicates(predicates, event, context):
                continue
            try:
                return track(event, context)
            except RouteNotFound:
                pass
        raise RouteNotFound

src/paddington/switches/sequential.py
Tishka17/paddington
from typing import Any, Callable

from .base import BaseSwitch, wrap_output
from ..context import Context
from ..protocols import Track, Tie


class TieJoint(BaseSwitch):
    def __init__(self, track: Track, tie: Tie):
        super().__init__()
        self.track = wrap_output(track)
        self.tie = tie

    def _dispatch(self, event: Any, context: Context):
        context.ties.append(self.tie)
        try:
            return self.track(event, context)
        finally:
            context.ties.pop()


def make_tie(track: Track) -> Callable[[Tie], TieJoint]:
    def tie_decorator(tie: Tie) -> TieJoint:
        return TieJoint(track, tie)

    return tie_decorator

src/paddington/switches/tiejoint.py
Tishka17/paddington
from typing import Any, Callable

from .base import BaseSwitch, wrap_output, OutputTrack
from ..context import Context
from ..protocols import Track, Tie


class WheelSet(BaseSwitch):
    def __init__(self, track: Track, flag: Any = None):
        super().__init__()
        self.flag = flag or self.__class__
        self.track = wrap_output(track)

    def _patch_track(self, track: Track):
        return track

    def tie(self, track: Callable, event: Any, context: Context):
        output = context.output
        if self.flag not in output.flags:
            output.track = self._patch_track(output.track)
            output.flags[self.flag] = True
        return track(event, context)

    def _dispatch(self, event: Any, context: Context):
        context.ties.append(self.tie)
        try:
            return self.track(event, context)
        finally:
            context.ties.pop()

src/paddington/switches/wheelset.py
Tishka17/paddington
from typing import Any
from unittest.mock import Mock

from paddington import Joint, Context, Track


class MyJoint(Joint):
    def __init__(
            self,
            track: Track,
            pre: Track,
            post: Track,
    ) -> None:
        super().__init__(track)
        self.pre = pre
        self.post = post

    def __call__(self, event: Any, context: Context):
        self.pre(event, context)
        super().__call__(event, context)
        self.post(event, context)


def test_joint():
    pre = Mock()
    handle = Mock()
    post = Mock()
    joint = MyJoint(handle, pre, post)
    c = Context()
    joint(1, c)
    pre.assert_called_once_with(1, c)
    handle.assert_called_once_with(1, c)
    post.assert_called_once_with(1, c)

tests/test_joint.py
Tishka17/paddington