mirror of
https://github.com/revng/revng
synced 2026-06-21 14:07:57 +00:00
Giacomo Vercesi
ff4c899954
Add the storage provider (and factory) for the Pypeline Remote Storage Server (PRSS).
121 lines
3.1 KiB
Python
121 lines
3.1 KiB
Python
#
|
|
# This file is distributed under the MIT License. See LICENSE.md for details.
|
|
#
|
|
|
|
import asyncio
|
|
import base64
|
|
import hashlib
|
|
import os
|
|
import re
|
|
from collections.abc import Buffer
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
from tarfile import TarFile, TarInfo
|
|
from typing import IO, AsyncIterator, Generator, Generic, TypeVar
|
|
from urllib.parse import ParseResult
|
|
|
|
from xdg import xdg_cache_home
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class PypelineException(Exception):
|
|
pass
|
|
|
|
|
|
def cache_directory() -> Path:
|
|
if "PYPELINE_CACHE_DIR" in os.environ:
|
|
return Path(os.environ["PYPELINE_CACHE_DIR"])
|
|
else:
|
|
return xdg_cache_home() / "pypeline"
|
|
|
|
|
|
def is_mime_type_text(mime_type: str) -> bool:
|
|
"""
|
|
Returns True if the mime type represents a text file, False otherwise.
|
|
"""
|
|
return mime_type.startswith("text/") or mime_type in {
|
|
"application/json",
|
|
"application/xml",
|
|
"application/x-yaml",
|
|
"image/svg",
|
|
}
|
|
|
|
|
|
def bytes_to_string(data: Buffer, is_text: bool) -> str:
|
|
"""Standard way to convert bytes to string in the codebase."""
|
|
if is_text:
|
|
return bytes(data).decode("utf-8")
|
|
else:
|
|
return base64.b64encode(data).decode("utf-8")
|
|
|
|
|
|
def string_to_bytes(data: str, is_text: bool) -> bytes:
|
|
"""Inverse of `bytes_to_string`"""
|
|
if is_text:
|
|
return data.encode("utf-8")
|
|
else:
|
|
return base64.b64decode(data)
|
|
|
|
|
|
def crypto_hash(data: bytes | str) -> str:
|
|
"""
|
|
Returns the SHA256 hash of the given data in hexadecimal format.
|
|
"""
|
|
hasher = hashlib.sha256()
|
|
if isinstance(data, str):
|
|
data = data.encode()
|
|
hasher.update(data)
|
|
return hasher.hexdigest()
|
|
|
|
|
|
class Locked(Generic[T]):
|
|
"""
|
|
A generic class that wraps an object with an asyncio.Lock
|
|
to ensure safe concurrent access in an async environment.
|
|
|
|
The intended use is:
|
|
```python
|
|
locked_obj = Locked(your_object)
|
|
async with locked_obj() as obj:
|
|
# Safely access and modify obj here
|
|
```
|
|
|
|
But careful that if the object itself it's immutable (like integers, strings, tuples),
|
|
you cannot modify it in place, making the lock useless.
|
|
```python
|
|
locked_int = Locked(0)
|
|
async with locked_int() as value:
|
|
value += 1 # This does NOT modify the locked object!
|
|
|
|
async with locked_int() as value:
|
|
print(value) # This will still print 0
|
|
```
|
|
"""
|
|
|
|
def __init__(self, obj: T) -> None:
|
|
self._value: T = obj
|
|
self._lock = asyncio.Lock()
|
|
|
|
@asynccontextmanager
|
|
async def __call__(self) -> AsyncIterator[T]:
|
|
"""
|
|
Asynchronous context manager that acquires the lock
|
|
and yields the protected object.
|
|
"""
|
|
async with self._lock:
|
|
yield self._value
|
|
|
|
|
|
def join_url(url: ParseResult, path: str) -> str:
|
|
new_path = os.path.normpath(url.path + path)
|
|
new_path = re.sub(r"^/+", "/", new_path)
|
|
return url._replace(path=new_path).geturl()
|
|
|
|
|
|
def tar_iterate_on_members(tar: TarFile) -> Generator[tuple[TarInfo, IO[bytes]]]:
|
|
while (member := tar.next()) is not None:
|
|
f = tar.extractfile(member)
|
|
assert f is not None
|
|
yield (member, f)
|