Compare commits

..
5 Commits
Author SHA1 Message Date
retoor dae7bc720b Update.
Build Base Application / Build (push) Successful in 1m38s
2024-12-03 23:30:11 +01:00
retoor 7e040bb70a Update. 2024-12-03 23:25:17 +01:00
bot da6ba57b48 Automated update of Base Application package. 2024-12-03 13:19:34 +00:00
retoor 6f1d1262e9 Updated build.
Build Base Application / Build (push) Successful in 1m23s
2024-12-03 14:18:06 +01:00
retoor fefbee83e4 Initial commit.
Build Base Application / Build (push) Failing after 0s
2024-12-03 14:16:44 +01:00
13 changed files with 28 additions and 798 deletions
-1
View File
@@ -1,6 +1,5 @@
.vscode .vscode
.history .history
.backup.*
.venv .venv
__pycache__ __pycache__
.trigger-2024-12-02 13:37:42 .trigger-2024-12-02 13:37:42
-1
View File
@@ -14,7 +14,6 @@ ensure_env: ensure_repo
install: ensure_env install: ensure_env
$(PIP) install -e . $(PIP) install -e .
$(PIP) install git+https://molodetz.nl/retoor/zhurnal.git
format: ensure_env format: ensure_env
$(PIP) install shed $(PIP) install shed
BIN
View File
Binary file not shown.
BIN
View File
Binary file not shown.
+1 -2
View File
@@ -15,10 +15,9 @@ package_dir =
python_requires = >=3.7 python_requires = >=3.7
install_requires = install_requires =
aiohttp aiohttp
aiohttp-jinja2
dataset dataset
zhurnal @ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
ipython ipython
openai
[options.packages.find] [options.packages.find]
where = src where = src
+2 -3
View File
@@ -1,4 +1,4 @@
Metadata-Version: 2.4 Metadata-Version: 2.1
Name: app Name: app
Version: 1.0.0 Version: 1.0.0
Summary: Base application Summary: Base application
@@ -8,7 +8,6 @@ License: MIT
Requires-Python: >=3.7 Requires-Python: >=3.7
Description-Content-Type: text/markdown Description-Content-Type: text/markdown
Requires-Dist: aiohttp Requires-Dist: aiohttp
Requires-Dist: aiohttp-jinja2
Requires-Dist: dataset Requires-Dist: dataset
Requires-Dist: zhurnal@ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
Requires-Dist: ipython Requires-Dist: ipython
Requires-Dist: openai
-4
View File
@@ -2,15 +2,11 @@ pyproject.toml
setup.cfg setup.cfg
src/app/__init__.py src/app/__init__.py
src/app/__main__.py src/app/__main__.py
src/app/agent.py
src/app/app.py src/app/app.py
src/app/args.py src/app/args.py
src/app/cache.py
src/app/cli.py src/app/cli.py
src/app/kim.py src/app/kim.py
src/app/queue.py
src/app/repl.py src/app/repl.py
src/app/rpc.py
src/app/server.py src/app/server.py
src/app/tests.py src/app/tests.py
src/app.egg-info/PKG-INFO src/app.egg-info/PKG-INFO
+1 -2
View File
@@ -1,5 +1,4 @@
aiohttp aiohttp
aiohttp-jinja2
dataset dataset
zhurnal@ git+https://retoor.molodetz.nl/retoor/zhurnal.git@main
ipython ipython
openai
-232
View File
@@ -1,232 +0,0 @@
"""
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())
+22 -152
View File
@@ -1,19 +1,10 @@
import argparse
import asyncio
import base64
import json import json
import pathlib
import time import time
import uuid import uuid
import aiohttp_jinja2
import dataset import dataset
import jinja2
from aiohttp import web from aiohttp import web
from app.agent import Agent
from app.rpc import Application as RPCApplication
from . import log from . import log
@@ -25,84 +16,37 @@ def get_timestamp():
return formatted_datetime return formatted_datetime
class BaseView(web.View): class BaseApplication(web.Application):
@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__( def __init__(
self, self,
basic_username=None, username=None,
basic_password=None, password=None,
cookie_name=None, cookie_name=None,
session=None, session=None,
template_path=None,
*args, *args,
**kwargs, **kwargs,
): ):
self.cookie_name = cookie_name or str(uuid.uuid4()) self.cookie_name = cookie_name or str(uuid.uuid4())
self.basic_username = basic_username self.username = username
self.basic_password = basic_password self.password = password
self.session = session or {} self.session = session or {}
middlewares = kwargs.pop("middlewares", []) middlewares = kwargs.pop("middlewares", [])
# middlewares.append(self.request_middleware) middlewares.append(self.request_middleware)
middlewares.append(self.base64_auth_middleware) middlewares.append(self.base64_auth_middleware)
middlewares.append(self.session_middleware) middlewares.append(self.session_middleware)
self.template_path = ( super().__init__(*args, **kwargs)
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): 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) web.run_app(self, *args, **kwargs)
async def authenticate(self, username, password): async def authenticate(self, username, password):
return self.basic_username == username and self.basic_password == password return self.username == username and self.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 @web.middleware
async def base64_auth_middleware(self, request, handler): async def base64_auth_middleware(request, handler):
auth_header = request.headers.get("Authorization") auth_header = request.headers.get("Authorization")
if not self.basic_username: if not self.username:
return await handler(request) return await handler(request)
if not auth_header or not auth_header.startswith("Basic "): if not auth_header or not auth_header.startswith("Basic "):
return web.Response( return web.Response(
@@ -127,16 +71,11 @@ class BaseApplication(RPCApplication):
return await handler(request) 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 @web.middleware
async def request_middleware(self, request: web.Request, handler): async def request_middleware(self, request, handler):
time_start = time.time() time_start = time.time()
created = get_timestamp() created = get_timestamp()
response = await handler(request) request = await handler(request)
time_end = time.time() time_end = time.time()
await self.insert( await self.insert(
"http_access", "http_access",
@@ -146,7 +85,7 @@ class BaseApplication(RPCApplication):
"duration": time_end - time_start, "duration": time_end - time_start,
}, },
) )
return response return request
@web.middleware @web.middleware
async def session_middleware(self, request, handler): async def session_middleware(self, request, handler):
@@ -164,21 +103,12 @@ class BaseApplication(RPCApplication):
class WebDbApplication(BaseApplication): 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__( def __init__(
self, db=None, db_web=False, db_path="sqlite:///:memory:", *args, **kwargs self, db=None, db_web=False, db_path="sqlite:///:memory:", *args, **kwargs
): ):
self.db_web = db_web self.db_web = db_web
self.db_path = db_path self.db_path = db_path
self.db = db or dataset.connect( self.db = db or dataset.connect(self.db_path)
self.db_path, engine_kwargs={"connect_args": {"check_same_thread": False}}
)
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
if not self.db_web: if not self.db_web:
@@ -191,14 +121,6 @@ class WebDbApplication(BaseApplication):
self.router.add_post("/db/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/get", self.get_handler)
self.router.add_post("/db/set", self.set_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): async def set_handler(self, request):
obj = await request.json() obj = await request.json()
@@ -245,37 +167,30 @@ class WebDbApplication(BaseApplication):
return web.json_response(response) return web.json_response(response)
async def set(self, key, value): 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) value = json.dumps(value, default=str)
return 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): 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) record = self.db["kv"].find_one(key=key)
if record: if record:
result = record.get("value", "null") return json.loads(record.get("value", "null"))
return result == "null" and default or json.loads(result)
return default return default
async def insert(self, table_name, data): async def insert(self, table_name, data):
return self.db[table_name].insert(data) return self.db[table_name].insert(data)
async def update(self, table_name, data, where=None): async def update(self, table_name, data, where):
return self.db[table_name].update(data, where or {}) return self.db[table_name].update(data, where)
async def upsert(self, table_name, data, keys=None): async def upsert(self, table_name, data, keys):
return self.db[table_name].upsert(data, keys or []) return self.db[table_name].upsert(data, keys or [])
async def find(self, table_name, filters=None): async def find(self, table_name, filters):
if not filters: if not filters:
filters = {} filters = {}
return [dict(record) for record in self.db[table_name].find(**filters)] return [dict(record) for record in self.db[table_name].find(**filters)]
async def find_one(self, table_name, filters=None): async def find_one(self, table_name, filters):
if not filters: if not filters:
filters = {} filters = {}
try: try:
@@ -283,8 +198,7 @@ class WebDbApplication(BaseApplication):
except ValueError: except ValueError:
return None return None
async def delete(self, table_name, where=None): async def delete(self, table_name, where):
where = where or {}
return self.db[table_name].delete(**where) return self.db[table_name].delete(**where)
@@ -325,51 +239,7 @@ class Application(WebDbApplication):
) )
argument_parser = argparse.ArgumentParser("Web service")
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): def create_app(*args, **kwargs):
global argument_parser app = Application(*args, **kwargs)
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 return app
-98
View File
@@ -1,98 +0,0 @@
# 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
-46
View File
@@ -1,46 +0,0 @@
# 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
-255
View File
@@ -1,255 +0,0 @@
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())