Automated update of Base Application package.

The get and set methods have been converted from synchronous to asynchronous
implementations, enabling non-blocking operations and better concurrency
in I/O-bound scenarios.

.gitignore

@@ -1,5 +1,6 @@
 .vscode
 .history
+.backup.*
 .venv
 __pycache__
 .trigger-2024-12-02 13:37:42

Makefile

@@ -14,6 +14,7 @@ ensure_env: ensure_repo
 
 install: ensure_env
 	$(PIP) install -e . 
+	$(PIP) install git+https://molodetz.nl/retoor/zhurnal.git
 
 format: ensure_env
 	$(PIP) install shed 
@@ -38,3 +39,6 @@ cli: ensure_env
 
 test: ensure_env
 	$(PYTHON) -m unittest $(APP_NAME).tests
+
+repl: ensure_env
+	$(BIN)repl

setup.cfg

@@ -15,8 +15,10 @@ package_dir =
 python_requires = >=3.7
 install_requires =
    aiohttp
+   aiohttp-jinja2
    dataset 
-   zhurnal @ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
+   ipython
+   openai
 
 [options.packages.find]
 where = src

src/app.egg-info/PKG-INFO

@@ -1,4 +1,4 @@
-Metadata-Version: 2.1
+Metadata-Version: 2.4
 Name: app
 Version: 1.0.0
 Summary: Base application
@@ -8,5 +8,7 @@ License: MIT
 Requires-Python: >=3.7
 Description-Content-Type: text/markdown
 Requires-Dist: aiohttp
+Requires-Dist: aiohttp-jinja2
 Requires-Dist: dataset
-Requires-Dist: zhurnal@ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
+Requires-Dist: ipython
+Requires-Dist: openai

src/app.egg-info/SOURCES.txt

@@ -2,11 +2,15 @@ pyproject.toml
 setup.cfg
 src/app/__init__.py
 src/app/__main__.py
+src/app/agent.py
 src/app/app.py
 src/app/args.py
+src/app/cache.py
 src/app/cli.py
 src/app/kim.py
+src/app/queue.py
 src/app/repl.py
+src/app/rpc.py
 src/app/server.py
 src/app/tests.py
 src/app.egg-info/PKG-INFO

src/app.egg-info/requires.txt

@@ -1,3 +1,5 @@
 aiohttp
+aiohttp-jinja2
 dataset
-zhurnal@ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
+ipython
+openai

src/app/agent.py

@@ -0,0 +1,232 @@
+"""
+Written in 2024 by retoor@molodetz.nl.
+
+MIT license. Enjoy!
+
+You'll need a paid OpenAI account, named a project in it, requested an api key and created an assistant.
+URL's to all these pages are described in the class for convenience.
+
+The API keys described in this document are fake but are in the correct format for educational purposes.
+
+How to start:
+ - sudo apt install python3.12-venv python3-pip -y
+ - python3 -m venv .venv
+ - . .venv/bin/activate
+ - pip install openapi
+
+This file is to be used as part of your project or a standalone after doing
+some modifications at the end of the file.
+"""
+
+try:
+    import os
+    import sys
+
+    sys.path.append(os.getcwd())
+    import env
+
+    API_KEY = env.API_KEY
+    ASSISTANT_ID = env.ASSISTANT_ID
+except:
+    pass
+
+
+import asyncio
+import functools
+from collections.abc import Generator
+from typing import Optional
+
+from openai import OpenAI
+
+
+class Agent:
+    """
+    This class translates into an instance a single user session with its own memory.
+
+    The messages property of this class is a list containing the full chat history about
+    what the user said and what the assistant (agent) said. This can be used in future to continue
+    where you left off. Format is described in the docs of __init__ function below.
+
+    Introduction API usage for if you want to extend this class:
+    https://platform.openai.com/docs/api-reference/introduction
+    """
+
+    def __init__(
+        self, api_key: str, assistant_id: int, messages: Optional[list] = None
+    ):
+        """
+        You can find and create API keys here:
+        https://platform.openai.com/api-keys
+
+        You can find assistant_id (agent_id) here. It is the id that starts with 'asst_', not your custom name:
+        https://platform.openai.com/assistants/
+
+        Messages are optional in this format, this is to keep a message history that you can later use again:
+        [
+            {"role": "user", "message": "What is choking the chicken?"},
+            {"role": "assistant", "message": "Lucky for the cock."}
+        ]
+        """
+
+        self.assistant_id = assistant_id
+        self.api_key = api_key
+        self.client = OpenAI(api_key=self.api_key)
+        self.messages = messages or []
+        self.tool_handlers = {}
+        self.thread = self.client.beta.threads.create(messages=self.messages)
+
+    async def register_tool_handler(self, name, method):
+        self.tool_handlers[name] = method
+
+    async def dalle2(
+        self, prompt: str, width: Optional[int] = 512, height: Optional[int] = 512
+    ) -> dict:
+        """
+        In my opinion dall-e-2 produces unusual results.
+        Sizes: 256x256, 512x512 or 1024x1024.
+        """
+        result = self.client.images.generate(
+            model="dall-e-2", prompt=prompt, n=1, size=f"{width}x{height}"
+        )
+        return result
+
+    @property
+    async def models(self):
+        """
+        List models in dict format. That's more convenient than the original
+        list method because this can be directly converted to json to be used
+        in your front end or api. That's not the original result which is a
+        custom list with unserializable models.
+        """
+        return [
+            {
+                "id": model.id,
+                "owned_by": model.owned_by,
+                "object": model.object,
+                "created": model.created,
+            }
+            for model in self.client.models.list()
+        ]
+
+    async def dalle3(
+        self, prompt: str, height: Optional[int] = 1024, width: Optional[int] = 1024
+    ) -> dict:
+        """
+        Sadly only big sizes allowed. Is more pricy.
+        Sizes: 1024x1024, 1792x1024, or 1024x1792.
+        """
+        result = self.client.images.generate(
+            model="dall-e-3", prompt=prompt, n=1, size=f"{width}x{height}"
+        )
+        print(result)
+        return result
+
+    def upload_file(file_name: str, purpose: str) -> str:
+        with open(file_name, "rb") as file_fd:
+            response = self.client.files.create(file=file_fd, purpose=purpose)
+        return response.id
+
+    async def chat(
+        self, message: str, interval: Optional[float] = 0.2
+    ) -> Generator[None, None, str]:
+        """
+        Chat with the agent. It yields on given interval to inform the caller it' still busy so you can
+        update the user with live status. It doesn't hang. You can use this fully async with other
+        instances of this class.
+
+        This function also updates the self.messages list with chat history for later use.
+        """
+        message_object = {"role": "user", "content": message}
+        self.messages.append(message_object)
+        self.client.beta.threads.messages.create(
+            self.thread.id,
+            role=message_object["role"],
+            content=message_object["content"],
+        )
+        run = self.client.beta.threads.runs.create(
+            thread_id=self.thread.id, assistant_id=self.assistant_id
+        )
+
+        while run.status != "completed":
+            # for tool in run.required_action.submit_tool_outputs.tool_calls:
+            #    tool_handler = self.tool_handlers[tool.name]
+            #    output = await tool_handler(tool.arguments)
+            #    outputs.append({"tool_call_id": tool.id, "output": output})
+            # if outputs:
+            #    run = client.beta.threads.runs.submit_tool_outputs_and_poll(
+            #        thread_id=self.thread.id, run_id=run.id, tool_outputs=outputs
+            #    )
+            run = self.client.beta.threads.runs.retrieve(
+                thread_id=self.thread.id, run_id=run.id
+            )
+
+            yield None
+            await asyncio.sleep(interval)
+
+        response_messages = self.client.beta.threads.messages.list(
+            thread_id=self.thread.id
+        ).data
+        last_message = response_messages[0].content[0].text.value
+        self.messages.append({"role": "assistant", "content": last_message})
+        print(last_message)
+        yield str(last_message)
+
+    async def chatp(self, message: str) -> str:
+        """
+        Just like regular chat function but with progress indication and returns string directly.
+        This is handy for interactive usage or for a process log.
+        """
+        asyncio.get_event_loop()
+        print("Processing", end="")
+        async for message in self.chat(message):
+            if not message:
+                print(".", end="", flush=True)
+                continue
+            print("")
+            break
+        return message
+
+    async def read_line(self, ps: Optional[str] = "> "):
+        """
+        Non blocking read_line.
+        Blocking read line can break web socket connections.
+        That's why.
+        """
+        loop = asyncio.get_event_loop()
+        patched_input = functools.partial(input, ps)
+        return await loop.run_in_executor(None, patched_input)
+
+    async def cli(self):
+        """
+        Interactive client. Can be used on terminal by user or a different process.
+        The bottom new line is so that a process can check for \n\n to check if it's end response
+        and there's nothing left to wait for and thus can send next prompt if the '>' shows.
+        """
+        while True:
+            try:
+                message = await self.read_line("> ")
+                if not message.strip():
+                    continue
+                response = await self.chatp(message)
+                print(response.content[0].text.value)
+                print("")
+            except KeyboardInterrupt:
+                print("Exiting..")
+                break
+
+
+async def main():
+    """
+    Example main function. The keys here are not real but look exactly like
+    the real ones for example purposes and that you're sure your key is in the
+    right format.
+    """
+    agent = Agent(api_key=API_KEY, assistant_id=ASSISTANT_ID)
+
+    # Run interactive chat
+    await agent.cli()
+
+
+if __name__ == "__main__":
+    # Only gets executed by direct execution of script. Not when important.
+    asyncio.run(main())

src/app/app.py

@@ -1,10 +1,19 @@
+import argparse
+import asyncio
+import base64
 import json
+import pathlib
 import time
 import uuid
 
+import aiohttp_jinja2
 import dataset
+import jinja2
 from aiohttp import web
 
+from app.agent import Agent
+from app.rpc import Application as RPCApplication
+
 from . import log
 
 
@@ -16,34 +25,84 @@ def get_timestamp():
     return formatted_datetime
 
 
-class BaseApplication(web.Application):
+class BaseView(web.View):
+
+    @property
+    def app(self):
+        return self.request.app
+
+    @property
+    def template_path(self):
+        return pathlib.Path(self.request.app.template_path)
+
+    async def render_template(self, name, context=None):
+        if not context:
+            context = {}
+        return await self.request.app.render_template(str(name), self.request, context)
+
+
+class BaseApplication(RPCApplication):
 
     def __init__(
         self,
-        username=None,
+        basic_username=None,
-        password=None,
+        basic_password=None,
         cookie_name=None,
         session=None,
+        template_path=None,
         *args,
         **kwargs,
     ):
         self.cookie_name = cookie_name or str(uuid.uuid4())
-        self.username = username
+        self.basic_username = basic_username
-        self.password = password
+        self.basic_password = basic_password
         self.session = session or {}
         middlewares = kwargs.pop("middlewares", [])
-        middlewares.append(self.request_middleware)
+        # middlewares.append(self.request_middleware)
         middlewares.append(self.base64_auth_middleware)
         middlewares.append(self.session_middleware)
-        super().__init__(*args, **kwargs)
+        self.template_path = (
+            template_path
+            and template_path
+            or pathlib.Path(__file__).parent.joinpath("templates")
+        )
+        self.agents = {}
+        super().__init__(middlewares=middlewares, *args, **kwargs)
+        self.jinja2_env = aiohttp_jinja2.setup(
+            self, loader=jinja2.FileSystemLoader(self.template_path)
+        )
+
+    def run(self, *args, **kwargs):
+        if kwargs.get("port"):
+            if not kwargs.get("host"):
+                kwargs["host"] = "127.0.0.1"
+            web.run_app(self, *args, **kwargs)
 
     async def authenticate(self, username, password):
-        return self.username == username and self.password == password
+        return self.basic_username == username and self.basic_password == password
+
+    async def agent_create_thread(self, api_key, assistent_id):
+        agent = Agent(api_key, assistent_id)
+        self.agents[str(agent.thread.id)] = agent
+        return str(agent.thread.id)
+
+    async def rpc_agent_create_thread(self, api_key, assistent_id):
+        return await self.agent_create_thread(api_key, assistent_id)
+
+    async def agent_prompt(self, thread_id, message):
+        try:
+            agent = self.agents[str(thread_id)]
+            return await agent.chat(message)
+        except Exception as ex:
+            return str(ex)
+
+    async def rpc_agent_prompt(self, thread_id, message):
+        return await self.agent_prompt(str(thread_id), message)
 
     @web.middleware
-    async def base64_auth_middleware(request, handler):
+    async def base64_auth_middleware(self, request, handler):
         auth_header = request.headers.get("Authorization")
-        if not self.username:
+        if not self.basic_username:
             return await handler(request)
         if not auth_header or not auth_header.startswith("Basic "):
             return web.Response(
@@ -68,11 +127,16 @@ class BaseApplication(web.Application):
 
         return await handler(request)
 
+    async def render_template(self, name, request=None, context=None):
+        response = aiohttp_jinja2.render_template(name, request, context)
+        response.headers["Content-Type"] = "text/html"
+        return response
+
     @web.middleware
-    async def request_middleware(self, request, handler):
+    async def request_middleware(self, request: web.Request, handler):
         time_start = time.time()
         created = get_timestamp()
-        request = await handler(request)
+        response = await handler(request)
         time_end = time.time()
         await self.insert(
             "http_access",
@@ -82,7 +146,7 @@ class BaseApplication(web.Application):
                 "duration": time_end - time_start,
             },
         )
-        return request
+        return response
 
     @web.middleware
     async def session_middleware(self, request, handler):
@@ -100,81 +164,118 @@ class BaseApplication(web.Application):
 
 class WebDbApplication(BaseApplication):
 
+    @property
+    def loop(self):
+        return asyncio.get_event_loop()
+
+    async def run_in_executor(self, func, *args, **kwargs):
+        return await self.loop.run_in_executor(None, func, *args, **kwargs)
+
     def __init__(
         self, db=None, db_web=False, db_path="sqlite:///:memory:", *args, **kwargs
     ):
-        super().__init__(*args, **kwargs)
         self.db_web = db_web
         self.db_path = db_path
-        self.db = db or dataset.connect(self.db_path)
+        self.db = db or dataset.connect(
+            self.db_path, engine_kwargs={"connect_args": {"check_same_thread": False}}
+        )
+
+        super().__init__(*args, **kwargs)
         if not self.db_web:
             return
-        self.router.add_post("/insert", self.insert_handler)
+        self.router.add_post("/db/insert", self.insert_handler)
-        self.router.add_post("/update", self.update_handler)
+        self.router.add_post("/db/update", self.update_handler)
-        self.router.add_post("/upsert", self.upsert_handler)
+        self.router.add_post("/db/upsert", self.upsert_handler)
-        self.router.add_post("/find", self.find_handler)
+        self.router.add_post("/db/find", self.find_handler)
-        self.router.add_post("/find_one", self.find_one_handler)
+        self.router.add_post("/db/find_one", self.find_one_handler)
-        self.router.add_post("/delete", self.delete_handler)
+        self.router.add_post("/db/delete", self.delete_handler)
+        self.router.add_post("/db/get", self.get_handler)
+        self.router.add_post("/db/set", self.set_handler)
+        self.rpc_set = self.set
+        self.rpc_get = self.get
+        self.rpc_insert = self.insert
+        self.rpc_update = self.update
+        self.rpc_upsert = self.upsert
+        self.rpc_find = self.find
+        self.rpc_fine_one = self.find_one
+        self.rpc_delete = self.delete
+
+    async def set_handler(self, request):
+        obj = await request.json()
+        response = await self.set(obj.get("key"), obj.get("value"))
+        return web.json_response(response)
+
+    async def get_handler(self, request):
+        obj = await request.json()
+        response = await self.get(obj.get("key"), None)
+        return web.json_response(response)
 
     async def insert_handler(self, request):
-        await request.json()
+        obj = await request.json()
-        response = await self.insert(request.get("table"), request.get("data"))
+        response = await self.insert(obj.get("table"), obj.get("data"))
         return web.json_response(response)
 
     async def update_handler(self, request):
-        await request.json()
+        obj = await request.json()
         response = await self.update(
-            request.get("table"), request.get("data"), request.get("where", {})
+            obj.get("table"), obj.get("data"), obj.get("where", {})
         )
         return web.json_response(response)
 
     async def upsert_handler(self, request):
-        await request.json()
+        obj = await request.json()
         response = await self.upsert(
-            request.get("table"), request.get("data"), request.get("keys", [])
+            obj.get("table"), obj.get("data"), obj.get("keys", [])
         )
         return web.json_response(response)
 
     async def find_handler(self, request):
-        await request.json()
+        obj = await request.json()
-        response = await self.find(request.get("table"), requesst.get("where", {}))
+        response = await self.find(obj.get("table"), obj.get("where", {}))
         return web.json_response(response)
 
     async def find_one_handler(self, request):
-        await request.json()
+        obj = await request.json()
-        response = await self.find_one(request.get("table"), requesst.get("where", {}))
+        response = await self.find_one(obj.get("table"), obj.get("where", {}))
         return web.json_response(response)
 
     async def delete_handler(self, request):
-        await request.json()
+        obj = await request.json()
-        response = await self.delete(request.get("table"), requesst.get("where", {}))
+        response = await self.delete(obj.get("table"), obj.get("where", {}))
         return web.json_response(response)
 
     async def set(self, key, value):
+        return await self.run_in_executor(self.sset, key, value)
+
+    def sset(self, key, value):
         value = json.dumps(value, default=str)
-        self.db["kv"].upsert({"key": key, "value": value}, ["key"])
+        return self.db["kv"].upsert({"key": key, "value": value}, ["key"])
 
     async def get(self, key, default=None):
+        return await self.run_in_executor(self.sget, key, default)
+
+    def sget(self, key, default=None):
         record = self.db["kv"].find_one(key=key)
         if record:
-            return json.loads(record.get("value", "null"))
+            result = record.get("value", "null")
+            return result == "null" and default or json.loads(result)
         return default
 
     async def insert(self, table_name, data):
         return self.db[table_name].insert(data)
 
-    async def update(self, table_name, data, where):
+    async def update(self, table_name, data, where=None):
-        return self.db[table_name].update(data, where)
+        return self.db[table_name].update(data, where or {})
 
-    async def upsert(self, table_name, data, keys):
+    async def upsert(self, table_name, data, keys=None):
         return self.db[table_name].upsert(data, keys or [])
 
-    async def find(self, table_name, filters):
+    async def find(self, table_name, filters=None):
         if not filters:
             filters = {}
         return [dict(record) for record in self.db[table_name].find(**filters)]
 
-    async def find_one(self, table_name, filters):
+    async def find_one(self, table_name, filters=None):
         if not filters:
             filters = {}
         try:
@@ -182,7 +283,8 @@ class WebDbApplication(BaseApplication):
         except ValueError:
             return None
 
-    async def delete(self, table_name, where):
+    async def delete(self, table_name, where=None):
+        where = where or {}
         return self.db[table_name].delete(**where)
 
 
@@ -191,7 +293,7 @@ class Application(WebDbApplication):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.on_startup.append(self.on_startup_task)
-        self.router.add_get("/", self.index_handler)
+        self.router.add_get("/stat", self.index_handler)
         self.request_count = 0
         self.time_started = time.time()
         self.running_since = None
@@ -223,7 +325,51 @@ class Application(WebDbApplication):
         )
 
 
-def create_app(*args, **kwargs):
+argument_parser = argparse.ArgumentParser("Web service")
-    app = Application(*args, **kwargs)
+argument_parser.add_argument(
+    "--host", default="0.0.0.0", required=False, type=str, help="Host to serve on."
+)
+argument_parser.add_argument(
+    "--port", default=8888, required=False, type=int, help="Port to serve on."
+)
+argument_parser.add_argument(
+    "--db-path",
+    default="sqlite:///:memory:",
+    required=False,
+    type=str,
+    help="SQLAlchemy db url. (e.g. sqlite:///app.db)",
+)
+argument_parser.add_argument(
+    "--basic-username",
+    default=None,
+    required=False,
+    type=str,
+    help="Basic Auth username.",
+)
+argument_parser.add_argument(
+    "--basic-password",
+    default=None,
+    required=False,
+    type=str,
+    help="Basic Auth password.",
+)
+argument_parser.add_argument(
+    "--db-web", action="store_true", help="Enable /db/* endpoints", default=False
+)
 
+
+def create_app(*args, **kwargs):
+    global argument_parser
+    args = argument_parser.parse_args()
+    app = create_app(
+        db_path=args.db_path,
+        db_web=args.db_web,
+        basic_username=args.basic_username,
+        basic_password=args.basic_password,
+    )
+    return app
+
+
+def main():
+    app = create_app()
     return app

src/app/cache.py

@@ -0,0 +1,98 @@
+# Written by retoor@molodetz.nl
+
+# This code provides decorators for caching function results with expiration times, supporting both synchronous and asynchronous functions.
+
+# Imports: `time`, `asyncio`, and `wraps` from `functools` are used in this script.
+
+# MIT License
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+
+import json
+import time
+from functools import wraps
+
+
+def time_cache(timeout: int = 600):
+    def decorator(func):
+        cache = {}
+
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            key = [
+                arg
+                for arg in args
+                if isinstance(arg, (int, str, bool, dict, list, tuple, set))
+            ]
+            if not key:
+                return func(*args, **kwargs)
+            key = (
+                json.dumps(key, default=str),
+                json.dumps(frozenset(kwargs.items()), default=str),
+            )
+            current_time = time.time()
+
+            if key in cache:
+                result, timestamp = cache[key]
+                if current_time - timestamp < timeout:
+                    return result
+
+            result = func(*args, **kwargs)
+            cache[key] = (result, current_time)
+            return result
+
+        return wrapper
+
+    return decorator
+
+
+def time_cache_async(timeout: int = 600):
+    def decorator(func):
+        cache = {}
+
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+
+            key = [
+                arg
+                for arg in args
+                if isinstance(arg, (int, str, bool, dict, list, tuple, set))
+            ]
+
+            if not key:
+                return await func(*args, **kwargs)
+            key = (
+                json.dumps(key, default=str),
+                json.dumps(frozenset(kwargs.items()), default=str),
+            )
+            current_time = time.time()
+
+            if key in cache:
+                result, timestamp = cache[key]
+                if current_time - timestamp < timeout:
+                    return result
+
+            result = await func(*args, **kwargs)
+            cache[key] = (result, current_time)
+            return result
+
+        return wrapper
+
+    return decorator

src/app/cli.py

@@ -20,21 +20,28 @@ async def cli_client(url):
 
 
 async def bench(url):
-    index = 0
+    index = 0.0
+    # Wait until server is up
+    await asyncio.sleep(2)
+    duration_limit = 15.0
+    time_start = time.time()
+    log.info("Benchmarking.")
+    async with ClientSession() as session:
         while True:
             index += 1
             try:
-            time_start = time.time()
+                async with session.get(url.rstrip("/") + "/stat"):
-
+                    pass
-            async with ClientSession() as session:
-                async with session.get(url) as response:
-                    print(await response.text())
-                    # print(await response.json())
                 time_end = time.time()
-            print(f"Request {index}. Duration: {time_end - time_start}")
+                duration = time_end - time_start
+                if duration >= duration_limit:
+                    break
             except Exception as ex:
                 log.exception(ex)
                 await asyncio.sleep(1)
+    log.info(
+        f"{index / duration_limit} requests per second. Made {index} requests in total in duration of {duration} seconds."
+    )
 
 
 def cli_bench():

src/app/queue.py

@@ -0,0 +1,46 @@
+# Written by retoor@molodetz.nl
+
+# This code defines a CallQueue class that queues function calls and executes them sequentially, supporting both synchronous and asynchronous functions.
+
+# Imports from standard modules: functools, deque, asyncio
+
+# MIT License
+#
+# (C) 2023 The Author
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+#
+# Copies or substantial portions of the Software are reproduced and must include the above copyright notice and other similar permission notice in this particular software, including in any accompanying documentation.
+
+
+import asyncio
+from collections import deque
+from functools import wraps
+
+
+class CallQueue:
+    def __init__(self):
+        self.call_queue = deque()
+
+    def __call__(self, func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            self.call_queue.append((func, args, kwargs))
+
+        return wrapper
+
+    async def execute_next(self):
+        if self.call_queue:
+            func, args, kwargs = self.call_queue.popleft()
+            if asyncio.iscoroutinefunction(func):
+                return await func(*args, **kwargs)
+            else:
+                return func(*args, **kwargs)
+        else:
+            raise IndexError("The call queue is empty.")
+
+    async def execute_all(self):
+        results = []
+        while self.call_queue:
+            results.append(await self.execute_next())
+        return results

src/app/repl.py

@@ -1,12 +1,15 @@
-import code
+from IPython.terminal.embed import InteractiveShellEmbed
 
 
 def repl(**kwargs):
-    varlables = {}
+
+    variables = {}
     variables.update(globals().copy())
     variables.update(locals())
     variables.update(kwargs)
-    code.interact(local=variables)
+
+    shell = InteractiveShellEmbed(exit_msg="Exiting...")
+    shell(local_ns=variables)
 
 
 if __name__ == "__main__":

src/app/rpc.py

@@ -0,0 +1,255 @@
+from datetime import datetime
+from functools import partial
+from xmlrpc.client import (
+    Fault,
+    MultiCall,
+    ServerProxy,
+    dumps,
+    loads,
+)
+from xmlrpc.server import resolve_dotted_attribute
+
+from aiohttp import web
+
+
+class AsyncSimpleXMLRPCDispatcher:
+    """
+    Original not async version of this class is in the original python std lib:
+    https://github.com/python/cpython/blob/main/Lib/xmlrpc/server.py.
+
+    use_builtin_types=True allows the use of bytes-object which is preferred
+    because else it's a custom xmlrpc.client.Binary which sucks.
+    """
+
+    def __init__(
+        self, instance, allow_none=True, encoding="utf-8", use_builtin_types=True
+    ):
+        self.setup_rpc(allow_none=allow_none, encoding=encoding, use_builtin_types=True)
+        self.register_instance(instance, True)
+
+    def setup_rpc(self, allow_none=True, encoding="utf-8", use_builtin_types=True):
+        self.funcs = {}
+        self.instance = None
+        self.allow_none = allow_none
+        self.encoding = encoding or "utf-8"
+        self.use_builtin_types = use_builtin_types
+
+    def register_instance(self, instance, allow_dotted_names=True):
+        self.instance = instance
+        self.allow_dotted_names = allow_dotted_names
+        self.register_multicall_functions()
+        self.register_introspection_functions()
+
+    def register_function(self, function=None, name=None):
+        if function is None:
+            return partial(self.register_function, name=name)
+
+        if name is None:
+            name = function.__name__
+        self.funcs[name] = function
+
+        return function
+
+    def register_introspection_functions(self):
+        self.funcs.update(
+            {
+                "system.listMethods": self.system_listMethods,
+                "system.methodSignature": self.system_methodSignature,
+                "system.methodHelp": self.system_methodHelp,
+            }
+        )
+
+    def register_multicall_functions(self):
+        self.funcs.update({"system.multicall": self.system_multicall})
+
+    async def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
+        try:
+            params, method = loads(data, use_builtin_types=self.use_builtin_types)
+
+            if dispatch_method is not None:
+                response = dispatch_method(method, params)
+            else:
+                response = await self._dispatch(method, params)
+            response = (response,)
+            response = dumps(
+                response,
+                methodresponse=1,
+                allow_none=self.allow_none,
+                encoding=self.encoding,
+            )
+        except Fault as fault:
+            response = dumps(fault, allow_none=self.allow_none, encoding=self.encoding)
+        except BaseException as exc:
+            response = dumps(
+                Fault(1, f"{type(exc)}:{exc}"),
+                encoding=self.encoding,
+                allow_none=self.allow_none,
+            )
+
+        return response.encode(self.encoding, "xmlcharrefreplace")
+
+    def system_listMethods(self):
+        methods = set(self.funcs.keys())
+        if self.instance is not None:
+            if hasattr(self.instance, "_listMethods"):
+                methods |= set(self.instance._listMethods())
+            elif not hasattr(self.instance, "_dispatch"):
+                methods |= set(list_public_methods(self.instance))
+        return sorted(methods)
+
+    def system_methodSignature(self, method_name):
+        return "signatures not supported"
+
+    def system_methodHelp(self, method_name):
+        method = None
+        if method_name in self.funcs:
+            method = self.funcs[method_name]
+        elif self.instance is not None:
+            if hasattr(self.instance, "_methodHelp"):
+                return self.instance._methodHelp(method_name)
+            elif not hasattr(self.instance, "_dispatch"):
+                try:
+                    method = resolve_dotted_attribute(
+                        self.instance, method_name, self.allow_dotted_names
+                    )
+                except AttributeError:
+                    pass
+        if method is None:
+            return ""
+        else:
+            return pydoc.getdoc(method)
+
+    async def system_multicall(self, call_list):
+        results = []
+        for call in call_list:
+            method_name = call["methodName"]
+            params = call["params"]
+
+            try:
+                results.append([await self._dispatch(method_name, params)])
+            except Fault as fault:
+                results.append(
+                    {"faultCode": fault.faultCode, "faultString": fault.faultString}
+                )
+            except BaseException as exc:
+                results.append({"faultCode": 1, "faultString": f"{type(exc)}:{exc}"})
+        return results
+
+    async def _dispatch(self, method, params):
+        try:
+            func = self.funcs[method]
+        except KeyError:
+            pass
+        else:
+            if func is not None:
+                return await func(*params)
+            raise Exception(f'method "{method}" is not supported')
+
+        if self.instance is not None:
+            if hasattr(self.instance, "_dispatch"):
+                return await self.instance._dispatch(method, params)
+
+            try:
+                func = resolve_dotted_attribute(
+                    self.instance, method, self.allow_dotted_names
+                )
+            except AttributeError:
+                pass
+            else:
+                if func is not None:
+                    return await func(*params)
+
+        raise Exception(f'method "{method}" is not supported')
+
+
+def rpc_wrap_instance(obj):
+
+    class Session:
+
+        def __init__(self, data=None):
+            self._data = data or {}
+
+        async def get(self, key, default=None):
+            return self._data.get(key, default)
+
+        async def set(self, key, value):
+            self._data[key] = value
+
+        async def delete(self, key):
+            try:
+                del self._data[key]
+                return True
+            except KeyError:
+                return False
+
+        async def exists(self, key):
+            return key in self._data
+
+    class Instance:
+
+        def __init__(self, _self):
+            self._self = self
+            self.session = Session()
+
+        def __get__(self, key):
+            return getattr(self._self, key)
+
+        def ping(self, *args, **kwargs):
+            return {"args": args, "kwargs": kwargs, "timestamp": str(datetime.now())}
+
+    instance = Instance(obj)
+
+    for attr in dir(obj):
+        if attr == "rpc_handler":
+            continue
+        if attr.startswith("rpc_") and callable(getattr(obj, attr)):
+            setattr(instance, attr[4:], getattr(obj, attr))
+
+    return instance
+
+
+class Application(web.Application):
+
+    def __init__(self, url=None, host=None, port=None, *args, **kwargs):
+        self.host = host
+        self.port = port
+        self._url = url
+        self._rpc = None
+        if self.rpc_url:
+            self._rpc = ServerProxy(self.rpc_url)
+        super().__init__(*args, **kwargs)
+        self.arpc = rpc_wrap_instance(self)
+        self.rpc_dispatcher = AsyncSimpleXMLRPCDispatcher(self.arpc)
+        self.router.add_post("/rpc", self.rpc_handler)
+
+    def __get__(self, key):
+        if self._rpc:
+            return getattr(self._rpc, key)
+        return getattr(self.arpc, key)
+
+    @property
+    def url(self):
+        if self._url:
+            return self._url
+        return f"http://{self.host}:{self.port}"
+
+    @property
+    def rpc_url(self):
+        return self.url.rstrip("/") + "/rpc"
+
+    def connect(self, url):
+        return ServerProxy(url)
+
+    def multicall(self, url):
+        return MultiCall(self.connect(url))
+
+    @property
+    def rpc(self):
+        if not self._rpc:
+            self._rpc = ServerProxy(url or self.rpc_url)
+        return self._rpc
+
+    async def rpc_handler(self, request):
+        request_body = await request.text()
+        response_body = await self.rpc_dispatcher._marshaled_dispatch(request_body)
+        return web.Response(text=response_body.decode())