Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/utils.py: 28%
107 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 11:54 +0000
1import asyncio
2import base64
3import glob
4import hashlib
5import hmac
6import logging
7import os
8import threading
9from datetime import datetime
10from hashlib import sha1
11from typing import Coroutine, Dict, List, Tuple
13import aiohttp
16def get_filepaths_with_glob(root_path: str, file_regex: str):
17 return glob.glob(os.path.join(root_path, file_regex))
20def hash_file(tmp_file_path):
21 BUF_SIZE = 65536 # lets read stuff in 64kb chunks!
22 sha256_hash = hashlib.sha256()
23 with open(tmp_file_path, "rb") as file:
24 while True:
25 data = file.read(BUF_SIZE)
26 if not data:
27 break
28 sha256_hash.update(data)
29 return sha256_hash.hexdigest()
32async def throw_if_bad_status_code(
33 response: aiohttp.ClientResponse, expected: List[int], logger=None
34) -> aiohttp.ClientResponse:
35 if response.status in expected:
36 return response
38 # else, bad status code
39 details = await response.json()
40 if logger:
41 logger.warning(
42 "Unexpected response code {status}: {details}",
43 status=response.status,
44 details=details,
45 )
46 raise ValueError(
47 f"unexpected response code while fetching bundle: {response.status}"
48 )
51def tuple_to_dict(tup: Tuple[str, str]) -> Dict[str, str]:
52 return dict([tup])
55def get_authorization_header(token: str) -> Tuple[str, str]:
56 return "Authorization", f"Bearer {token}"
59def build_aws_rest_auth_headers(
60 key_id: str, secret_key: str, host: str, path: str, region: str
61):
62 """Use the AWS signature algorithm (https://docs.aws.amazon.com/AmazonS3/la
63 test/userguide/RESTAuthentication.html) to generate the hTTP headers.
65 Args:
66 key_id (str): Access key (aka user ID) of an account in the S3 service.
67 secret_key (str): Secret key (aka password) of an account in the S3 service.
68 host (str): S3 storage host
69 path (str): path to bundle file in s3 storage (including bucket)
71 Returns: http headers
72 """
74 def sign(key, msg):
75 return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()
77 def getSignatureKey(key, dateStamp, regionName, serviceName):
78 kDate = sign(("AWS4" + key).encode("utf-8"), dateStamp)
79 kRegion = sign(kDate, regionName)
80 kService = sign(kRegion, serviceName)
81 kSigning = sign(kService, "aws4_request")
82 return kSigning
84 # SHA256 of empty string. This is needed when S3 request payload is empty.
85 SHA256_EMPTY = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
87 t = datetime.utcnow()
88 amzdate = t.strftime("%Y%m%dT%H%M%SZ")
89 datestamp = t.strftime("%Y%m%d")
91 canonical_headers = "host:" + host + "\n" + "x-amz-date:" + amzdate + "\n"
92 signed_headers = "host;x-amz-date"
94 payload_hash = hashlib.sha256("".encode("utf-8")).hexdigest()
96 canonical_request = (
97 "GET"
98 + "\n"
99 + path
100 + "\n"
101 + "\n"
102 + canonical_headers
103 + "\n"
104 + signed_headers
105 + "\n"
106 + payload_hash
107 )
109 algorithm = "AWS4-HMAC-SHA256"
110 credential_scope = datestamp + "/" + region + "/" + "s3" + "/" + "aws4_request"
112 string_to_sign = (
113 algorithm
114 + "\n"
115 + amzdate
116 + "\n"
117 + credential_scope
118 + "\n"
119 + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
120 )
121 signing_key = getSignatureKey(secret_key, datestamp, region, "s3")
122 signature = hmac.new(
123 signing_key, (string_to_sign).encode("utf-8"), hashlib.sha256
124 ).hexdigest()
126 authorization_header = (
127 algorithm
128 + " "
129 + "Credential="
130 + key_id
131 + "/"
132 + credential_scope
133 + ", "
134 + "SignedHeaders="
135 + signed_headers
136 + ", "
137 + "Signature="
138 + signature
139 )
141 return {
142 "x-amz-date": amzdate,
143 "x-amz-content-sha256": SHA256_EMPTY,
144 "Authorization": authorization_header,
145 }
148def sorted_list_from_set(s: set) -> list:
149 l = list(s)
150 l.sort()
151 return l
154async def thread_worker(queue: asyncio.Queue, logger: logging.Logger):
155 """The worker task is *running and then awaiting* a coroutine that was
156 scheduled on the thread's async loop from *OUTSIDE* (i.e: from another
157 thread).
159 Args:
160 queue (asyncio.Queue): The Queue
161 engine (BaseFetchingEngine): The engine itself
162 """
163 while True:
164 # get the next coroutine scheduled on the thread's queue
165 # this may block until another coroutine is scheduled
166 coro: Coroutine = await queue.get()
168 try:
169 # await on the coroutine and possibly block *this* worker
170 await coro
171 except Exception as err:
172 logger.exception(f"Scheduled coroutine - {coro} failed")
173 finally:
174 # Notify the queue that the "work item" has been processed.
175 queue.task_done()
178class AsyncioEventLoopThread(threading.Thread):
179 """This class enable a sync (or async) program to run (another) asyncio
180 event loop in a separate thread without blocking the main thread or
181 interfering with the main thread's asyncio loop if such exists.
183 usage:
184 t = AsyncioEventLoopThread()
186 # not yet running
187 t.create_task(coroutine1())
188 t.create_task(coroutine2())
190 # will start the event loop and all scheduled tasks
191 t.start()
192 """
194 DEFAULT_WORKER_COUNT = 5
196 def __init__(self, *args, loop=None, worker_count=DEFAULT_WORKER_COUNT, **kwargs):
197 super().__init__(*args, **kwargs)
198 self.daemon = True
199 self.running = False
200 self.loop = loop or asyncio.new_event_loop()
201 # the thread is assigned a main logger bearing its name
202 self.logger = logging.getLogger(self.name)
203 # The internal task queue
204 self._queue = asyncio.Queue(loop=self.loop)
205 # Worker working the queue
206 self._tasks = []
208 # create worker tasks
209 for _ in range(worker_count):
210 self._create_worker()
212 def run(self):
213 """Called by the default threading.Thread.start() method.
215 runs the main activity of the thread, which in our case is
216 simply running the asyncio loop until it stop.
217 """
218 self.running = True
219 # does not return (thread will keep running) until loop.stop() is called
220 if not self.loop.is_running():
221 self.loop.run_forever()
223 def stop(self):
224 """Stops the thread.
226 (Stop the async loop running on the thread and then joins the
227 main thread).
228 """
229 self.run_coro(self._shutdown()) # will block until _shutdown() returns
230 self.join() # will block until run() exits
231 self.running = False
233 def _create_worker(self) -> asyncio.Task:
234 """Create an asyncio worker task to work the thread's queue."""
235 task = self.loop.create_task(thread_worker(self._queue, self.logger))
236 self._tasks.append(task)
237 return task
239 async def _shutdown(self):
240 """Cancel and wait on the thread's async tasks."""
241 tasks = [
242 t
243 for t in asyncio.all_tasks(loop=self.loop)
244 if t is not asyncio.current_task()
245 ]
246 for task in tasks:
247 task.cancel()
248 # Wait until all tasks are cancelled.
249 await asyncio.gather(*tasks, return_exceptions=True)
250 # stop the thread async loop
251 self.loop.stop()
253 def create_task(self, coro: Coroutine):
254 """Creates a task on the thread's asyncio loop *without* waiting for it
255 to finish. This is intended to be called from the parent thread as a
256 set-and-forget.
258 the scheduled coroutine is put on the thread's queue and is
259 consumed by one of the thread workers.
260 """
262 async def _schedule_task():
263 """Since the queue is infinite, queue.put() will not block."""
264 await self._queue.put(coro)
266 # the asyncio loop might not be running yet (if the thread was
267 # not yet started), therefore we do not block on the result.
268 return asyncio.run_coroutine_threadsafe(_schedule_task(), loop=self.loop)
270 def run_coro(self, coro: Coroutine):
271 """Can be called from the main thread, but will run the coroutine on
272 the event loop thread.
274 the main thread will block until a result is returned. calling
275 run_coro() is thread-safe.
276 """
277 return asyncio.run_coroutine_threadsafe(coro, loop=self.loop).result()