932b0b48-8d1a-4bbd-a9fd-a090d6114750

Build # 932b0b48-8d1a-4bbd-a9fd-a090d6114750

Build Type

push

circleci

Committed by Sym Roe

Commit Message

Use S3 ballot cache rather than WCIVF directly

Run Details

507 of 582 relevant lines covered (87.11%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

71.43

/api/common/async_requests.py

import asyncio
from json import JSONDecodeError
from typing import Dict

import httpx


class UpstreamApiError(Exception):
    def __init__(self, response_dict: dict):
        try:
            self.message = response_dict["response"].json()["detail"]
        except JSONDecodeError:
            self.message = ""
        self.status = response_dict["response"].status_code


async def get_url(key, url_data, request_urls):
    async with httpx.AsyncClient() as client:
        response: httpx.Response = client.get(
            url=url_data["url"],
            params=url_data.get("params", {}),
            headers=url_data.get("headers", {}),
        )
        request_urls[key]["response"] = await response
        request_urls[key]["response"].raise_for_status()


async def async_get_urls(requst_urls) -> Dict[str, httpx.Response]:
    await asyncio.gather(
        *[get_url(key, requst_urls[key], requst_urls) for key in requst_urls]
    )

    for url, result in requst_urls.items():
        if result["response"].status_code >= 400:
            raise UpstreamApiError(result)
    return requst_urls


class AsyncRequester:
    """
    Used HTTPX and async to request URLs in parallel

    Pass in a dict with the following structure:

    {
        "key1": {
            "url": "https://example.com/url1/",
            "params": {},
            "headers": {}
        },
        "key2": {
            "url": "https://example.com/url2/",
            "params": {},
            "headers": {}
        }
    }

    An async request for each URL will be started, and when they're all complete
    the dict will be returned with `response` objects:

    {
        "key1": {
            "url": "https://example.com/url1/",
            "params": {},
            "headers": {},
            "response": <httpx response object>
        },
        "key2": {
            "url": "https://example.com/url2/",
            "params": {},
            "headers": {},
            "response": <httpx response object>
        }
    }

    """

    USER_AGENT = "devs.DC API"

    def __init__(self, request_dict: Dict):
        self.request_dict = request_dict

    @property
    def get_default_headers(self):
        return {"Accept": "application/json", "User-Agent": self.USER_AGENT}

    def add_default_headers(self):
        for key, value in self.request_dict.items():
            headers = value.get("headers", {})
            headers.update(self.get_default_headers)

    async def get_urls(self) -> dict:
        return await async_get_urls(self.request_dict)

1	import asyncio	1✔
2	from json import JSONDecodeError	1✔
3	from typing import Dict	1✔
4
5	import httpx	1✔
6
7
8	class UpstreamApiError(Exception):	1✔
9	def __init__(self, response_dict: dict):	1✔
10	try:	×
11	self.message = response_dict["response"].json()["detail"]	×
12	except JSONDecodeError:	×
13	self.message = ""	×
14	self.status = response_dict["response"].status_code	×
15
16
17	async def get_url(key, url_data, request_urls):	1✔
18	async with httpx.AsyncClient() as client:	1✔
19	response: httpx.Response = client.get(	1✔
20	url=url_data["url"],
21	params=url_data.get("params", {}),
22	headers=url_data.get("headers", {}),
23	)
24	request_urls[key]["response"] = await response	1✔
25	request_urls[key]["response"].raise_for_status()	1✔
26
27
28	async def async_get_urls(requst_urls) -> Dict[str, httpx.Response]:	1✔
29	await asyncio.gather(	1✔
30	*[get_url(key, requst_urls[key], requst_urls) for key in requst_urls]
31	)
32
33	for url, result in requst_urls.items():	1✔
34	if result["response"].status_code >= 400:	1✔
35	raise UpstreamApiError(result)	×
36	return requst_urls	1✔
37
38
39	class AsyncRequester:	1✔
40	"""
41	Used HTTPX and async to request URLs in parallel
42
43	Pass in a dict with the following structure:
44
45	{
46	"key1": {
47	"url": "https://example.com/url1/",
48	"params": {},
49	"headers": {}
50	},
51	"key2": {
52	"url": "https://example.com/url2/",
53	"params": {},
54	"headers": {}
55	}
56	}
57
58	An async request for each URL will be started, and when they're all complete
59	the dict will be returned with `response` objects:
60
61	{
62	"key1": {
63	"url": "https://example.com/url1/",
64	"params": {},
65	"headers": {},
66	"response": <httpx response object>
67	},
68	"key2": {
69	"url": "https://example.com/url2/",
70	"params": {},
71	"headers": {},
72	"response": <httpx response object>
73	}
74	}
75
76	"""
77
78	USER_AGENT = "devs.DC API"	1✔
79
80	def __init__(self, request_dict: Dict):	1✔
81	self.request_dict = request_dict	1✔
82
83	@property	1✔
84	def get_default_headers(self):	1✔
85	return {"Accept": "application/json", "User-Agent": self.USER_AGENT}	×
86
87	def add_default_headers(self):	1✔
88	for key, value in self.request_dict.items():	×
89	headers = value.get("headers", {})	×
90	headers.update(self.get_default_headers)	×
91
92	async def get_urls(self) -> dict:	1✔
93	return await async_get_urls(self.request_dict)	1✔

DemocracyClub / aggregator-api / 932b0b48-8d1a-4bbd-a9fd-a090d6114750

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous