Skip to main content

Async interactor for HTTP requests using a template method.

Project description

Logo

Async interactor for HTTP requests using a template method.
Documentation

Simple example

# main.py

import asyncio
import json
from collections.abc import Sequence
from typing import NamedTuple

import httpx
import pydepot

from httperactor import HttpClient, HttpInteractor, Request


class Book(NamedTuple):
    title: str
    author: str


class State(NamedTuple):
    books: tuple[Book, ...]


class StateSubscriber:
    def on_state(self, state: State) -> None:
        print(f"[StoreSubscriber] on_state called with {state}")


class SetBooksAction(pydepot.Action):
    def __init__(self, books: Sequence[Book]):
        self.books: Sequence[Book] = books


class SetBooksReducer(pydepot.Reducer[SetBooksAction, State]):
    @property
    def action_type(self) -> type[SetBooksAction]:
        return SetBooksAction

    def apply(self, action: SetBooksAction, state: State) -> State:
        return State(books=tuple(action.books))


class GetBooksRequest(Request[Sequence[Book]]):
    @property
    def path(self) -> str:
        return "/books"

    def map_response(self, response: str) -> Sequence[Book]:
        return json.loads(response, object_hook=lambda res: Book(**res))


class GetBooksInteractor(HttpInteractor[httpx.Request, Sequence[Book], State]):
    @property
    def request(self) -> Request[Sequence[Book]]:
        return GetBooksRequest()

    def actions(self, response: Sequence[Book]) -> Sequence[pydepot.Action]:
        return [SetBooksAction(books=response)]


async def main() -> None:
    store = pydepot.Store(initial_state=State(books=()))
    store.register(SetBooksReducer())

    subscriber = StateSubscriber()
    store.subscribe(subscriber)

    interactor = GetBooksInteractor(
        http_client=HttpClient(httpx.AsyncClient(base_url="http://localhost:5000")),
        store=store,
    )

    await interactor.execute()


if __name__ == "__main__":
    asyncio.run(main())
$ curl localhost:5000/books

[
  {
    "author": "Alice",
    "title": "foo"
  },
  {
    "author": "Bob",
    "title": "bar"
  },
  {
    "author": "Charlie",
    "title": "baz"
  }
]

$ python3 main.py

[StoreSubscriber] on_state called with:
    State(
        books=(
            Book(title='foo', author='Alice'),
            Book(title='bar', author='Bob'),
            Book(title='baz', author='Charlie')
        )
    )

Installation

Httperactor is available as httperactor on PyPI:

pip install httperactor

Usage

For detailed quickstart and API reference, visit the Documentation.

License

AGPLv3

Copyright (C) 2023 tombartk 

This program is free software: you can redistribute it and/or modify it under the terms
of the GNU Affero General Public License as published by the Free Software Foundation,
either version 3 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
See the GNU Affero General Public License for more details.

You should have received a copy of the GNU Affero General Public License along with this program.
If not, see https://www.gnu.org/licenses/.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

httperactor-0.1.1.tar.gz (25.7 kB view hashes)

Uploaded Source

Built Distribution

httperactor-0.1.1-py3-none-any.whl (20.8 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page