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

@@ -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(

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,
 )

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,

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(

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,

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"""

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}")

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:

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.
 

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