1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
import asyncio
import random
import aiohttp
from .errors import RPCError, MaxRetries
from .types import Results, UrlParameterType
DEFAULT_AURWEB_VERSION = 5
MAX_WORKERS = 10
METHODS = {'search', 'info', 'multiinfo', 'msearch', 'suggest',
'suggest-pkgbase', 'get-comment-form'}
FIELDS = {'name', 'name-desc', 'maintainer', 'depends',
'makedepends', 'optdepends', 'checkdepends'}
class AsyncRPC:
"""
https://wiki.archlinux.org/title/Aurweb_RPC_interface
"""
RPC_URL = 'https://aur.archlinux.org/rpc/'
def __init__(self, max_workers: int = MAX_WORKERS):
self._client = aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(10))
self.sem = asyncio.Semaphore(max_workers)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.close()
async def close(self):
await self._client.close()
async def _post(self, params: dict) -> dict:
async with self.sem:
for i in range(5):
try:
async with self._client.get(self.RPC_URL, params=params) as resp:
resp.raise_for_status()
return await resp.json()
except aiohttp.ServerTimeoutError:
await asyncio.sleep(3 * random.random())
continue
raise MaxRetries
async def request(self,
method: str,
method_params: dict[str, UrlParameterType],
):
params = {
'v': DEFAULT_AURWEB_VERSION,
'type': method,
**method_params
}
json_response = await self._post(params)
if 'error' in json_response:
raise RPCError(method, json_response['error'])
return Results.parse(json_response)
async def search(self,
arg: UrlParameterType,
search_by: str = None,
):
method_params = {
'by': search_by or [],
'arg': arg,
}
return await self.request('search', method_params)
async def info(self,
arg: UrlParameterType,
):
return await self.multiinfo(arg)
async def multiinfo(self,
arg: UrlParameterType,
):
method_params = {
'arg[]': arg,
}
return await self.request('multiinfo', method_params)
async def msearch(self, arg: UrlParameterType):
return await self.search(arg, search_by='maintainer')
|
