From c53e615fa0fecc7fdcdf34bf2a03dee96206935b Mon Sep 17 00:00:00 2001 From: Giacomo Vercesi Date: Tue, 24 Feb 2026 13:58:53 +0100 Subject: [PATCH] Cleanup pypeline code Remove some cruft and apply trivial changes to the existing pypeline code, especially on the daemon side. --- python/revng/pypeline/cli/common_options.py | 4 +- python/revng/pypeline/cli/project/__init__.py | 4 +- python/revng/pypeline/cli/project/daemon.py | 8 ---- python/revng/pypeline/daemon/app.py | 12 ++--- python/revng/pypeline/daemon/daemon.py | 43 ++++++++--------- .../daemon/notification_broker/__init__.py | 8 ++-- .../notification_broker/local_broker.py | 12 ++--- python/revng/pypeline/daemon/utils.py | 2 +- python/revng/pypeline/pipeline.py | 7 +-- python/revng/pypeline/web_schema.yml | 3 +- tests/pypeline/python/daemon/base.py | 15 ++---- tests/pypeline/python/daemon/json_daemon.py | 48 ++++--------------- .../python/daemon/starlette_daemon.py | 46 ++++-------------- tests/pypeline/python/pipebox.py | 3 +- 14 files changed, 66 insertions(+), 149 deletions(-) diff --git a/python/revng/pypeline/cli/common_options.py b/python/revng/pypeline/cli/common_options.py index 7ba536d8c..c26150d05 100644 --- a/python/revng/pypeline/cli/common_options.py +++ b/python/revng/pypeline/cli/common_options.py @@ -16,7 +16,7 @@ from revng.pypeline.runner_context import RunnerContext project_id_option = click.option( "--project-id", type=str, - help=("Project id to use for the storage provider."), + help="Project id to use for the storage provider.", envvar="PYPELINE_PROJECT_ID", show_default=True, ) @@ -85,7 +85,7 @@ def container_format_options(func): default=ContainerFormat.YAML.value, show_default=True, callback=handle_format_option, - help=("Format to use for the output container, either on stdout or in the result path."), + help="Format to use for the output container, either on stdout or in the result path.", )(func) for member in ContainerFormat: func = click.option( diff --git a/python/revng/pypeline/cli/project/__init__.py b/python/revng/pypeline/cli/project/__init__.py index 36927e72b..83957964d 100644 --- a/python/revng/pypeline/cli/project/__init__.py +++ b/python/revng/pypeline/cli/project/__init__.py @@ -37,7 +37,7 @@ from .daemon import run_daemon "--storage-provider", "storage_provider", type=StorageProviderUrl(), - help=("The URL of the storage provider to use."), + help="The URL of the storage provider to use.", default="local://", envvar="PYPELINE_STORAGE_PROVIDER", show_default=True, @@ -46,7 +46,7 @@ from .daemon import run_daemon "--cache-dir", "cache_dir", type=click.Path(exists=False, file_okay=False, dir_okay=True, writable=True), - help=("The directory to use for caching."), + help="The directory to use for caching.", default=str(cache_directory()), show_default=True, ) diff --git a/python/revng/pypeline/cli/project/daemon.py b/python/revng/pypeline/cli/project/daemon.py index 27de5b593..1dea0bd3f 100644 --- a/python/revng/pypeline/cli/project/daemon.py +++ b/python/revng/pypeline/cli/project/daemon.py @@ -6,9 +6,7 @@ import os import click import uvicorn -import yaml -import revng from revng.pypeline.cli.context import ClickContext, pass_context from revng.pypeline.cli.utils import PypeCommand from revng.pypeline.daemon.app import make_starlette @@ -25,9 +23,6 @@ from revng.pypeline.daemon.daemon import Daemon def run_daemon(ctx: ClickContext, production, **kwargs): """Start the HTTP daemon.""" - with open(ctx.obj.pipeline_path, "r", encoding="utf-8") as f: - pipeline_yaml = yaml.safe_load(f.read()) - # Configure uvicorn logging log_config = uvicorn.config.LOGGING_CONFIG @@ -47,10 +42,7 @@ def run_daemon(ctx: ClickContext, production, **kwargs): kwargs.setdefault("host", "0.0.0.0") daemon = Daemon( - version=revng.__version__, - pipeline_yaml=pipeline_yaml, pipeline=ctx.obj.pipeline, - debug=not production, storage_provider_url=ctx.obj.storage_provider_url, cache_dir=ctx.obj.cache_dir, base_directory=ctx.obj.base_directory, diff --git a/python/revng/pypeline/daemon/app.py b/python/revng/pypeline/daemon/app.py index be648e909..fdc99b90a 100644 --- a/python/revng/pypeline/daemon/app.py +++ b/python/revng/pypeline/daemon/app.py @@ -6,7 +6,7 @@ import json import os import traceback from functools import wraps -from typing import Any +from typing import Any, Mapping from starlette.applications import Starlette from starlette.exceptions import HTTPException @@ -46,9 +46,9 @@ def basic_exception_handler(request: Request, exc: Exception) -> JSONResponse: ) -def get_project_id(headers: dict[str, str]) -> str | None: +def get_project_id(headers: Mapping[str, str]) -> str | None: """Extract project ID from headers, return none if missing""" - return headers.get("x-projectid") + return headers.get("x-project-id") async def invalidation_websocket(websocket: WebSocket): @@ -57,8 +57,7 @@ async def invalidation_websocket(websocket: WebSocket): subscriber = None try: - project_id = get_project_id(dict(websocket.headers)) - assert project_id is not None, "Project ID is required" + project_id = get_project_id(websocket.headers) subscriber = await notification_broker.subscribe(project_id, WebSocketStream(websocket)) await subscriber.listen_for_messages() except BasicHTTPException as e: @@ -78,7 +77,7 @@ def prepare_endpoint(func): @wraps(func) async def wrapper(request: Request) -> JSONResponse: - project_id = get_project_id(dict(request.headers)) + project_id = get_project_id(request.headers) # Prepare the data dictionary with the common attributes we extract # from the headers data = { @@ -88,7 +87,6 @@ def prepare_endpoint(func): response = await func(request, data) # Forward any websocket notification for notification in response.notifications: - assert project_id is not None, "Project ID is required" await notification_broker.notify(project_id, json.dumps(notification)) # Convert the daemon response to a JSON response return JSONResponse( diff --git a/python/revng/pypeline/daemon/daemon.py b/python/revng/pypeline/daemon/daemon.py index 325d0cb00..53af136e2 100644 --- a/python/revng/pypeline/daemon/daemon.py +++ b/python/revng/pypeline/daemon/daemon.py @@ -9,6 +9,7 @@ from typing import Any import jsonschema import yaml +import revng.pypeline from revng.pypeline.container import Container from revng.pypeline.model import Model, ReadOnlyModel from revng.pypeline.object import Kind @@ -29,9 +30,7 @@ class Response: code: int """The HTTP status code of the response.""" body: Any - """ - The response body of the request. - """ + """The response body of the request.""" headers: dict[str, str] = field(default_factory=dict) """The headers of the response.""" notifications: list[Any] = field(default_factory=list) @@ -49,17 +48,14 @@ class Response: return result -def get_web_pipeline(version: str, pipeline: Pipeline) -> dict[str, Any]: +def get_pipeline_description(pipeline: Pipeline) -> dict[str, Any]: """ Build and validate the web representation of the pipeline. """ - root = Path(__file__).resolve().parent.parent - with open(root / "web_schema.yml", "r", encoding="utf-8") as f: - schema = yaml.safe_load(f) # Build the web pipeline - web_pipeline = { - "version": version, + pipeline_description = { + "version": revng.pypeline.__version__, "kinds": get_singleton(Kind).type_dict(), # type: ignore "containers": [ container.type_dict() @@ -78,10 +74,13 @@ def get_web_pipeline(version: str, pipeline: Pipeline) -> dict[str, Any]: } # Ensure that it respects the agreed schema + root = Path(__file__).resolve().parent.parent + with open(root / "web_schema.yml", "r", encoding="utf-8") as f: + schema = yaml.safe_load(f) validator = jsonschema.Draft7Validator(schema) - validator.validate(web_pipeline) + validator.validate(pipeline_description) - return web_pipeline + return pipeline_description class Daemon: @@ -89,25 +88,19 @@ class Daemon: def __init__( self, - version: str, - pipeline_yaml: Any, - pipeline: Any, - debug: bool, + pipeline: Pipeline, storage_provider_url: str, cache_dir: str, base_directory: Path, ): - self.version = version - self.pipeline_yaml = pipeline_yaml self.pipeline = pipeline - self.debug = debug self.cache_dir = cache_dir self.base_directory = base_directory self.storage_provider_factory = storage_provider_factory_factory(storage_provider_url) - self.web_pipeline = get_web_pipeline(version, pipeline) + self.pipeline_description = get_pipeline_description(pipeline) def _get_storage_provider_context(self, request): - project_id = request["project_id"] + project_id = request.get("project_id") token = request.get("token") return self.storage_provider_factory.get( base_directory=self.base_directory, @@ -141,7 +134,7 @@ class Daemon: def get_pipeline(self) -> Response: return Response( code=200, - body=self.web_pipeline, + body=self.pipeline_description, ) async def artifact(self, request) -> Response: @@ -276,6 +269,8 @@ class Daemon: ) else: analysis_list = self.pipeline.analysis_lists[analysis] + if len(configuration) == 0: + configuration = ["" for _ in analysis_list.analyses] new_model, invalidated = self.pipeline.run_analysis_list( model=ReadOnlyModel(model), analysis_list=analysis_list, @@ -285,7 +280,7 @@ class Daemon: ) # Compute the diff between the original model and the final one - diff = str(model.diff(new_model)) + diff_raw = model.diff(new_model).serialize() # TODO: this can be done much more efficiently new_epoch = storage_provider.get_epoch() @@ -298,11 +293,13 @@ class Daemon: invalidated_artifacts.append( { "name": artifact.name, - "configuration": artifact.configuration, + "configuration": container_location.configuration_id, "object_ids": [object_id.serialize() for object_id in object_ids], } ) + model_type = get_singleton(Model) # type: ignore[type-abstract] + diff = bytes_to_string(diff_raw, model_type.is_text()) # Return the updated model return Response( code=200, diff --git a/python/revng/pypeline/daemon/notification_broker/__init__.py b/python/revng/pypeline/daemon/notification_broker/__init__.py index 0825a86ee..51c25d19f 100644 --- a/python/revng/pypeline/daemon/notification_broker/__init__.py +++ b/python/revng/pypeline/daemon/notification_broker/__init__.py @@ -34,7 +34,7 @@ class WebSocketStream(Stream): class NotificationSubscriber: """Represents a notification subscriber with its own message queue""" - def __init__(self, project_id: ProjectID, stream: Stream): + def __init__(self, project_id: ProjectID | None, stream: Stream): self.project_id = project_id self.stream = stream self.message_queue: asyncio.Queue[str] = asyncio.Queue() @@ -76,7 +76,9 @@ class NotificationBroker(ABC): """ @abstractmethod - async def subscribe(self, project_id: ProjectID, stream: Stream) -> NotificationSubscriber: + async def subscribe( + self, project_id: ProjectID | None, stream: Stream + ) -> NotificationSubscriber: """Register to receive notifications for project \"project_id\" on the given stream""" @abstractmethod @@ -84,5 +86,5 @@ class NotificationBroker(ABC): """Stop receiving notifications for project \"project_id\".""" @abstractmethod - async def notify(self, project_id: ProjectID, message: str): + async def notify(self, project_id: ProjectID | None, message: str): """Notify all subscribers of project \"project_id\" of the given message""" diff --git a/python/revng/pypeline/daemon/notification_broker/local_broker.py b/python/revng/pypeline/daemon/notification_broker/local_broker.py index 645e4f911..26a85af22 100644 --- a/python/revng/pypeline/daemon/notification_broker/local_broker.py +++ b/python/revng/pypeline/daemon/notification_broker/local_broker.py @@ -4,7 +4,6 @@ from __future__ import annotations -import asyncio from collections import defaultdict from revng.pypeline.storage.storage_provider import ProjectID @@ -16,12 +15,13 @@ from . import NotificationBroker, NotificationSubscriber, Stream class LocalNotificationBroker(NotificationBroker): def __init__(self): - self._lock = asyncio.Lock() - self.subscribers: Locked[dict[ProjectID, Locked[set[NotificationSubscriber]]]] = Locked( - defaultdict(lambda: Locked(set())) + self.subscribers: Locked[dict[ProjectID | None, Locked[set[NotificationSubscriber]]]] = ( + Locked(defaultdict(lambda: Locked(set()))) ) - async def subscribe(self, project_id: ProjectID, stream: Stream) -> NotificationSubscriber: + async def subscribe( + self, project_id: ProjectID | None, stream: Stream + ) -> NotificationSubscriber: subscriber = NotificationSubscriber(project_id, stream) async with self.subscribers() as subscribers: @@ -41,7 +41,7 @@ class LocalNotificationBroker(NotificationBroker): subscriber.close() pypeline_logger.debug_log(f"Stream unsubscribed from project {subscriber.project_id}") - async def notify(self, project_id: ProjectID, message: str): + async def notify(self, project_id: ProjectID | None, message: str): async with self.subscribers() as subscribers: if project_id not in subscribers: pypeline_logger.debug_log(f"No subscribers for project {project_id}") diff --git a/python/revng/pypeline/daemon/utils.py b/python/revng/pypeline/daemon/utils.py index f7105b983..908177c51 100644 --- a/python/revng/pypeline/daemon/utils.py +++ b/python/revng/pypeline/daemon/utils.py @@ -42,6 +42,7 @@ def compute_objects( # Otherwise we have to parse the objects provided by the user. # Parse the objects into ObjectSet + obj_type = get_singleton(ObjectID) # type: ignore [type-abstract] objset = set() for obj in objects: if not isinstance(obj, str): @@ -53,7 +54,6 @@ def compute_objects( ) # Deserialize the object ID - obj_type = get_singleton(ObjectID) # type: ignore [type-abstract] try: obj_id = obj_type.deserialize(obj) except ValueError as e: diff --git a/python/revng/pypeline/pipeline.py b/python/revng/pypeline/pipeline.py index cf97a9613..d678b7e18 100644 --- a/python/revng/pypeline/pipeline.py +++ b/python/revng/pypeline/pipeline.py @@ -6,7 +6,7 @@ from __future__ import annotations from dataclasses import dataclass from itertools import chain -from typing import Dict, Generator, Generic, Iterable, List, Mapping, Optional, Set, TypeVar +from typing import Dict, Generator, Iterable, List, Mapping, Optional, Set import yaml @@ -60,10 +60,7 @@ class Artifact: } -C = TypeVar("C", bound=Model) - - -class Pipeline(Generic[C]): +class Pipeline: """ A pipeline is a tree of tasks. diff --git a/python/revng/pypeline/web_schema.yml b/python/revng/pypeline/web_schema.yml index 3c10ff7de..e6d6205c6 100644 --- a/python/revng/pypeline/web_schema.yml +++ b/python/revng/pypeline/web_schema.yml @@ -35,7 +35,8 @@ properties: type: - string - "null" - description: "The name of the parent kind, or null if this is a root kind" + description: > + The name of the parent kind, or null if this is a root kind containers: type: array description: List of container types available in the pipeline diff --git a/tests/pypeline/python/daemon/base.py b/tests/pypeline/python/daemon/base.py index 18b0eab41..1de259666 100644 --- a/tests/pypeline/python/daemon/base.py +++ b/tests/pypeline/python/daemon/base.py @@ -15,9 +15,7 @@ class Response: code: int """The HTTP status code of the response.""" body: Any - """ - The response body of the request. - """ + """The response body of the request.""" class TestServer(ABC): @@ -26,15 +24,8 @@ class TestServer(ABC): test_dir = Path(__file__).parent.parent self.tmp_dir = tempfile.TemporaryDirectory(prefix="test_daemon") self.tmp_dir_path = Path(self.tmp_dir.name) - for file in [ - "pipebox.py", - "pipeline.yml", - "model.yml", - ]: - shutil.copyfile( - test_dir / file, - self.tmp_dir_path / file, - ) + for file in ["pipebox.py", "pipeline.yml", "model.yml"]: + shutil.copyfile(test_dir / file, self.tmp_dir_path / file) self.pipebox_path = str(self.tmp_dir_path / "pipebox.py") self.pipeline_path = str(self.tmp_dir_path / "pipeline.yml") diff --git a/tests/pypeline/python/daemon/json_daemon.py b/tests/pypeline/python/daemon/json_daemon.py index a0ebb12ca..c1a833623 100644 --- a/tests/pypeline/python/daemon/json_daemon.py +++ b/tests/pypeline/python/daemon/json_daemon.py @@ -8,7 +8,6 @@ import os import queue from tempfile import TemporaryDirectory -import revng from revng.pypeline.daemon.daemon import Daemon from revng.pypeline.pipeline_parser import load_pipeline_yaml @@ -41,14 +40,10 @@ class JsonTestServer(TestServer): self.project_id = "test_project_id_json" with open(self.pipeline_path, "r") as pipeline_file: - self.pipeline_yaml = pipeline_file.read() - self.pipeline = load_pipeline_yaml(self.pipeline_yaml) + self.pipeline = load_pipeline_yaml(pipeline_file.read()) self.daemon = Daemon( - version=revng.__version__, - pipeline_yaml=self.pipeline_yaml, pipeline=self.pipeline, - debug=True, storage_provider_url=storage_provider_url, cache_dir=self.cache_dir_tmp.name, base_directory=self.tmp_dir_path, @@ -57,55 +52,28 @@ class JsonTestServer(TestServer): self.websocket = WebsocketMock() def get_epoch(self) -> Response: - response = asyncio.run( - self.daemon.get_epoch( - { - "project_id": self.project_id, - } - ) - ) - return Response( - code=response.code, - body=response.body, - ) + response = asyncio.run(self.daemon.get_epoch({"project_id": self.project_id})) + return Response(code=response.code, body=response.body) def get_pipeline(self) -> Response: response = self.daemon.get_pipeline() - return Response( - code=response.code, - body=response.body, - ) + return Response(code=response.code, body=response.body) def get_model(self) -> Response: - response = asyncio.run( - self.daemon.get_model( - { - "project_id": self.project_id, - } - ) - ) - return Response( - code=response.code, - body=response.body, - ) + response = asyncio.run(self.daemon.get_model({"project_id": self.project_id})) + return Response(code=response.code, body=response.body) def run_analysis(self, analysis_request) -> Response: analysis_request.setdefault("project_id", self.project_id) response = asyncio.run(self.daemon.analyze(analysis_request)) for notification in response.notifications: self.websocket.send(json.dumps(notification)) - return Response( - code=response.code, - body=response.body, - ) + return Response(code=response.code, body=response.body) def get_artifact(self, artifact_request) -> Response: artifact_request.setdefault("project_id", self.project_id) response = asyncio.run(self.daemon.artifact(artifact_request)) - return Response( - code=response.code, - body=response.body, - ) + return Response(code=response.code, body=response.body) def subscribe(self): return self.websocket diff --git a/tests/pypeline/python/daemon/starlette_daemon.py b/tests/pypeline/python/daemon/starlette_daemon.py index a4672e318..3e0a2d739 100644 --- a/tests/pypeline/python/daemon/starlette_daemon.py +++ b/tests/pypeline/python/daemon/starlette_daemon.py @@ -63,7 +63,7 @@ class StarletteTestServer(TestServer): main( ( "-C", - self.tmp_dir_path, + self.tmp_dir.name, "--pipebox", self.pipebox_path, "project", @@ -98,61 +98,31 @@ class StarletteTestServer(TestServer): logger.info("Getting epoch") r = self.session.get(f"{self.base_url}/api/epoch") logger.info("Epoch response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) + return Response(code=r.status_code, body=r.json()) def get_pipeline(self) -> Response: logger.info("Getting pipeline") r = self.session.get(f"{self.base_url}/api/pipeline") logger.info("Pipeline response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) + return Response(code=r.status_code, body=r.json()) def get_model(self) -> Response: logger.info("Getting model") r = self.session.get(f"{self.base_url}/api/model") logger.info("Model response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) - - def get_monitoring(self) -> Response: - logger.info("Getting model") - r = self.session.get(f"{self.base_url}/api/monitoring") - logger.info("Monitoring response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) + return Response(code=r.status_code, body=r.json()) def run_analysis(self, analysis_request) -> Response: logger.info("Running analysis with request %s", analysis_request) - r = self.session.post( - f"{self.base_url}/api/analysis", - json=analysis_request, - ) + r = self.session.post(f"{self.base_url}/api/analysis", json=analysis_request) logger.info("Analysis response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) + return Response(code=r.status_code, body=r.json()) def get_artifact(self, artifact_request) -> Response: logger.info("Getting Artifact with request %s", artifact_request) - r = self.session.post( - f"{self.base_url}/api/artifact", - json=artifact_request, - ) + r = self.session.post(f"{self.base_url}/api/artifact", json=artifact_request) logger.info("Artifact response: %s", r.text) - return Response( - code=r.status_code, - body=r.json(), - ) + return Response(code=r.status_code, body=r.json()) def subscribe(self): return websockets.sync.client.connect( diff --git a/tests/pypeline/python/pipebox.py b/tests/pypeline/python/pipebox.py index 10c46efea..b749f4e54 100644 --- a/tests/pypeline/python/pipebox.py +++ b/tests/pypeline/python/pipebox.py @@ -5,6 +5,7 @@ from __future__ import annotations +import json import sys from abc import ABC, ABCMeta from collections.abc import Buffer @@ -246,7 +247,7 @@ class DictModelDiff(ModelDiff): return self._paths def serialize(self) -> bytes: - raise NotImplementedError() + return json.dumps(list(self._paths)).encode() class DictModel(Model):