admin/revng-revng
1
0
Fork 0
mirror of https://github.com/revng/revng synced 2026-06-21 14:07:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

revng.project: port to pypeline

Browse Source 
Convert `revng.project` to use the `revng2` command-line and the `revng
project daemon` server instead of the legacy ones.
This commit is contained in:
Giacomo Vercesi
2026-03-09 15:30:23 +01:00
committed by Alessandro Di Federico
parent b229fcc469
commit 8ad19f2f3d
12 changed files with 670 additions and 579 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/CMakeLists.txt
+16 -1
View File
@@ -249,6 +249,7 @@ add_custom_command(
add_custom_command(
COMMAND
env "PYTHONPATH=${CMAKE_BINARY_DIR}/${PYTHON_INSTALL_PATH}"
"${CMAKE_SOURCE_DIR}/scripts/generate-project-mixins.sh"
"${CMAKE_SOURCE_DIR}/python/revng/project/model/mixins.py.tpl"
"${CMAKE_BINARY_DIR}/${PYTHON_INSTALL_PATH}/revng/project/model/mixins.py"
@@ -260,9 +261,22 @@ add_custom_command(
"${CMAKE_SOURCE_DIR}/python/revng/project/model/mixins.py.tpl"
WORKING_DIRECTORY "${CMAKE_BINARY_DIR}")
add_custom_command(
COMMAND
"${CMAKE_SOURCE_DIR}/scripts/jsonschema_to_py.sh"
"${CMAKE_SOURCE_DIR}/python/revng/pypeline/pipeline-description-schema.yml"
"${CMAKE_BINARY_DIR}/${PYTHON_INSTALL_PATH}/revng/project/pipeline_description.py"
"PipelineDescription"
OUTPUT
"${CMAKE_BINARY_DIR}/${PYTHON_INSTALL_PATH}/revng/project/pipeline_description.py"
DEPENDS
"${CMAKE_SOURCE_DIR}/scripts/jsonschema_to_py.sh"
"${CMAKE_SOURCE_DIR}/python/revng/pypeline/pipeline-description-schema.yml")
set(REVNG_PROJECT_MODULE_FILES
revng/project/__init__.py
revng/project/cli_project.py
revng/project/common.py
revng/project/daemon_project.py
revng/project/project.py
revng/project/local_daemon_project.py
@@ -277,7 +291,8 @@ python_module(
${REVNG_PROJECT_MODULE_FILES}
MODULE_GENERATED_FILES
revng/project/model/_generated.py
revng/project/model/mixins.py)
revng/project/model/mixins.py
revng/project/pipeline_description.py)
#
# Install revng.ptml
python/revng/internal/cli/support.py
+3 -20
View File
@@ -8,9 +8,7 @@ import signal
import sys
from contextlib import contextmanager
from dataclasses import dataclass
from io import BytesIO
from subprocess import Popen
from tarfile import open as tar_open
from tempfile import NamedTemporaryFile
from typing import Any, Callable, Dict, Iterable, List, Literal, Mapping, NoReturn, Optional
from typing import Tuple, Union
@@ -19,7 +17,7 @@ import yaml
from revng.internal.support.collect import collect_libraries
from revng.internal.support.elf import is_executable
from revng.support import get_command
from revng.support import TarDictionary, get_command
OptionalEnv = Optional[Mapping[str, str]]
@@ -164,27 +162,12 @@ def executable_name() -> str:
return os.path.basename(sys.argv[0])
def is_tar(raw: bytes) -> bool:
return raw.startswith(b"\x1f\x8b")
def to_string(filename: str, raw: bytes) -> str:
return raw.decode("utf8")
def extract_tar(raw: bytes, process: Callable[[str, bytes], Any] = to_string) -> Dict[str, Any]:
if not is_tar(raw):
raise ValueError("A tar archive was expected")
result: Dict[str, Any] = {}
with tar_open(fileobj=BytesIO(raw), mode="r:gz") as file:
for element in file.getmembers():
extracted_element = file.extractfile(element)
if extracted_element is not None:
result[element.name] = process(element.name, extracted_element.read())
return result
def extract_tar[T](raw: bytes, process: Callable[[str, bytes], Any] = to_string) -> Dict[str, Any]:
return {key: process(key, value) for key, value in TarDictionary(raw).items()}
def to_yaml(filename: str, raw: bytes) -> str:
python/revng/project/cli_project.py
+80 -83
View File
@@ -2,109 +2,106 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import os
from tempfile import NamedTemporaryFile
from typing import Optional, Set
import hashlib
import json
import re
import shutil
from pathlib import Path
from subprocess import PIPE
from tempfile import NamedTemporaryFile, TemporaryFile
from typing import Dict, List, Optional, Set, Union
import yaml
from revng.pipeline_description import PipelineDescription # type: ignore[attr-defined]
from revng.pipeline_description import YamlLoader # type: ignore[attr-defined]
from revng.project.model import Binary, DiffSet # type: ignore[attr-defined]
from .project import CLIProjectMixin, Project, ResumeProjectMixin
from revng.project.common import ALL_OBJECTS, AllObjects, CLIHelper
from revng.project.model import Binary # type: ignore[attr-defined]
from revng.project.pipeline_description import PipelineDescription
from revng.project.project import Project
from revng.support import TarDictionary
class CLIProject(Project, CLIProjectMixin, ResumeProjectMixin):
class CLIProject(Project):
"""
This class is used to run revng analysis and artifact through
the CLI with subprocess.
the revng2 CLI with subprocess.
"""
def __init__(
self, resume_path: Optional[str] = None, revng_executable_path: Optional[str] = None
self, project_directory: Optional[str] = None, executable_path: Optional[str] = None
):
CLIProjectMixin.__init__(self, revng_executable_path)
ResumeProjectMixin.__init__(self, resume_path)
Project.__init__(self)
self._input_binary_path: Optional[str] = None
self._cli_helper = CLIHelper(project_directory, executable_path)
# The _cli_helper needs to be initialized first before calling the
# parent's init since that calls `_get_pipeline_description`
super().__init__()
self._load_model()
def set_binary_path(self, binary_path: str):
assert os.path.isfile(binary_path)
self._input_binary_path = binary_path
def upload_binary(self, binary_path: Union[str, Path]) -> str:
binary_path = Path(binary_path)
shutil.copy(binary_path, self._cli_helper.project_directory / binary_path.name)
with open(binary_path, "rb") as f:
return hashlib.file_digest(f, "sha256").hexdigest()
def _get_artifact_impl(self, artifact_name: str, targets: Set[str]) -> bytes:
assert self._input_binary_path is not None
args = [
"artifact",
f"--resume={self._resume_path}",
artifact_name,
self._input_binary_path,
*targets,
]
def _get_artifact_impl(
self,
artifact_name: str,
objects: Union[Set[str], AllObjects],
configuration: Dict[str, str],
) -> Dict[str, bytes]:
with NamedTemporaryFile() as tmp_file:
args = ["artifact", artifact_name, "--tar", "-o", tmp_file.name]
if objects is not ALL_OBJECTS:
args.append(",".join(objects))
if len(configuration) > 0:
args.extend(("-c", json.dumps(configuration)))
return self._run_revng_cli(args, capture_output=True).stdout
self._cli_helper.run(args)
tmp_file.seek(0)
return dict(TarDictionary(tmp_file))
def _analyze(self, analysis_name: str, targets={}, options={}):
assert analysis_name in self._analysis_names
self._analyze_impl(analysis_name, targets, options)
def _analyze_impl(
self,
analysis_name: str,
configuration: Dict[str, str],
containers: Dict[str, List[str]],
) -> int:
args = ["analyze", analysis_name]
def _analyses_list(self, analysis_list_name: str):
assert analysis_list_name in self._analyses_list_names
return self._analyze_impl(analysis_list_name)
for name, value in configuration.items():
if name == analysis_name and analysis_name in self._analysis_names:
# Single-analysis specialization
args.extend(("-c", value))
else:
args.extend((f"--{self._normalize_argument(value)}-configuration", value))
def _analyze_impl(self, analysis_name: str, targets={}, options={}):
assert self._input_binary_path is not None
# TODO: add `targets` when `revng` allows it
assert len(targets) == 0
for container_name, obj_list in containers.items():
if len(obj_list) > 0:
args.extend(
(f"--{self._normalize_argument(container_name)}-objects", ",".join(obj_list))
)
args = ["analyze", f"--resume={self._resume_path}", analysis_name]
if options:
for name, value in options.items():
args.append(f"--{analysis_name}-{name}={value}")
args.append(self._input_binary_path)
with TemporaryFile("w+") as temp_file:
self._cli_helper.run(args, stdout=temp_file)
temp_file.seek(0)
model = Binary.deserialize(temp_file, self)
result = self._run_revng_cli(args, capture_output=True).stdout
model = Binary.deserialize(result.decode("utf-8"))
self._set_model(model)
def _commit(self):
diff = DiffSet.make(self._last_saved_model, self.model)
if len(diff.Changes) == 0:
return
with NamedTemporaryFile(mode="w") as tmp_diff_file:
diff.serialize(tmp_diff_file)
tmp_diff_file.flush()
args = [
"analyze",
f"--resume={self._resume_path}",
"apply-diff",
"--apply-diff-global-name=model.yml",
f"--apply-diff-diff-content-path={tmp_diff_file.name}",
self._input_binary_path,
]
model = self._run_revng_cli(args, capture_output=True).stdout
self._last_saved_model = Binary.deserialize(model.decode("utf-8"))
def _get_pipeline_description(self) -> PipelineDescription:
pipeline_description_path = f"{self._resume_path}/pipeline-description.yml"
if not os.path.isfile(pipeline_description_path):
self._run_revng_cli(["init", f"--resume={self._resume_path}"])
assert os.path.isfile(pipeline_description_path)
with open(pipeline_description_path) as f:
return yaml.load(f, Loader=YamlLoader)
return self._epoch + 1
def _load_model(self):
# TODO: We hardcode the path to the `model.yml`, fix when there is a
# command to get the model.
model_path = f"{self._resume_path}/context/model.yml"
if os.path.isfile(model_path):
with open(model_path) as f:
model = Binary.deserialize(f)
else:
model = Binary()
self._set_model(model)
with open(self._cli_helper.model_path) as f:
data = f.read()
if data != "":
model = Binary.deserialize(data, self)
self._set_model(model)
def _get_pipeline_description(self) -> PipelineDescription:
process = self._cli_helper.run(["dump-pipeline"], stdout=PIPE)
return yaml.safe_load(process.stdout)
def _get_epoch(self) -> int:
return 0
@staticmethod
def _normalize_argument(input_: str) -> str:
return re.sub(r"\s+", " ", input_).strip().replace(" ", "-").replace("_", "-")
python/revng/project/common.py
+102
View File
@@ -0,0 +1,102 @@
#
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import os
import shutil
from enum import Enum, auto
from pathlib import Path
from subprocess import PIPE, CalledProcessError, Popen, run
from tempfile import TemporaryDirectory
from typing import Iterable, Optional, Union
import yaml
from revng.project.pipeline_description import PipelineDescription
# This enum and the constant below are all needed to have OK typing in the
# public project interface. Eventually PEP661 (sentinel value) can be used once
# it's merged and has widespread support.
class AllObjects(Enum):
ALL_OBJECTS = auto()
ALL_OBJECTS = AllObjects.ALL_OBJECTS
"""Marker value indicating that all the objects of the specified artifact must
be produced"""
class ProjectError(Exception):
pass
class CLIError(ProjectError):
pass
class HTTPError(ProjectError):
pass
class CLIHelper:
"""Mixin for classes that use the `revng2` command-line tool"""
def __init__(self, project_directory: Union[str, Path, None], executable_path: Optional[str]):
# Handle the executable_path parameter, using `which` if it's missing
if not executable_path:
_executable_path = shutil.which("revng2")
else:
_executable_path = executable_path
assert _executable_path is not None
assert os.access(_executable_path, os.F_OK | os.X_OK)
self._executable_path = _executable_path
# Handle the project_directory parameter, using a temporary directory
# if not supplied
if project_directory is None:
self._tmp_dir = TemporaryDirectory()
project_directory = self._tmp_dir.name
if not os.path.exists(project_directory):
os.makedirs(project_directory)
assert os.path.isdir(project_directory)
self._project_directory = str(project_directory)
# We will be invoking `revng2 project` a bunch. This family of commands
# needs the file `${model_name}` to be present in the project
# directory. Since the project directory might not contain a model,
# touch one if missing.
process = self.run(["dump-pipeline"], stdout=PIPE)
pipeline_description: PipelineDescription = yaml.safe_load(process.stdout)
model_name: str = pipeline_description["model"]["name"]
self._model_path = Path(project_directory) / model_name
if not self._model_path.exists():
self._model_path.touch()
@property
def model_path(self) -> Path:
return self._model_path
@property
def project_directory(self) -> Path:
return Path(self._project_directory)
def _make_command_line(self, cmd_args: Iterable[str]) -> list[str]:
return [self._executable_path, "-C", self._project_directory, "project", *cmd_args]
def run(self, cmd_args: Iterable[str], *args, stdout=None):
cmdline = self._make_command_line(cmd_args)
try:
return run( # type: ignore[call-overload]
cmdline, *args, stdout=stdout, stderr=PIPE, check=True
)
except CalledProcessError as e:
error_message = f"Command {" ".join(cmdline)} exited with code {e.returncode}\n"
error_message += "Stderr output:\n"
error_message += e.stderr
raise CLIError(error_message) from e
def popen(self, cmd_args: Iterable[str], *args, stdout, stderr) -> Popen:
return Popen( # type: ignore
self._make_command_line(cmd_args), *args, stdout=stdout, stderr=stderr
)
python/revng/project/daemon_project.py
+158 -149
View File
@@ -2,180 +2,189 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import json
from base64 import b64decode
from typing import Dict, List, Set
import os
from io import IOBase
from pathlib import Path
from tempfile import SpooledTemporaryFile
from typing import Dict, List, Set, Union
from urllib.parse import urlparse
import yaml
from gql import Client, gql
from gql.transport.aiohttp import AIOHTTPTransport
import requests
from urllib3.response import HTTPResponse
from revng.pipeline_description import PipelineDescription # type: ignore[attr-defined]
from revng.pipeline_description import YamlLoader # type: ignore[attr-defined]
from revng.project.model import Binary, DiffSet # type: ignore[attr-defined]
from revng.project.common import ALL_OBJECTS, AllObjects, HTTPError, ProjectError
from revng.project.model import Binary # type: ignore[attr-defined]
from revng.project.project import Project
from revng.support import TarDictionary
from .project import Project
class BufferedReader(IOBase):
"""
Adapter class that buffers an urllib3 HTTPResponse content in a spooled file
"""
CHUNK_SIZE = 1 * 1024 * 1024
def __init__(self, response: HTTPResponse):
self.response = response
self.file = SpooledTemporaryFile(max_size=2 * 1024 * 1024)
self.max_offset = 0
self.position = 0
self.end = False
def close(self):
self.response.close()
self.file.close()
def fileno(self):
# We do not want to return self.file.fileno(), we want users to use
# read instead so we can perform buffering gradually
raise OSError
def seekable(self) -> bool:
return True
def readable(self) -> bool:
return True
def writable(self) -> bool:
return False
def read(self, size: int = -1) -> bytes:
if size == 0:
return b""
if size == -1 and not self.end:
self._read_internal(-1)
if size > 0 and self.position + size > self.max_offset:
self._read_internal(self.position + size - self.max_offset)
self.file.seek(self.position, os.SEEK_SET)
result = self.file.read(size)
self.position += len(result)
return result
def _read_internal(self, size: int):
chunk_size = self.__class__.CHUNK_SIZE
self.file.seek(self.max_offset)
if size == -1:
while (buffer := self.response.read(chunk_size)) != b"":
self.file.write(buffer)
self.max_offset += len(buffer)
self.end = True
else:
while size > 0:
size_to_read = chunk_size if size > chunk_size else size
buffer = self.response.read(size_to_read)
if buffer == b"":
self.end = True
break
self.file.write(buffer)
self.max_offset += len(buffer)
size -= len(buffer)
def seek(self, offset: int, whence=os.SEEK_SET):
if whence == os.SEEK_SET:
self.position = offset
return self.position
elif whence == os.SEEK_CUR:
self.position += offset
return self.position
else:
raise ValueError
def tell(self) -> int:
return self.position
class DaemonProject(Project):
"""
This class is used to run revng analysis and artifact through the
revng daemon via GraphQL.
This class is used to run revng analysis and artifact through
the revng daemon via HTTP.
"""
def __init__(self, url: str):
self.client = Client(
transport=AIOHTTPTransport(url=url),
fetch_schema_from_transport=True,
execute_timeout=None,
)
self._base_url = urlparse(url)
self._session = requests.Session()
self._session.hooks["response"].append(self._response_hook)
# The constructor calls `_get_pipeline_description`, so we need to
# initialize our variables first
super().__init__()
self._set_model(self._get_model())
def set_binary_path(self, binary_path: str):
query = gql(
"""
mutation upload($file: Upload!) {
uploadFile(file: $file, container: "input")
}
"""
)
@staticmethod
def _response_hook(response: requests.Response, *args, **kwargs):
try:
response.raise_for_status()
except requests.RequestException as e:
raise HTTPError from e
with open(binary_path, "rb") as binary_file:
result = self.client.execute(
query, variable_values={"file": binary_file}, upload_files=True
def upload_binary(self, binary_path: Union[str, Path]) -> str:
with open(binary_path, "rb") as f:
response = self._session.post(
self._join_url("/api/put-file"),
files={"file": (Path(binary_path).name, f)},
)
return response.json()["hash"]
if not result["uploadFile"]:
raise RuntimeError("File upload failed")
def _get_artifact_impl(self, artifact_name: str, targets: Set[str]) -> Dict[str, bytes]:
query = gql(
"""
query($step: String!, $paths: String!, $index: BigInt!) {
produceArtifacts(step: $step, paths: $paths, index: $index) {
__typename
... on Produced {
result
}
}
}
"""
)
variables = {
"step": artifact_name,
"paths": ",".join(targets),
"index": self._get_content_commit_index(),
def _get_artifact_impl(
self,
artifact_name: str,
objects: Union[Set[str], AllObjects],
configuration: Dict[str, str],
) -> Dict[str, bytes]:
body: dict = {
"artifact": artifact_name,
"epoch": self._get_epoch(),
"format": "tar",
}
if objects is not ALL_OBJECTS:
body["objects"] = list(objects)
if len(configuration) > 0:
body["configuration"] = configuration
response = self.client.execute(query, variable_values=variables)
assert response["produceArtifacts"]["__typename"] == "Produced"
data: Dict[str, str] = json.loads(response["produceArtifacts"]["result"])
container_mime = self._get_artifact_container(artifact_name).MIMEType
return {
k.rsplit(":", 1)[0]: self._decode_result_on_mime(v, container_mime)
for k, v in data.items()
}
response = self._session.post(self._join_url("/api/artifact"), json=body, stream=True)
# TODO: right now we return the response as a dict, with BufferedReader
# we could return things more lazily and leverage pipelining
# (e.g. parsing input while the rest of the request is incoming)
return dict(TarDictionary(BufferedReader(response.raw)))
def _mapped_artifact_mime(self, artifact_name: str) -> str:
container_mime = self._get_artifact_container(artifact_name).MIMEType
if container_mime.endswith("+tar+gz"):
return container_mime.removesuffix("+tar+gz")
raise ValueError
def _analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
):
step = self._get_step_name(analysis_name)
variables = {
"step": step,
def _analyze_impl(
self,
analysis_name: str,
configuration: Dict[str, str] = {},
containers: Dict[str, List[str]] = {},
) -> int:
body = {
"analysis": analysis_name,
"container": json.dumps(targets),
"options": json.dumps(options),
"index": self._get_content_commit_index(),
"epoch": self._epoch,
"configuration": configuration,
"containers": containers,
}
query = gql(
"""
mutation($step: String!, $analysis: String!, $container: String!,
$options: String!, $index: BigInt!)
{
runAnalysis(step: $step, analysis: $analysis, options: $options,
containerToTargets: $container, index: $index)
{
__typename
}
}
"""
)
if analysis_name not in self._analyses_list and analysis_name not in self._analysis_names:
raise ProjectError(f"Analysis {analysis_name} does not exist")
response = self.client.execute(query, variable_values=variables)
assert response["runAnalysis"]["__typename"] == "Diff"
req = self._session.post(self._join_url("/api/analysis"), json=body)
self._set_model(self._get_model())
return req.json()["epoch"]
def _analyses_list(self, analysis_name: str):
query = gql(
"""
mutation($analysis: String!, $index: BigInt!) {
runAnalysesList(name: $analysis, index: $index) {
__typename
}
}
"""
)
variables = {"analysis": analysis_name, "index": self._get_content_commit_index()}
response = self.client.execute(query, variable_values=variables)
assert response["runAnalysesList"]["__typename"] == "Diff"
self._set_model(self._get_model())
def _commit(self):
diff = DiffSet.make(self._last_saved_model, self.model)
if len(diff.Changes) == 0:
return
diff_yaml = diff.serialize()
query = gql(
"""
mutation ($options: String!, $index: BigInt!) {
runAnalysis(step: "initial", analysis: "apply-diff", index: $index,
options: $options) {
__typename
... on Diff {
diff
}
}
}
"""
)
variables = {
"options": json.dumps(
{"apply-diff-global-name": "model.yml", "apply-diff-diff-content": diff_yaml}
),
"index": self._get_content_commit_index(),
}
response = self.client.execute(query, variable_values=variables)
assert response["runAnalysis"]["__typename"] == "Diff"
self._last_saved_model = self._get_model()
def _get_pipeline_description(self) -> PipelineDescription:
response = self.client.execute(gql("{ pipelineDescription }"))
return yaml.load(response["pipelineDescription"], Loader=YamlLoader)
def _decode_result_on_mime(self, result: str, container_mime: str) -> bytes:
if container_mime.startswith("text/") or container_mime == "image/svg":
return result.encode("utf-8")
else:
return b64decode(result)
def _get_pipeline_description(self):
response = self._session.get(self._join_url("/api/pipeline"))
return response.json()
def _get_model(self) -> Binary:
response = self.client.execute(gql("""{ getGlobal(name: "model.yml") }"""))
return Binary.deserialize(response["getGlobal"])
response = self._session.get(self._join_url("/api/model"))
data = response.json()
return Binary.deserialize(data["model"], self)
def _get_content_commit_index(self) -> int:
# TODO: Keep track of the commit index with websockets
response = self.client.execute(gql("{ contextCommitIndex }"))
return int(response["contextCommitIndex"])
def _get_epoch(self) -> int:
response = self._session.get(self._join_url("/api/epoch"))
return response.json()["epoch"]
def _join_url(self, path: str) -> str:
new_path = os.path.normpath(self._base_url.path + path)
return self._base_url._replace(path=new_path).geturl()
python/revng/project/local_daemon_project.py
+29 -29
View File
@@ -2,20 +2,22 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import os
import socket
from pathlib import Path
from signal import SIGINT
from subprocess import DEVNULL, STDOUT, Popen
from time import sleep
from typing import Optional
from urllib.error import URLError
from urllib.request import urlopen
from typing import Optional, Union
from .daemon_project import DaemonProject
from .project import CLIProjectMixin, ResumeProjectMixin
import requests
from requests.exceptions import RequestException
from urllib3.exceptions import HTTPError
from revng.project.common import CLIHelper
from revng.project.daemon_project import DaemonProject
class LocalDaemonProject(DaemonProject, CLIProjectMixin, ResumeProjectMixin):
class LocalDaemonProject(DaemonProject):
"""
This class extends the DaemonProject. When initialized it starts the revng daemon
server and setups the DaemonProject client used to connect to the server.
@@ -23,41 +25,39 @@ class LocalDaemonProject(DaemonProject, CLIProjectMixin, ResumeProjectMixin):
def __init__(
self,
resume_path: Optional[str] = None,
revng_executable_path: Optional[str] = None,
project_directory: Union[str, Path, None] = None,
executable_path: Optional[str] = None,
*,
connection_retries: int = 10,
):
CLIProjectMixin.__init__(self, revng_executable_path)
ResumeProjectMixin.__init__(self, resume_path)
self._cli_helper = CLIHelper(project_directory, executable_path)
self._host = "127.0.0.1"
self._port: int = self._get_port()
self._daemon_process: Optional[Popen] = None
self.start_daemon(connection_retries)
super().__init__(f"http://127.0.0.1:{self._port}/graphql/")
self._daemon_url = f"http://{self._host}:{self._port}/"
self._daemon_process: Optional[Popen] = self._start_daemon(connection_retries)
# We need to delay calling the parent's constructor because we first
# needed to start the server that the parent will connect to.
super().__init__(self._daemon_url)
def __del__(self):
self.stop_daemon()
self._stop_daemon()
def start_daemon(self, connection_retries: int):
def _start_daemon(self, connection_retries: int):
"""
Start the `revng` daemon and wait for it to be ready.
Start the `revng2 project daemon` and wait for it to be ready.
"""
env = os.environ
env["REVNG_DATA_DIR"] = self._resume_path
cli_args = [self._revng_executable_path, "daemon", "-b", f"tcp:127.0.0.1:{self._port}"]
self._daemon_process = Popen(cli_args, stdout=DEVNULL, stderr=STDOUT, env=env)
cli_args = ["daemon", "--host", self._host, "--port", str(self._port)]
process = self._cli_helper.popen(cli_args, stdout=DEVNULL, stderr=STDOUT)
failed_retries = 0
while failed_retries <= connection_retries:
if self._is_server_running():
return
return process
failed_retries += 1
sleep(1)
raise RuntimeError(f"Couldn't connect to daemon server at http://127.0.0.1:{self._port}")
raise RuntimeError(f"Couldn't connect to daemon server at {self._daemon_url}")
def stop_daemon(self) -> int:
def _stop_daemon(self) -> int:
"""
Stop the daemon server.
"""
@@ -70,9 +70,9 @@ class LocalDaemonProject(DaemonProject, CLIProjectMixin, ResumeProjectMixin):
def _is_server_running(self) -> bool:
try:
urlopen(f"http://127.0.0.1:{self._port}/status", timeout=5)
return True
except URLError:
req = requests.get(f"{self._daemon_url}status", timeout=5)
return req.ok
except (RequestException, HTTPError):
return False
def _get_port(self) -> int:
python/revng/project/model/__init__.py
+3 -1
View File
@@ -11,6 +11,7 @@ import yaml
from revng.model.metaaddress import *
from revng.model.metaaddress import init_metaaddress_yaml_classes
from revng.support import IgnoreDeepCopy
from revng.tupletree import DiffSet as TTDiffSet
from revng.tupletree import Reference, StructBase, TypesMetadata, _FieldVisitor
from revng.tupletree import _get_element_by_path, init_reference_yaml_classes
@@ -28,13 +29,14 @@ init_reference_yaml_classes(DiffYamlLoader, DiffYamlDumper)
class Binary(_generated.Binary):
@staticmethod
def deserialize(input_: Union[str, TextIOBase], project) -> Binary:
project_idc = IgnoreDeepCopy(project)
obj = yaml.load(input_, Loader=YamlLoader)
accessor = lambda x: get_element_by_path(x, obj)
for _, field_obj in iterate_fields(obj):
if isinstance(field_obj, Reference) and field_obj.is_valid():
field_obj._accessor = accessor
if isinstance(field_obj, AllMixin):
field_obj._project = project
field_obj._project = project_idc
return obj
python/revng/project/model/mixins.py.tpl
+69 -73
View File
@@ -2,13 +2,21 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
from typing import Dict, Iterable, List, Mapping, Optional, TypeAlias, Union
from enum import Enum, auto
from typing import TYPE_CHECKING, Dict, List, Mapping, Optional, Sequence, Union, TypeVar
import revng.support.artifacts as artifacts
from revng.project.common import AllObjects, ALL_OBJECTS
from revng.support import IgnoreDeepCopy
from revng.support.artifacts import Artifact
from revng.tupletree import StructBase
ArtifactResult: TypeAlias = Union[Artifact, Mapping[str, Artifact]]
StructBaseT = TypeVar("StructBaseT", bound=StructBase)
def _only(dict_: Mapping[str, Artifact]) -> Artifact:
assert isinstance(dict_, dict)
assert len(dict_) == 1
return dict_[list(dict_)[0]]
class AllMixin:
@@ -17,7 +25,55 @@ class AllMixin:
"""
_project = None
_keys = None
_location = None
class BinaryMixin(AllMixin):
@property
def _location(self):
return "/binary"
def get_artifact(
self,
artifact_name: str,
objects: Union[Sequence[Union[str, StructBaseT]], AllObjects] = ALL_OBJECTS,
) -> Mapping[str, Artifact]:
"""
Fetch the artifacts from the `Binary`.
"""
return self._project.get()._get_artifact(artifact_name, objects) # type: ignore[attr-defined]
def commit(self):
"""
Persist the changes to the backend.
"""
self._project.get()._commit() # type: ignore[attr-defined]
def revert(self):
"""
Revert changes made since the last call to `commit()`.
"""
self._project.get()._revert() # type: ignore[attr-defined]
def analyze(
self,
analysis_name: str,
configuration: Dict[str, str] = {},
containers: Dict[str, List[str]] = {},
):
"""
Run a single analysis or an analysis list. In addition to the
`analysis_name` you can optionally specify a dict of objects in
`containers`. Some analyses require a configuration, which can be
passed in the `configuration` dict.
"""
return self._project.get()._analyze(analysis_name, configuration, containers)
{% for artifact in binary_artifacts %}
@property
def {{ artifact.name | normalize }}(self) -> artifacts.{{ artifact.type_ }}:
return _only(self.get_artifact("{{ artifact.name }}"))
{% endfor %}
class _ArtifactMixin(AllMixin):
@@ -25,80 +81,18 @@ class _ArtifactMixin(AllMixin):
Mixin used to implement the get_artifact function
"""
def get_artifact(self, artifact_name: str) -> Artifact:
def get_artifact(self, name: str) -> Artifact:
"""
Fetch the artifacts that belong to the current model entity
"""
result = self._project._get_artifact(artifact_name, [self]) # type: ignore[attr-defined]
if isinstance(result, Mapping):
keys = list(result.keys())
assert len(keys) == 1
return result[keys[0]]
else:
return result
class BinaryMixin(_ArtifactMixin):
def get_artifact(self, artifact_name: str, targets=None):
"""
Fetch the artifacts from the `Binary`.
"""
if targets is None:
return self._project._get_artifact(artifact_name, [self]) # type: ignore[attr-defined]
else:
return self._project._get_artifact(artifact_name, targets) # type: ignore[attr-defined]
def commit(self):
"""
Persist the changes to the backend.
"""
self._project._commit() # type: ignore[attr-defined]
def revert(self):
"""
Revert changes made since the last call to `commit()`.
"""
self._project._revert() # type: ignore[attr-defined]
def get_artifacts(
self, params: Dict[str, Optional[Iterable[Union[str, StructBase]]]]
) -> Dict[str, ArtifactResult]:
"""
Allows fetching multiple artifacts at once. The `params` is a dict
containing the name of the artifact and a list of targets (it can be
empty to fetch all the targets).
Example `params`:
params = {
"disassemble": [Function_1, Function_2],
"decompile": []
}
"""
return self._project._get_artifacts(params) # type: ignore[attr-defined]
def analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
):
"""
Run a single analysis. In addition to the `analysis_name` you need to
specify a dict of targets, some analysis require you to also pass an
`options` dict.
"""
return self._project._analyze(analysis_name, targets, options) # type: ignore[attr-defined]
def analyses_list(self, analysis_name: str):
"""
Run analysis list, these are predefined list of analysis that run sequentially.
"""
return self._project._analyses_list(analysis_name) # type: ignore[attr-defined]
{% for artifact in binary_artifacts %}
@property
def {{ artifact.name | normalize }}(self) -> artifacts.{{ artifact.type_ }}:
return self.get_artifact("{{ artifact.name }}")
{% endfor %}
return _only(self._project.get()._get_artifact(name, [self])) # type: ignore[attr-defined]
class FunctionMixin(_ArtifactMixin):
@property
def _location(self):
return f"/function/{self.key()}"
{% for artifact in function_artifacts %}
@property
def {{ artifact.name | normalize }}(self) -> artifacts.{{ artifact.type_ }}:
@@ -107,10 +101,12 @@ class FunctionMixin(_ArtifactMixin):
class TypeDefinitionMixin(_ArtifactMixin):
@property
def _location(self):
return f"/type-definition/{self.key()}"
{% for artifact in typedefinition_artifacts %}
@property
def {{ artifact.name | normalize }}(self) -> artifacts.{{ artifact.type_ }}:
return self.get_artifact("{{ artifact.name }}")
{% else %}
pass
{% endfor %}
python/revng/project/project.py
+129 -220
View File
@@ -2,28 +2,34 @@
# This file is distributed under the MIT License. See LICENSE.md for details.
#
import os
import re
import shutil
from abc import ABC, abstractmethod
from collections import defaultdict
from copy import deepcopy
from itertools import chain
from subprocess import CalledProcessError, run
from tempfile import TemporaryDirectory
from typing import Dict, Iterable, List, Mapping, Optional, Sequence, Set, TypeAlias, TypeVar
from typing import Union
from pathlib import Path
from typing import Dict, List, Mapping, Sequence, Set, TypeVar, Union
from revng.pipeline_description import Container, PipelineDescription # type: ignore[attr-defined]
from revng.project.model import Binary, iterate_fields # type: ignore[attr-defined]
from revng.project.common import ALL_OBJECTS, AllObjects, ProjectError
from revng.project.model import Binary, BinaryIdentifier, DiffSet # type: ignore[attr-defined]
from revng.project.model.mixins import AllMixin
from revng.project.pipeline_description import PipelineDescription
from revng.support.artifacts import Artifact
from revng.tupletree import StructBase
StructBaseT = TypeVar("StructBaseT", bound=StructBase)
ArtifactResult: TypeAlias = Union[Artifact, Mapping[str, Artifact]]
# TODO: In the future we need to have more features for
# interactivity/multi-user:
# 1. Cache artifacts
# 2. Handle cache invalidations
# 3. Have some form of locking of the model to enable running
# "client-side analysis". Without this, complex analysis in Python would
# be problematic since, without a lock, we might spend a long time doing
# our computations and then just fail because someone else has renamed a
# function.
# The idea would be to have a Python context manager holding the lock.
# 4. We might also have to offer an `async` flavor of the class project.
# The non-`async` version would have an opt-in thread to monitor
# notifications coming from the websocket.
class Project(ABC):
"""
This class is the basis for building a project. It defines
@@ -34,158 +40,121 @@ class Project(ABC):
def __init__(self):
self.model: Binary = Binary()
self._last_saved_model: Binary = Binary()
self._epoch = self._get_epoch()
self._pipeline_description: PipelineDescription = self._get_pipeline_description()
self._rank_matchers: Dict[str, re.Pattern] = self._compute_rank_matchers()
self._rank_targets: Dict[str, List[str]] = defaultdict(list)
self._analysis_names = list(
chain.from_iterable(
[an.Name for an in st.Analyses] for st in self._pipeline_description.Steps
)
)
self._analyses_list_names = [x.Name for x in self._pipeline_description.AnalysesLists]
self._analysis_names: Set[str] = {
entry["name"] for entry in self._pipeline_description["analyses"]
}
self._analyses_list: Dict[str, List[str]] = {
entry["name"]: entry["analyses"]
for entry in self._pipeline_description["analyses_lists"]
}
# Compute the container type of an artifact, these will be used later
# to quickly get the mime type or kind of an artifact
container_types = {e["class"]: e for e in self._pipeline_description["container_types"]}
self._artifact_container_types = {}
for entry in self._pipeline_description["artifacts"]:
container_type = self._pipeline_description["containers"][entry["container"]]
self._artifact_container_types[entry["name"]] = container_types[container_type]
@abstractmethod
def set_binary_path(self, binary_path: str):
def upload_binary(self, binary_path: Union[str, Path]) -> str:
"""
Use this method to import the binary into the project. This method is a
prerequisite before running any analyses or getting artifacts.
Add the specified file to storage. Will return the hash of the file
with which the file can be referred to.
"""
pass
def list_targets(self, artifact_name: str) -> List[str]:
"""Retrieve the list of targets given the artifact"""
kind_name = self._pipeline_description.Steps[artifact_name].Artifacts.Kind
rank_name = self._pipeline_description.Kinds[kind_name].Rank
return self._rank_targets[rank_name]
def _get_artifact(
self, artifact_name: str, targets: Optional[Sequence[Union[str, StructBaseT]]] = None
) -> ArtifactResult:
self,
artifact_name: str,
objects: Union[Sequence[Union[str, StructBaseT]], AllObjects] = ALL_OBJECTS,
configuration: Dict[str, str] = {},
) -> Mapping[str, Artifact]:
"""
Use the method to fetch one or more targets from an artifact.
`targets` is a list of targets to produce, each element can be either a
string which is interpreted as a target string or a model object, which
will be converted to the corresponding target string.
If the `targets` parameter is omitted then all targets of the specified
Use the method to fetch one or more objects from an artifact.
`objects` is a list of objects to produce, each element can be either a
string which is interpreted as a ObjectID string or a model object, which
will be converted to the corresponding ObjectID string.
If the `objects` parameter is omitted then all objects of the specified
artifacts are fetched.
This method automatically performs MIME conversion and dispatches
the work to the `_get_artifact`.
The return value can either be a single Artifact object or a dictionary
of them.
The return value is `Dict[ObjectID, Artifact]` which maps each object
to the corresponding artifact.
"""
all_targets = set(self.list_targets(artifact_name))
if targets is None or (
len(targets) == 1
and isinstance(targets[0], StructBase)
and isinstance(targets[0], AllMixin)
and targets[0]._keys == []
):
actual_targets = all_targets
actual_objects: Set[str] | AllObjects
if objects is ALL_OBJECTS:
actual_objects = ALL_OBJECTS
else:
actual_targets = set()
for target in targets:
if isinstance(target, str):
actual_targets.add(target)
elif isinstance(target, StructBase) and isinstance(target, AllMixin):
actual_targets.add("/".join(target._keys))
actual_objects = set()
for object_ in objects:
if isinstance(object_, str):
actual_objects.add(object_)
elif isinstance(object_, AllMixin):
assert object_._location is not None
actual_objects.add(object_._location)
else:
raise ValueError(f"Unknown parameter: {target}")
assert len(actual_targets) > 0
raise ValueError(f"Unknown parameter: {object_}")
if len(actual_objects) == 0:
return {}
assert actual_targets.issubset(all_targets)
results = self._get_artifact_impl(artifact_name, actual_targets)
container_mime = self._get_artifact_container(artifact_name).MIMEType
kind_name = self._pipeline_description.Steps[artifact_name].Artifacts.Kind
rank_name = self._pipeline_description.Kinds[kind_name].Rank
rank = self._pipeline_description.Ranks[rank_name]
if isinstance(results, dict):
if rank.Depth == 0:
keys = list(results.keys())
assert len(keys) == 1
return Artifact.make(results[keys[0]], container_mime)
else:
container_mime = self._mapped_artifact_mime(artifact_name)
return {k: Artifact.make(v, container_mime) for k, v in results.items()}
else:
return Artifact.make(results, container_mime)
results = self._get_artifact_impl(artifact_name, actual_objects, configuration)
container_mime = self._artifact_container_types[artifact_name]["mime_type"]
return {k: Artifact.make(v, container_mime) for k, v in results.items()}
@abstractmethod
def _get_artifact_impl(
self, artifact_name: str, targets: Set[str]
) -> Union[bytes, Dict[str, bytes]]:
self,
artifact_name: str,
objects: Union[Set[str], AllObjects],
configuration: Dict[str, str],
) -> Dict[str, bytes]:
"""
This method implements the actual fetching of the artifact(s). It is
implemented by the children of this class.
"""
pass
def _mapped_artifact_mime(self, artifact_name: str) -> str:
"""
Return the artifact's MIME if _get_artifact returns a mapping. By
default throws an exception. Can be overridden by child classes.
"""
raise ValueError
def _get_artifacts(
self, params: Dict[str, Optional[Sequence[Union[str, StructBase]]]]
) -> Dict[str, ArtifactResult]:
"""
Allows fetching multiple artifacts at once. The `params` is a dict
containing the name of the artifact and a list of targets (it can be
empty to fetch all the targets).
Example `params`:
params = {
"disassemble": [Function_1, Function_2],
"decompile": []
}
"""
result: Dict[str, ArtifactResult] = {}
for artifact_name, targets in params.items():
result[artifact_name] = self._get_artifact(artifact_name, targets)
return result
@abstractmethod
def _analyze(
self, analysis_name: str, targets: Dict[str, List[str]] = {}, options: Dict[str, str] = {}
self,
analysis_name: str,
configuration: Dict[str, str] = {},
containers: Dict[str, List[str]] = {},
):
"""
Run a single analysis. In addition to the `analysis_name` you need to
specify a dict of targets, some analysis require you to also pass an
`options` dict.
Run a single analysis or an analysis list. In addition to the
`analysis_name` you can optionally specify a dict of objects in
`containers`. Some analyses require a configuration, which can be
passed in the `configuration` dict.
"""
if analysis_name not in self._analyses_list and analysis_name not in self._analysis_names:
raise ProjectError(f"Analysis {analysis_name} does not exist")
self._epoch = self._analyze_impl(analysis_name, configuration, containers)
@abstractmethod
def _analyze_impl(
self,
analysis_name: str,
configuration: Dict[str, str],
containers: Dict[str, List[str]],
) -> int:
"""
Actual method that subclasses must implement to run an analysis, it's
called when preliminary checks have already run, so it's guaranteed
that all the inputs are well-formed.
Returns the new epoch after running the analysis (list).
"""
pass
@abstractmethod
def _analyses_list(self, analysis_name: str):
"""
Run analysis list, these are predefined list of analysis that run sequentially.
"""
pass
def get_analysis_inputs(self, analysis_name: str) -> Dict[str, List[str]]:
"""
Get the analysis container inputs and the associate acceptable kinds from
the `pipeline description`.
"""
inputs: Dict[str, List[str]] = {}
for step in self._pipeline_description.Steps:
if analysis_name in step.Analyses:
for i in step.Analyses[analysis_name].ContainerInputs:
inputs[i.Name] = i.AcceptableKinds
break
return inputs
@abstractmethod
def _commit(self):
"""
Persist the changes from `self.model` to the backend.
"""
pass
diff = DiffSet.make(self._last_saved_model, self.model)
if len(diff.Changes) == 0:
return
self._analyze("apply-diff", configuration={"apply-diff": diff.serialize()})
def _revert(self):
"""
@@ -193,6 +162,14 @@ class Project(ABC):
"""
self._set_model(deepcopy(self._last_saved_model))
@abstractmethod
def _get_epoch(self) -> int:
"""
Get the current epoch at rest. Only done at startup, afterwards the
epoch will be tracked manually assuming we're the only users.
"""
pass
@abstractmethod
def _get_pipeline_description(self) -> PipelineDescription:
"""
@@ -201,97 +178,29 @@ class Project(ABC):
"""
pass
def import_and_analyze(self, binary_path: str):
def import_and_analyze(self, binary_path: Union[str, Path]):
"""
A helper method for setting up the project. This method imports
the binary and runs the `revng` initial auto analysis.
the binary and runs the initial auto analysis.
"""
self.set_binary_path(binary_path)
self._analyses_list("revng-initial-auto-analysis")
# TODO: remove once we have multi-binary in place
if len(self.model.Binaries) > 0:
raise ProjectError(
"Cannot import a binary when there is one already present in the model"
)
# TODO: maybe have a method of that allows checking if the file is
# already present
hash_ = self.upload_binary(binary_path)
size = Path(binary_path).stat().st_size
self.model.Binaries.append(
BinaryIdentifier(Index=0, Hash=hash_, Size=size, Name=Path(binary_path).name)
)
self._commit()
self._analyze("initial-auto-analysis")
def _set_model(self, model: Binary):
self.model = model
self._last_saved_model = deepcopy(self.model)
# visit the model, and:
# * Collect the keys based on _rank_matchers
# * Fill _project on all classes
# * Fill _keys where needed and add it to _rank_targets
new_targets = defaultdict(list)
for path, obj in iterate_fields(self.model):
if not isinstance(obj, AllMixin):
continue
obj._project = self # type: ignore
for rank, pattern in self._rank_matchers.items():
match = pattern.match(path)
if match is not None:
keys = [k.split("::", 1)[1] if "::" in k else k for k in match.groups()]
obj._keys = keys # type: ignore
new_targets[rank].append("/".join(keys))
self._rank_targets = new_targets
def _get_step_name(self, analysis_name: str) -> str:
for step in self._pipeline_description.Steps:
if analysis_name in step.Analyses:
return step.Name
raise RuntimeError(f"Couldn't find step for analysis: {analysis_name}")
def _get_artifact_container(self, artifact_name: str) -> Container:
artifact_container = self._pipeline_description.Steps[artifact_name].Artifacts.Container
container = self._pipeline_description.Containers[artifact_container]
if container is None:
raise RuntimeError(f"Couldn't find step name {artifact_name}")
return container
def _compute_rank_matchers(self) -> Dict[str, re.Pattern]:
result = {}
for rank in self._pipeline_description.Ranks:
if rank.TupleTreePath == "":
continue
path = rank.TupleTreePath
pattern = "^" + re.sub(r"\\\$\d+", "([^/]*)", re.escape(path)) + "$"
result[rank.Name] = re.compile(pattern)
return result
class CLIError(Exception):
pass
class CLIProjectMixin:
"""Mixin for classes that use the `revng` command-line tool"""
def __init__(self, revng_executable_path: Optional[str]):
"""
Check if the path of the user supplied or default revng executable
is present on the system
"""
if not revng_executable_path:
_revng_executable_path = shutil.which("revng")
else:
_revng_executable_path = revng_executable_path
assert _revng_executable_path is not None
assert os.access(_revng_executable_path, os.F_OK | os.X_OK)
self._revng_executable_path = _revng_executable_path
def _run_revng_cli(self, cmd_args: Iterable[str], *args, **kwargs):
try:
return run([self._revng_executable_path, *cmd_args], *args, **kwargs, check=True)
except CalledProcessError as e:
raise CLIError() from e
class ResumeProjectMixin:
"""Mixin for classes that use a local resume directory"""
def __init__(self, resume_path: Optional[str]):
if resume_path is None:
self._tmp_dir = TemporaryDirectory()
resume_path = self._tmp_dir.name
if not os.path.exists(resume_path):
os.makedirs(resume_path)
assert os.path.isdir(resume_path)
self._resume_path = resume_path
python/revng/pyproject.toml
+2 -2
View File
@@ -10,10 +10,10 @@ dependencies = [
"PyYAML>=6.0.1,<7.0",
"yachalk>=0.1.7,<1.0",
"llvmcpy>=0.2.1,<1.0",
"gql[aiohttp]>=3.5.3,<4.0",
"aiohttp>=3.12.11,<4.0",
"zstandard>=0.23.0,<1.0",
"pycparser>=2.22,<3.0",
"requests>=2.32.5,<3.0",
"requests-toolbelt>=1.0.0,<2.0",
]
[build-system]
python/revng/support/__init__.py
+51 -1
View File
@@ -5,10 +5,13 @@ import mmap
import os
import re
import sys
import tarfile
from collections.abc import Mapping
from contextlib import contextmanager
from io import BytesIO, IOBase
from pathlib import Path
from shutil import which
from typing import IO, Generator, Iterable, List, Optional, TypeVar, Union, cast
from typing import IO, Generator, Generic, Iterable, List, Optional, TypeVar, Union, cast
from llvmcpy import LLVMCPy
@@ -136,3 +139,50 @@ def get_example_binary_path() -> str:
if re.match(r"^calc-x86-64-static-revng-qa\.compiled-[0-9a-f]{8}$", entry.name):
return str(entry.resolve())
raise ValueError("Could not find calc binary")
class TarDictionary(Mapping):
"""Class that allows wrapping a tar file and treat it as a dictionary, the
entries are eagerly computed, the content of the files will be lazily
returned when calling __getitem__."""
def __init__(self, data: Union[IO[bytes], IOBase, bytes], mode="r"):
fileobj: IO[bytes] | IOBase
if isinstance(data, bytes):
fileobj = BytesIO(data)
else:
fileobj = data
self._tar_file: tarfile.TarFile = tarfile.open(fileobj=fileobj, mode=mode)
self._keys = {m.name: m for m in self._tar_file.getmembers()}
def __getitem__(self, key: str) -> bytes:
if key not in self._keys:
raise KeyError
member = self._tar_file.extractfile(self._keys[key])
assert member is not None
return member.read()
def __contains__(self, key) -> bool:
return key in self._keys
def __len__(self) -> int:
return len(self._keys)
def __iter__(self):
return iter(self._keys)
class IgnoreDeepCopy(Generic[T]):
"""This class is a wrapper class that allows to have a value that will not
be deep-copied when using the deepcopy library function.
"""
def __init__(self, value: T):
self.value: T = value
def get(self) -> T:
return self.value
def __deepcopy__(self, memo):
return self
scripts/jsonschema_to_py.sh
Executable
+28
View File
@@ -0,0 +1,28 @@
#!/usr/bin/env bash
set -euo pipefail
#
# This file is distributed under the MIT License. See LICENSE.md for details.
#
# This script converts a jsonschema json/yaml file to a python file with
# TypedDict declarations, useful for having typing hints and checks on the data
# represented by the schema.
INPUT="$1"
OUTPUT="$2"
ROOT_NAME="$3"
# The python package only works with JSON, use `yq` to normalize both YAML
# and JSON inputs into JSON
python -m jsonschema_to_typeddict \
--output-path /dev/stdout \
--root-name "$ROOT_NAME" \
<(yq . "$INPUT") | \
# The template used by the python package is a bit ugly and uses
# `typing_extensions` to be compatible with old python versions. Use `sed` to
# make the output look a bit nicer.
sed \
-e 's;typ\.;;' \
-e 's;import typing_extensions as typ;from typing import Literal, TypedDict, TypeAlias;' \
> "$OUTPUT"